@Test void searchingInALargeNumberOfMailboxesShouldReturnAllMailboxesMessagesUid() throws Exception { MailboxSession session = MailboxSessionUtil.create(USERNAME); int numberOfMailboxes = 700; List<MailboxPath> mailboxPaths = IntStream .range(0, numberOfMailboxes) .mapToObj(index -> MailboxPath.forUser(USERNAME, "mailbox" + index)) .collect(Guavate.toImmutableList()); List<MailboxId> mailboxIds = mailboxPaths.stream() .map(Throwing.<MailboxPath, MailboxId>function(mailboxPath -> storeMailboxManager.createMailbox(mailboxPath, session).get()).sneakyThrow()) .collect(Guavate.toImmutableList()); List<ComposedMessageId> composedMessageIds = mailboxPaths.stream() .map(Throwing.<MailboxPath, ComposedMessageId>function(mailboxPath -> addMessage(session, mailboxPath)).sneakyThrow()) .collect(Guavate.toImmutableList()); elasticSearch.awaitForElasticSearch(); MultimailboxesSearchQuery multimailboxesSearchQuery = MultimailboxesSearchQuery .from(SearchQuery.of(SearchQuery.all())) .inMailboxes(mailboxIds) .build(); List<MessageId> expectedMessageIds = composedMessageIds .stream() .map(ComposedMessageId::getMessageId) .collect(Guavate.toImmutableList()); assertThat(storeMailboxManager.search(multimailboxesSearchQuery, session, numberOfMailboxes + 1) .collectList().block()) .containsExactlyInAnyOrderElementsOf(expectedMessageIds); }
public Flux<MessageSearchIndex.SearchResult> search(Collection<MailboxId> mailboxIds, SearchQuery query, Optional<Integer> limit) { SearchRequest searchRequest = prepareSearch(mailboxIds, query, limit); Flux<MessageSearchIndex.SearchResult> pairStream = new ScrolledSearch(client, searchRequest) .searchHits() .flatMap(this::extractContentFromHit); return limit.map(pairStream::take) .orElse(pairStream); }
ElasticSearchSearcher { public Flux<MessageSearchIndex.SearchResult> search(Collection<MailboxId> mailboxIds, SearchQuery query, Optional<Integer> limit) { SearchRequest searchRequest = prepareSearch(mailboxIds, query, limit); Flux<MessageSearchIndex.SearchResult> pairStream = new ScrolledSearch(client, searchRequest) .searchHits() .flatMap(this::extractContentFromHit); return limit.map(pairStream::take) .orElse(pairStream); } }
ElasticSearchSearcher { public Flux<MessageSearchIndex.SearchResult> search(Collection<MailboxId> mailboxIds, SearchQuery query, Optional<Integer> limit) { SearchRequest searchRequest = prepareSearch(mailboxIds, query, limit); Flux<MessageSearchIndex.SearchResult> pairStream = new ScrolledSearch(client, searchRequest) .searchHits() .flatMap(this::extractContentFromHit); return limit.map(pairStream::take) .orElse(pairStream); } ElasticSearchSearcher(ReactorElasticSearchClient client, QueryConverter queryConverter, int size,
                                 MailboxId.Factory mailboxIdFactory, MessageId.Factory messageIdFactory,
                                 ReadAliasName aliasName, RoutingKey.Factory<MailboxId> routingKeyFactory); }
ElasticSearchSearcher { public Flux<MessageSearchIndex.SearchResult> search(Collection<MailboxId> mailboxIds, SearchQuery query, Optional<Integer> limit) { SearchRequest searchRequest = prepareSearch(mailboxIds, query, limit); Flux<MessageSearchIndex.SearchResult> pairStream = new ScrolledSearch(client, searchRequest) .searchHits() .flatMap(this::extractContentFromHit); return limit.map(pairStream::take) .orElse(pairStream); } ElasticSearchSearcher(ReactorElasticSearchClient client, QueryConverter queryConverter, int size,
                                 MailboxId.Factory mailboxIdFactory, MessageId.Factory messageIdFactory,
                                 ReadAliasName aliasName, RoutingKey.Factory<MailboxId> routingKeyFactory); Flux<MessageSearchIndex.SearchResult> search(Collection<MailboxId> mailboxIds, SearchQuery query,
                                                        Optional<Integer> limit); }
ElasticSearchSearcher { public Flux<MessageSearchIndex.SearchResult> search(Collection<MailboxId> mailboxIds, SearchQuery query, Optional<Integer> limit) { SearchRequest searchRequest = prepareSearch(mailboxIds, query, limit); Flux<MessageSearchIndex.SearchResult> pairStream = new ScrolledSearch(client, searchRequest) .searchHits() .flatMap(this::extractContentFromHit); return limit.map(pairStream::take) .orElse(pairStream); } ElasticSearchSearcher(ReactorElasticSearchClient client, QueryConverter queryConverter, int size,
                                 MailboxId.Factory mailboxIdFactory, MessageId.Factory messageIdFactory,
                                 ReadAliasName aliasName, RoutingKey.Factory<MailboxId> routingKeyFactory); Flux<MessageSearchIndex.SearchResult> search(Collection<MailboxId> mailboxIds, SearchQuery query,
                                                        Optional<Integer> limit); static final int DEFAULT_SEARCH_SIZE; }