@Test void addShouldPropagateExceptionWhenExceptionOccurs() throws Exception { elasticSearch.getDockerElasticSearch().pause(); Thread.sleep(Duration.FIVE_SECONDS.getValueInMS()); assertThatThrownBy(() -> testee.add(session, mailbox, MESSAGE_1).block()) .hasCauseInstanceOf(IOException.class); elasticSearch.getDockerElasticSearch().unpause(); }
@Override public Mono<Void> add(MailboxSession session, Mailbox mailbox, MailboxMessage message) { LOGGER.info("Indexing mailbox {}-{} of user {} on message {}", mailbox.getName(), mailbox.getMailboxId(), session.getUser().asString(), message.getUid()); RoutingKey from = routingKeyFactory.from(mailbox.getMailboxId()); DocumentId id = indexIdFor(mailbox.getMailboxId(), message.getUid()); return Mono.fromCallable(() -> generateIndexedJson(mailbox, message, session)) .flatMap(jsonContent -> elasticSearchIndexer.index(id, jsonContent, from)) .then(); }
ElasticSearchListeningMessageSearchIndex extends ListeningMessageSearchIndex { @Override public Mono<Void> add(MailboxSession session, Mailbox mailbox, MailboxMessage message) { LOGGER.info("Indexing mailbox {}-{} of user {} on message {}", mailbox.getName(), mailbox.getMailboxId(), session.getUser().asString(), message.getUid()); RoutingKey from = routingKeyFactory.from(mailbox.getMailboxId()); DocumentId id = indexIdFor(mailbox.getMailboxId(), message.getUid()); return Mono.fromCallable(() -> generateIndexedJson(mailbox, message, session)) .flatMap(jsonContent -> elasticSearchIndexer.index(id, jsonContent, from)) .then(); } }
ElasticSearchListeningMessageSearchIndex extends ListeningMessageSearchIndex { @Override public Mono<Void> add(MailboxSession session, Mailbox mailbox, MailboxMessage message) { LOGGER.info("Indexing mailbox {}-{} of user {} on message {}", mailbox.getName(), mailbox.getMailboxId(), session.getUser().asString(), message.getUid()); RoutingKey from = routingKeyFactory.from(mailbox.getMailboxId()); DocumentId id = indexIdFor(mailbox.getMailboxId(), message.getUid()); return Mono.fromCallable(() -> generateIndexedJson(mailbox, message, session)) .flatMap(jsonContent -> elasticSearchIndexer.index(id, jsonContent, from)) .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> add(MailboxSession session, Mailbox mailbox, MailboxMessage message) { LOGGER.info("Indexing mailbox {}-{} of user {} on message {}", mailbox.getName(), mailbox.getMailboxId(), session.getUser().asString(), message.getUid()); RoutingKey from = routingKeyFactory.from(mailbox.getMailboxId()); DocumentId id = indexIdFor(mailbox.getMailboxId(), message.getUid()); return Mono.fromCallable(() -> generateIndexedJson(mailbox, message, session)) .flatMap(jsonContent -> elasticSearchIndexer.index(id, jsonContent, from)) .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> add(MailboxSession session, Mailbox mailbox, MailboxMessage message) { LOGGER.info("Indexing mailbox {}-{} of user {} on message {}", mailbox.getName(), mailbox.getMailboxId(), session.getUser().asString(), message.getUid()); RoutingKey from = routingKeyFactory.from(mailbox.getMailboxId()); DocumentId id = indexIdFor(mailbox.getMailboxId(), message.getUid()); return Mono.fromCallable(() -> generateIndexedJson(mailbox, message, session)) .flatMap(jsonContent -> elasticSearchIndexer.index(id, jsonContent, from)) .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);  }