@Test void updateMessageShouldThrowWhenJsonIsNull() { assertThatThrownBy(() -> testee.update(ImmutableList.of( new UpdatedRepresentation(DOCUMENT_ID, null)), ROUTING).block()) .isInstanceOf(IllegalArgumentException.class); }
public Mono<BulkResponse> update(List<UpdatedRepresentation> updatedDocumentParts, RoutingKey routingKey) { Preconditions.checkNotNull(updatedDocumentParts); Preconditions.checkNotNull(routingKey); BulkRequest request = new BulkRequest(); updatedDocumentParts.forEach(updatedDocumentPart -> request.add( new UpdateRequest(aliasName.getValue(), NodeMappingFactory.DEFAULT_MAPPING_NAME, updatedDocumentPart.getId().asString()) .doc(updatedDocumentPart.getUpdatedDocumentPart(), XContentType.JSON) .routing(routingKey.asString()))); return client.bulk(request, RequestOptions.DEFAULT) .onErrorResume(ValidationException.class, exception -> { LOGGER.warn("Error while updating index", exception); return Mono.empty(); }); }
ElasticSearchIndexer { public Mono<BulkResponse> update(List<UpdatedRepresentation> updatedDocumentParts, RoutingKey routingKey) { Preconditions.checkNotNull(updatedDocumentParts); Preconditions.checkNotNull(routingKey); BulkRequest request = new BulkRequest(); updatedDocumentParts.forEach(updatedDocumentPart -> request.add( new UpdateRequest(aliasName.getValue(), NodeMappingFactory.DEFAULT_MAPPING_NAME, updatedDocumentPart.getId().asString()) .doc(updatedDocumentPart.getUpdatedDocumentPart(), XContentType.JSON) .routing(routingKey.asString()))); return client.bulk(request, RequestOptions.DEFAULT) .onErrorResume(ValidationException.class, exception -> { LOGGER.warn("Error while updating index", exception); return Mono.empty(); }); } }
ElasticSearchIndexer { public Mono<BulkResponse> update(List<UpdatedRepresentation> updatedDocumentParts, RoutingKey routingKey) { Preconditions.checkNotNull(updatedDocumentParts); Preconditions.checkNotNull(routingKey); BulkRequest request = new BulkRequest(); updatedDocumentParts.forEach(updatedDocumentPart -> request.add( new UpdateRequest(aliasName.getValue(), NodeMappingFactory.DEFAULT_MAPPING_NAME, updatedDocumentPart.getId().asString()) .doc(updatedDocumentPart.getUpdatedDocumentPart(), XContentType.JSON) .routing(routingKey.asString()))); return client.bulk(request, RequestOptions.DEFAULT) .onErrorResume(ValidationException.class, exception -> { LOGGER.warn("Error while updating index", exception); return Mono.empty(); }); } ElasticSearchIndexer(ReactorElasticSearchClient client,
                                WriteAliasName aliasName); @VisibleForTesting  ElasticSearchIndexer(ReactorElasticSearchClient client,
                                WriteAliasName aliasName,
                                int batchSize); }
ElasticSearchIndexer { public Mono<BulkResponse> update(List<UpdatedRepresentation> updatedDocumentParts, RoutingKey routingKey) { Preconditions.checkNotNull(updatedDocumentParts); Preconditions.checkNotNull(routingKey); BulkRequest request = new BulkRequest(); updatedDocumentParts.forEach(updatedDocumentPart -> request.add( new UpdateRequest(aliasName.getValue(), NodeMappingFactory.DEFAULT_MAPPING_NAME, updatedDocumentPart.getId().asString()) .doc(updatedDocumentPart.getUpdatedDocumentPart(), XContentType.JSON) .routing(routingKey.asString()))); return client.bulk(request, RequestOptions.DEFAULT) .onErrorResume(ValidationException.class, exception -> { LOGGER.warn("Error while updating index", exception); return Mono.empty(); }); } 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); }
ElasticSearchIndexer { public Mono<BulkResponse> update(List<UpdatedRepresentation> updatedDocumentParts, RoutingKey routingKey) { Preconditions.checkNotNull(updatedDocumentParts); Preconditions.checkNotNull(routingKey); BulkRequest request = new BulkRequest(); updatedDocumentParts.forEach(updatedDocumentPart -> request.add( new UpdateRequest(aliasName.getValue(), NodeMappingFactory.DEFAULT_MAPPING_NAME, updatedDocumentPart.getId().asString()) .doc(updatedDocumentPart.getUpdatedDocumentPart(), XContentType.JSON) .routing(routingKey.asString()))); return client.bulk(request, RequestOptions.DEFAULT) .onErrorResume(ValidationException.class, exception -> { LOGGER.warn("Error while updating index", exception); return Mono.empty(); }); } 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);  }