@Test public void bulkIndexPayload() { TestDoc testDoc1 = new TestDoc("id1", "Accepted", Instant.now().getEpochSecond()); TestDoc testDoc2 = new TestDoc("id2", "Running", Instant.now().getEpochSecond()); TestDoc testDoc3 = new TestDoc("id3", "Stopped", Instant.now().getEpochSecond()); List<TestDoc> testDocs = Arrays.asList(testDoc1, testDoc2, testDoc3); DefaultEsWebClientFactory defaultEsWebClientFactory = new DefaultEsWebClientFactory(getClientConfiguration()); DefaultEsClient<TestDoc> esClient = new DefaultEsClient<>(defaultEsWebClientFactory); final String bulkIndexPayload = esClient.buildBulkIndexPayload(testDocs, "titustasks", "default"); assertThat(bulkIndexPayload).isNotNull(); assertThat(bulkIndexPayload).isNotEmpty(); final String[] payloadLines = bulkIndexPayload.split("\n"); assertThat(payloadLines.length).isEqualTo(testDocs.size() * 2); String line1 = "{\"index\":{\"_index\":\"titustasks\",\"_type\":\"default\",\"_id\":\"id1\"}}"; String line2 = "{\"index\":{\"_index\":\"titustasks\",\"_type\":\"default\",\"_id\":\"id2\"}}"; String line3 = "{\"index\":{\"_index\":\"titustasks\",\"_type\":\"default\",\"_id\":\"id3\"}}"; assertThat(payloadLines[0]).isEqualTo(line1); assertThat(payloadLines[2]).isEqualTo(line2); assertThat(payloadLines[4]).isEqualTo(line3); }
@VisibleForTesting String buildBulkIndexPayload(List<T> tasks, String index, String type) { StringBuilder sb = new StringBuilder(); final ObjectMapper mapper = CommonObjectMappers.jacksonDefaultMapper(); tasks.forEach(taskDocument -> { final IndexHeader indexHeader = new IndexHeader(index, type, taskDocument.getId()); final IndexHeaderLine indexHeaderLine = new IndexHeaderLine(indexHeader); try { final String indexLine = mapper.writeValueAsString(indexHeaderLine); sb.append(indexLine); sb.append("\n"); final String fieldsLine = mapper.writeValueAsString(taskDocument); sb.append(fieldsLine); sb.append("\n"); } catch (JsonProcessingException e) { logger.error("Exception in transforming taskDocument into JSON ", e); } }); return sb.toString(); }
DefaultEsClient implements EsClient<T> { @VisibleForTesting String buildBulkIndexPayload(List<T> tasks, String index, String type) { StringBuilder sb = new StringBuilder(); final ObjectMapper mapper = CommonObjectMappers.jacksonDefaultMapper(); tasks.forEach(taskDocument -> { final IndexHeader indexHeader = new IndexHeader(index, type, taskDocument.getId()); final IndexHeaderLine indexHeaderLine = new IndexHeaderLine(indexHeader); try { final String indexLine = mapper.writeValueAsString(indexHeaderLine); sb.append(indexLine); sb.append("\n"); final String fieldsLine = mapper.writeValueAsString(taskDocument); sb.append(fieldsLine); sb.append("\n"); } catch (JsonProcessingException e) { logger.error("Exception in transforming taskDocument into JSON ", e); } }); return sb.toString(); } }
DefaultEsClient implements EsClient<T> { @VisibleForTesting String buildBulkIndexPayload(List<T> tasks, String index, String type) { StringBuilder sb = new StringBuilder(); final ObjectMapper mapper = CommonObjectMappers.jacksonDefaultMapper(); tasks.forEach(taskDocument -> { final IndexHeader indexHeader = new IndexHeader(index, type, taskDocument.getId()); final IndexHeaderLine indexHeaderLine = new IndexHeaderLine(indexHeader); try { final String indexLine = mapper.writeValueAsString(indexHeaderLine); sb.append(indexLine); sb.append("\n"); final String fieldsLine = mapper.writeValueAsString(taskDocument); sb.append(fieldsLine); sb.append("\n"); } catch (JsonProcessingException e) { logger.error("Exception in transforming taskDocument into JSON ", e); } }); return sb.toString(); } DefaultEsClient(EsWebClientFactory esWebClientFactory); }
DefaultEsClient implements EsClient<T> { @VisibleForTesting String buildBulkIndexPayload(List<T> tasks, String index, String type) { StringBuilder sb = new StringBuilder(); final ObjectMapper mapper = CommonObjectMappers.jacksonDefaultMapper(); tasks.forEach(taskDocument -> { final IndexHeader indexHeader = new IndexHeader(index, type, taskDocument.getId()); final IndexHeaderLine indexHeaderLine = new IndexHeaderLine(indexHeader); try { final String indexLine = mapper.writeValueAsString(indexHeaderLine); sb.append(indexLine); sb.append("\n"); final String fieldsLine = mapper.writeValueAsString(taskDocument); sb.append(fieldsLine); sb.append("\n"); } catch (JsonProcessingException e) { logger.error("Exception in transforming taskDocument into JSON ", e); } }); return sb.toString(); } DefaultEsClient(EsWebClientFactory esWebClientFactory); @Override Mono<EsIndexResp> indexDocument(T document, String indexName, String documentType); @Override Mono<BulkEsIndexResp> bulkIndexDocuments(List<T> taskDocuments, String index, String type); @Override Mono<EsRespSrc<T>> findDocumentById(String id, String index, String type,
                                               ParameterizedTypeReference<EsRespSrc<T>> responseTypeRef); @Override Mono<EsRespCount> getTotalDocumentCount(String index, String type); }
DefaultEsClient implements EsClient<T> { @VisibleForTesting String buildBulkIndexPayload(List<T> tasks, String index, String type) { StringBuilder sb = new StringBuilder(); final ObjectMapper mapper = CommonObjectMappers.jacksonDefaultMapper(); tasks.forEach(taskDocument -> { final IndexHeader indexHeader = new IndexHeader(index, type, taskDocument.getId()); final IndexHeaderLine indexHeaderLine = new IndexHeaderLine(indexHeader); try { final String indexLine = mapper.writeValueAsString(indexHeaderLine); sb.append(indexLine); sb.append("\n"); final String fieldsLine = mapper.writeValueAsString(taskDocument); sb.append(fieldsLine); sb.append("\n"); } catch (JsonProcessingException e) { logger.error("Exception in transforming taskDocument into JSON ", e); } }); return sb.toString(); } DefaultEsClient(EsWebClientFactory esWebClientFactory); @Override Mono<EsIndexResp> indexDocument(T document, String indexName, String documentType); @Override Mono<BulkEsIndexResp> bulkIndexDocuments(List<T> taskDocuments, String index, String type); @Override Mono<EsRespSrc<T>> findDocumentById(String id, String index, String type,
                                               ParameterizedTypeReference<EsRespSrc<T>> responseTypeRef); @Override Mono<EsRespCount> getTotalDocumentCount(String index, String type);  }