@Test void updateShouldPropagateExceptionWhenExceptionOccurs() throws Exception { elasticSearch.getDockerElasticSearch().pause(); Thread.sleep(Duration.FIVE_SECONDS.getValueInMS()); Flags newFlags = new Flags(Flags.Flag.ANSWERED); UpdatedFlags updatedFlags = UpdatedFlags.builder() .uid(MESSAGE_UID_1) .modSeq(MOD_SEQ) .oldFlags(new Flags()) .newFlags(newFlags) .build(); assertThatThrownBy(() -> testee.update(session, mailbox.getMailboxId(), Lists.newArrayList(updatedFlags)).block()) .hasCauseInstanceOf(IOException.class); elasticSearch.getDockerElasticSearch().unpause(); }
@Override public Mono<Void> update(MailboxSession session, MailboxId mailboxId, List<UpdatedFlags> updatedFlagsList) { RoutingKey routingKey = routingKeyFactory.from(mailboxId); return Flux.fromIterable(updatedFlagsList) .map(Throwing.<UpdatedFlags, UpdatedRepresentation>function( updatedFlags -> createUpdatedDocumentPartFromUpdatedFlags(mailboxId, updatedFlags)) .sneakyThrow()) .window(FLAGS_UPDATE_PROCESSING_WINDOW_SIZE) .concatMap(flux -> flux.collect(toImmutableList()) .flatMap(updates -> elasticSearchIndexer.update(updates, routingKey))) .then(); }
ElasticSearchListeningMessageSearchIndex extends ListeningMessageSearchIndex { @Override public Mono<Void> update(MailboxSession session, MailboxId mailboxId, List<UpdatedFlags> updatedFlagsList) { RoutingKey routingKey = routingKeyFactory.from(mailboxId); return Flux.fromIterable(updatedFlagsList) .map(Throwing.<UpdatedFlags, UpdatedRepresentation>function( updatedFlags -> createUpdatedDocumentPartFromUpdatedFlags(mailboxId, updatedFlags)) .sneakyThrow()) .window(FLAGS_UPDATE_PROCESSING_WINDOW_SIZE) .concatMap(flux -> flux.collect(toImmutableList()) .flatMap(updates -> elasticSearchIndexer.update(updates, routingKey))) .then(); } }
ElasticSearchListeningMessageSearchIndex extends ListeningMessageSearchIndex { @Override public Mono<Void> update(MailboxSession session, MailboxId mailboxId, List<UpdatedFlags> updatedFlagsList) { RoutingKey routingKey = routingKeyFactory.from(mailboxId); return Flux.fromIterable(updatedFlagsList) .map(Throwing.<UpdatedFlags, UpdatedRepresentation>function( updatedFlags -> createUpdatedDocumentPartFromUpdatedFlags(mailboxId, updatedFlags)) .sneakyThrow()) .window(FLAGS_UPDATE_PROCESSING_WINDOW_SIZE) .concatMap(flux -> flux.collect(toImmutableList()) .flatMap(updates -> elasticSearchIndexer.update(updates, routingKey))) .then(); } @Inject  ElasticSearchListeningMessageSearchIndex(MailboxSessionMapperFactory factory,
                                                    @Named(MailboxElasticSearchConstants.InjectionNames.MAILBOX) ElasticSearchIndexer indexer,
                                                    ElasticSearchSearcher searcher, MessageToElasticSearchJson messageToElasticSearchJson,
                                                    SessionProvider sessionProvider, RoutingKey.Factory<MailboxId> routingKeyFactory); }
ElasticSearchListeningMessageSearchIndex extends ListeningMessageSearchIndex { @Override public Mono<Void> update(MailboxSession session, MailboxId mailboxId, List<UpdatedFlags> updatedFlagsList) { RoutingKey routingKey = routingKeyFactory.from(mailboxId); return Flux.fromIterable(updatedFlagsList) .map(Throwing.<UpdatedFlags, UpdatedRepresentation>function( updatedFlags -> createUpdatedDocumentPartFromUpdatedFlags(mailboxId, updatedFlags)) .sneakyThrow()) .window(FLAGS_UPDATE_PROCESSING_WINDOW_SIZE) .concatMap(flux -> flux.collect(toImmutableList()) .flatMap(updates -> elasticSearchIndexer.update(updates, routingKey))) .then(); } @Inject  ElasticSearchListeningMessageSearchIndex(MailboxSessionMapperFactory factory,
                                                    @Named(MailboxElasticSearchConstants.InjectionNames.MAILBOX) ElasticSearchIndexer indexer,
                                                    ElasticSearchSearcher searcher, MessageToElasticSearchJson messageToElasticSearchJson,
                                                    SessionProvider sessionProvider, RoutingKey.Factory<MailboxId> routingKeyFactory); @Override Group getDefaultGroup(); @Override EnumSet<SearchCapabilities> getSupportedCapabilities(EnumSet<MessageCapabilities> messageCapabilities); @Override Stream<MessageUid> search(MailboxSession session, Mailbox mailbox, SearchQuery searchQuery); @Override Flux<MessageId> search(MailboxSession session, Collection<MailboxId> mailboxIds, SearchQuery searchQuery, long limit); @Override Mono<Void> add(MailboxSession session, Mailbox mailbox, MailboxMessage message); @Override Mono<Void> delete(MailboxSession session, MailboxId mailboxId, Collection<MessageUid> expungedUids); @Override Mono<Void> deleteAll(MailboxSession session, MailboxId mailboxId); @Override Mono<Void> update(MailboxSession session, MailboxId mailboxId, List<UpdatedFlags> updatedFlagsList); @Override Mono<Flags> retrieveIndexedFlags(Mailbox mailbox, MessageUid uid); }
ElasticSearchListeningMessageSearchIndex extends ListeningMessageSearchIndex { @Override public Mono<Void> update(MailboxSession session, MailboxId mailboxId, List<UpdatedFlags> updatedFlagsList) { RoutingKey routingKey = routingKeyFactory.from(mailboxId); return Flux.fromIterable(updatedFlagsList) .map(Throwing.<UpdatedFlags, UpdatedRepresentation>function( updatedFlags -> createUpdatedDocumentPartFromUpdatedFlags(mailboxId, updatedFlags)) .sneakyThrow()) .window(FLAGS_UPDATE_PROCESSING_WINDOW_SIZE) .concatMap(flux -> flux.collect(toImmutableList()) .flatMap(updates -> elasticSearchIndexer.update(updates, routingKey))) .then(); } @Inject  ElasticSearchListeningMessageSearchIndex(MailboxSessionMapperFactory factory,
                                                    @Named(MailboxElasticSearchConstants.InjectionNames.MAILBOX) ElasticSearchIndexer indexer,
                                                    ElasticSearchSearcher searcher, MessageToElasticSearchJson messageToElasticSearchJson,
                                                    SessionProvider sessionProvider, RoutingKey.Factory<MailboxId> routingKeyFactory); @Override Group getDefaultGroup(); @Override EnumSet<SearchCapabilities> getSupportedCapabilities(EnumSet<MessageCapabilities> messageCapabilities); @Override Stream<MessageUid> search(MailboxSession session, Mailbox mailbox, SearchQuery searchQuery); @Override Flux<MessageId> search(MailboxSession session, Collection<MailboxId> mailboxIds, SearchQuery searchQuery, long limit); @Override Mono<Void> add(MailboxSession session, Mailbox mailbox, MailboxMessage message); @Override Mono<Void> delete(MailboxSession session, MailboxId mailboxId, Collection<MessageUid> expungedUids); @Override Mono<Void> deleteAll(MailboxSession session, MailboxId mailboxId); @Override Mono<Void> update(MailboxSession session, MailboxId mailboxId, List<UpdatedFlags> updatedFlagsList); @Override Mono<Flags> retrieveIndexedFlags(Mailbox mailbox, MessageUid uid);  }