added documentCreator to WriteDOcumentsToElasticsearchIndexChain

Author: Manuel Seiche

src/main/java/io/github/cupybara/javalangchains/chains/data/writer/WriteDocumentsToElasticsearchIndexChain.java

@@ -37,11 +37,6 @@ public class WriteDocumentsToElasticsearchIndexChain implements Chain<Stream<Map
 	 */
 	private final RestClientBuilder restClientBuilder;
 
-	/**
-	 * {@link ObjectMapper} used for document serialization
-	 */
-	private final ObjectMapper objectMapper;
-
 	/**
 	 * Optional {@link Function} which provides an ID value for a document. If set,
 	 * documents are indexed using PUT /_doc/${id} instead of POST /_doc
@@ -54,6 +49,28 @@ public class WriteDocumentsToElasticsearchIndexChain implements Chain<Stream<Map
 	 */
 	private final BiConsumer<String, RestClient> indexCreator;
 
+	/**
+	 * creates the effective document json from an input document
+	 */
+	private final Function<Map<String, String>, String> documentJsonCreator;
+
+	/**
+	 * @param index               {@link #index}
+	 * @param restClientBuilder   {@link #restClientBuilder}
+	 * @param idProvider          {@link #idProvider}
+	 * @param indexCreator        {@link #indexCreator}
+	 * @param documentJsonCreator {@link #documentJsonCreator}
+	 */
+	public WriteDocumentsToElasticsearchIndexChain(final String index, final RestClientBuilder restClientBuilder,
+			final Function<Map<String, String>, String> idProvider, final BiConsumer<String, RestClient> indexCreator,
+			final Function<Map<String, String>, String> documentJsonCreator) {
+		this.index = index;
+		this.restClientBuilder = restClientBuilder;
+		this.idProvider = idProvider;
+		this.indexCreator = indexCreator;
+		this.documentJsonCreator = documentJsonCreator;
+	}
+
 	/**
 	 * @param index             {@link #index}
 	 * @param restClientBuilder {@link #restClientBuilder}
@@ -64,11 +81,8 @@ public class WriteDocumentsToElasticsearchIndexChain implements Chain<Stream<Map
 	public WriteDocumentsToElasticsearchIndexChain(final String index, final RestClientBuilder restClientBuilder,
 			final ObjectMapper objectMapper, final Function<Map<String, String>, String> idProvider,
 			final BiConsumer<String, RestClient> indexCreator) {
-		this.index = index;
-		this.restClientBuilder = restClientBuilder;
-		this.objectMapper = objectMapper;
-		this.idProvider = idProvider;
-		this.indexCreator = indexCreator;
+		this(index, restClientBuilder, idProvider, defaultIndexCreator(objectMapper),
+				defaultDocumentJsonCreator(objectMapper));
 	}
 
 	/**
@@ -122,13 +136,7 @@ public Void run(final Stream<Map<String, String>> input) {
 			}
 
 			input.forEach(document -> {
-				final String documentJson;
-				try {
-					documentJson = objectMapper.writeValueAsString(document);
-				} catch (final JsonProcessingException jsonProcessingException) {
-					throw new IllegalStateException("error creating json for document " + document,
-							jsonProcessingException);
-				}
+				final String documentJson = documentJsonCreator.apply(document);
 
 				final Request indexRequest = createIndexRequest(document);
 				indexRequest.setJsonEntity(documentJson);
@@ -177,6 +185,24 @@ private synchronized void createIndexIfNotExists(final RestClient restClient) th
 		this.indexCreator.accept(index, restClient);
 	}
 
+	/**
+	 * creates the default {@link Function} for creating json documents from input
+	 * documents for this chain
+	 * 
+	 * @param objectMapper {@link ObjectMapper} for json operations
+	 * @return default {@link #documentJsonCreator}
+	 */
+	public static Function<Map<String, String>, String> defaultDocumentJsonCreator(final ObjectMapper objectMapper) {
+		return document -> {
+			try {
+				return objectMapper.writeValueAsString(document);
+			} catch (final JsonProcessingException jsonProcessingException) {
+				throw new IllegalStateException("error creating json for document " + document,
+						jsonProcessingException);
+			}
+		};
+	}
+
 	/**
 	 * Realizes the default way of creating an elasticsearch index using the method
 	 * from