ElasticSearchIndexer { public Mono<IndexResponse> index(DocumentId id, String content, RoutingKey routingKey) { checkArgument(content); logContent(id, content); return client.index(new IndexRequest(aliasName.getValue()) .type(NodeMappingFactory.DEFAULT_MAPPING_NAME) .id(id.asString()) .source(content, XContentType.JSON) .routing(routingKey.asString()), RequestOptions.DEFAULT); } ElasticSearchIndexer(ReactorElasticSearchClient client,
                                WriteAliasName aliasName); @VisibleForTesting  ElasticSearchIndexer(ReactorElasticSearchClient client,
                                WriteAliasName aliasName,
                                int batchSize); Mono<IndexResponse> index(DocumentId id, String content, RoutingKey routingKey); Mono<BulkResponse> update(List<UpdatedRepresentation> updatedDocumentParts, RoutingKey routingKey); Mono<BulkResponse> delete(List<DocumentId> ids, RoutingKey routingKey); Mono<Void> deleteAllMatchingQuery(QueryBuilder queryBuilder, RoutingKey routingKey); Mono<GetResponse> get(DocumentId id, RoutingKey routingKey);  }
@Test void indexMessageShouldWork() { DocumentId documentId = DocumentId.fromString("1"); String content = "{\"message\": \"trying out Elasticsearch\"}"; testee.index(documentId, content, useDocumentId(documentId)).block(); elasticSearch.awaitForElasticSearch(); SearchResponse searchResponse = client.search( new SearchRequest(INDEX_NAME.getValue()) .source(new SearchSourceBuilder().query(QueryBuilders.matchQuery("message", "trying"))), RequestOptions.DEFAULT) .block(); assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1); }
@Test void indexMessageShouldThrowWhenJsonIsNull() { assertThatThrownBy(() -> testee.index(DOCUMENT_ID, null, ROUTING)) .isInstanceOf(IllegalArgumentException.class); }