MSE-4933 To get the the esPublishDate field as accurate as possible, …

vpro-shared-elasticsearch-client/src/main/java/nl/vpro/elasticsearchclient/BulkRequestEntry.java

@@ -1,9 +1,11 @@
 package nl.vpro.elasticsearchclient;
 
 import lombok.Getter;
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.Map;
+import java.util.function.Consumer;
 import java.util.function.UnaryOperator;
 
 import com.fasterxml.jackson.databind.JsonNode;
@@ -26,11 +28,37 @@ public class BulkRequestEntry {
     final Map<String, String> mdc;
     final String id;
 
-    public BulkRequestEntry(ObjectNode action, ObjectNode source, UnaryOperator<String> unalias, Map<String, String> mdc) {
+    @Getter
+    @Setter
+    Consumer<ObjectNode> sourceConsumer;
+
+
+    public BulkRequestEntry(
+        ObjectNode action,
+        ObjectNode source,
+        UnaryOperator<String> unalias,
+        Map<String, String> mdc) {
+        this(action, source, unalias, mdc, null);
+    }
+    @lombok.Builder
+    BulkRequestEntry(
+        ObjectNode action,
+        ObjectNode source,
+        UnaryOperator<String> unalias,
+        Map<String, String> mdc,
+        Consumer<ObjectNode> sourceConsumer) {
         this.action = action;
         this.source = source;
         this.mdc = mdc;
         this.id = idFromActionNode(action, unalias);
+        this.sourceConsumer = sourceConsumer;
+    }
+
+
+    public void use() {
+        if (sourceConsumer != null) {
+            sourceConsumer.accept(source);
+        }
     }
 
     public static String idFromActionNode(ObjectNode action) {

vpro-shared-elasticsearch-client/src/main/java/nl/vpro/elasticsearchclient/IndexHelper.java

@@ -985,12 +985,12 @@ private final BulkRequestEntry _indexRequest(String type, String id, Integer ver
         }
 
 
-          ObjectNode objectNode  = objectMapper.valueToTree(o);
-        for (Consumer<ObjectNode> c : consumers) {
-            c.accept(objectNode);
-        }
-
-        return new BulkRequestEntry(actionLine, objectNode, this::unalias, mdcSupplier.get());
+        ObjectNode objectNode  = objectMapper.valueToTree(o);
+        return new BulkRequestEntry(actionLine, objectNode, this::unalias, mdcSupplier.get(), (on) -> {
+            for (Consumer<ObjectNode> c : consumers) {
+                c.accept(on);
+            }
+        });
     }
 
     @SafeVarargs
@@ -1002,12 +1002,14 @@ private final BulkRequestEntry _updateRequest(String id, Object o, Consumer<Obje
         update.put(RETRY_ON_CONFLICT, 3);
         ObjectNode updateNode  = objectMapper.createObjectNode();
         ObjectNode objectNode = objectMapper.valueToTree(o);
-        for (Consumer<ObjectNode> c : consumers) {
-            c.accept(objectNode);
-        }
+
         updateNode.set(Fields.DOC, objectNode);
         updateNode.put(Fields.DOC_AS_UPSERT, false);
-        return new BulkRequestEntry(actionLine, updateNode, this::unalias, mdcSupplier.get());
+        return new BulkRequestEntry(actionLine, updateNode, this::unalias, mdcSupplier.get(), (on) -> {
+            for (Consumer<ObjectNode> c : consumers) {
+                c.accept(on);
+            }
+        });
     }
 
     /**
@@ -1026,7 +1028,6 @@ public BulkRequestEntry indexRequest(String type, String id, Object o, String ro
     /**
      * @deprecated Types are deprecated in elasticsearch, and will disappear in 8.
      */
-    @SuppressWarnings("DeprecatedIsStillUsed")
     @Deprecated
     public BulkRequestEntry deleteRequest(String type, String id) {
         return _deleteRequest(type, id);