DHFPROD-7701: Restoring deleted code that DHHCE needs

Author: rjrudin

marklogic-data-hub/src/main/java/com/marklogic/hub/flow/FlowRunner.java

@@ -129,4 +129,12 @@ public interface FlowRunner {
      * @throws TimeoutException if times out
      */
     void awaitCompletion(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException;
+
+    /**
+     * Sets the status change listener on the flowrunner object
+     * @param listener - the listener for when the status changes
+     * @return the flow runner object
+     */
+    FlowRunner onStatusChanged(FlowStatusListener listener);
+
 }

marklogic-data-hub/src/main/java/com/marklogic/hub/flow/FlowStatusListener.java

@@ -0,0 +1,23 @@
+package com.marklogic.hub.flow;
+
+
+import com.marklogic.hub.step.impl.Step;
+
+/**
+ * This is no longer used in DHF, but is still required by DHCCE.
+ */
+public interface FlowStatusListener {
+
+    /**
+     *
+     * @param jobId - the id of the running job
+     * @param step -  the current running step
+     * @param jobStatus - status of the running job
+     * @param percentComplete - the percentage of completeness expressed as an int
+     * @param successfulEvents - counter for success
+     * @param failedEvents - counter for failure
+     * @param message - the message you'd like to send along with it
+     */
+    void onStatusChanged(String jobId, Step step, String jobStatus, int percentComplete, long successfulEvents, long failedEvents, String message) ;
+
+}

marklogic-data-hub/src/main/java/com/marklogic/hub/flow/impl/FlowRunnerImpl.java

@@ -68,6 +68,8 @@ public class FlowRunnerImpl implements FlowRunner {
 
     private Queue<String> jobQueue = new ConcurrentLinkedQueue<>();
 
+    private List<FlowStatusListener> flowStatusListeners = new ArrayList<>();
+
     private ThreadPoolExecutor threadPool;
 
     public FlowRunnerImpl() {
@@ -117,6 +119,12 @@ public FlowRunnerImpl(HubConfig hubConfig) {
         this(hubConfig.newHubClient());
     }
 
+    @Override
+    public FlowRunner onStatusChanged(FlowStatusListener listener) {
+        this.flowStatusListeners.add(listener);
+        return this;
+    }
+
     @Deprecated
     public RunFlowResponse runFlow(String flowName) {
         return runFlow(flowName, null, null, new HashMap<>(), new HashMap<>());
@@ -317,6 +325,9 @@ public void run() {
             Map<String, RunStepResponse> stepOutputs = new HashMap<>();
             String stepNum = null;
 
+            final long[] currSuccessfulEvents = {0};
+            final long[] currFailedEvents = {0};
+            final int[] currPercentComplete = {0};
             while (! stepQueue.isEmpty()) {
                 stepNum = stepQueue.poll();
                 runningStep = runningFlow.getSteps().get(stepNum);
@@ -339,8 +350,16 @@ public void run() {
                             if(flow.isStopOnError()){
                                 stopJobOnError(jobId);
                             }
+                        })
+                        .onStatusChanged((jobId, percentComplete, jobStatus, successfulEvents, failedEvents, message) ->{
+                            flowStatusListeners.forEach((FlowStatusListener listener) -> {
+                                currSuccessfulEvents[0] = successfulEvents;
+                                currFailedEvents[0] = failedEvents;
+                                currPercentComplete[0] = percentComplete;
+                                listener.onStatusChanged(jobId, runningStep, jobStatus, percentComplete, successfulEvents, failedEvents, runningStep.getName() + " : " + message);
+                            });
                         });
-
+                    //If property values are overriden in UI, use those values over any other.
                     if(flow.getOverrideStepConfig() != null) {
                         stepRunner.withStepConfig(flow.getOverrideStepConfig());
                     }
@@ -371,7 +390,15 @@ public void run() {
                     else{
                         stepResp.withStatus(JobStatus.FAILED_PREFIX + stepNum);
                     }
-
+                    RunStepResponse finalStepResp = stepResp;
+                    try {
+                        flowStatusListeners.forEach((FlowStatusListener listener) -> {
+                            listener.onStatusChanged(jobId, runningStep, JobStatus.FAILED.toString(), currPercentComplete[0], currSuccessfulEvents[0], currFailedEvents[0],
+                                runningStep.getName() + " " + Arrays.toString(finalStepResp.stepOutput.toArray()));
+                        });
+                    } catch (Exception ex) {
+                        logger.error("Unable to invoke FlowStatusListener: " + ex.getMessage());
+                    }
                     if(runningFlow.isStopOnError()) {
                         stopJobOnError(runningJobId);
                     }
@@ -438,6 +465,24 @@ else if (!isJobSuccess.get()) {
                     }
                 }
 
+                if (!isJobSuccess.get()) {
+                    try {
+                        flowStatusListeners.forEach((FlowStatusListener listener) -> {
+                            listener.onStatusChanged(jobId, runningStep, jobStatus.toString(), currPercentComplete[0], currSuccessfulEvents[0], currFailedEvents[0], JobStatus.FAILED.toString());
+                        });
+                    } catch (Exception ex) {
+                        logger.error("Unable to invoke FlowStatusListener: " + ex.getMessage());
+                    }
+                } else {
+                    try {
+                        flowStatusListeners.forEach((FlowStatusListener listener) -> {
+                            listener.onStatusChanged(jobId, runningStep, jobStatus.toString(), currPercentComplete[0], currSuccessfulEvents[0], currFailedEvents[0], JobStatus.FINISHED.toString());
+                        });
+                    } catch (Exception ex) {
+                        logger.error("Unable to invoke FlowStatusListener: " + ex.getMessage());
+                    }
+                }
+
                 jobQueue.remove();
                 stepsMap.remove(jobId);
                 flowMap.remove(jobId);

marklogic-data-hub/src/main/java/com/marklogic/hub/mapping/MappingValidator.java

@@ -0,0 +1,28 @@
+package com.marklogic.hub.mapping;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.marklogic.client.DatabaseClient;
+import com.marklogic.client.extensions.ResourceManager;
+import com.marklogic.client.io.Format;
+import com.marklogic.client.io.JacksonHandle;
+import com.marklogic.client.io.StringHandle;
+import com.marklogic.client.util.RequestParameters;
+
+/**
+ * This is no longer used in DHF, but is still required by DHCCE.
+ */
+public class MappingValidator extends ResourceManager {
+
+    public MappingValidator(DatabaseClient client) {
+        super();
+        client.init("mlMappingValidator", this);
+    }
+
+    public JsonNode validateJsonMapping(String jsonMapping, String uri) {
+        RequestParameters params = new RequestParameters();
+        params.add("uri", uri);
+        return getServices().post(params,
+            new StringHandle(jsonMapping).withFormat(Format.JSON), new JacksonHandle()).get();
+    }
+
+}

marklogic-data-hub/src/main/java/com/marklogic/hub/step/StepRunner.java

@@ -102,6 +102,13 @@ public interface StepRunner {
      */
     StepRunner onItemFailed(StepItemFailureListener listener);
 
+    /**
+     * Sets the status change listener on the flowrunner object
+     * @param listener - the listener for when the status changes
+     * @return the step runner object
+     */
+    StepRunner onStatusChanged(StepStatusListener listener);
+
     /**
      * Blocks until the step execution is complete.
      */

marklogic-data-hub/src/main/java/com/marklogic/hub/step/impl/QueryStepRunner.java

@@ -46,6 +46,7 @@ public class QueryStepRunner extends LoggingObject implements StepRunner {
     private int batchSize;
     private int threadCount;
     private Map<String, Object> combinedOptions;
+    private int previousPercentComplete;
     private boolean stopOnFailure = false;
     private String jobId;
     private boolean isFullOutput = false;
@@ -54,6 +55,7 @@ public class QueryStepRunner extends LoggingObject implements StepRunner {
 
     private List<StepItemCompleteListener> stepItemCompleteListeners = new ArrayList<>();
     private List<StepItemFailureListener> stepItemFailureListeners = new ArrayList<>();
+    private List<StepStatusListener> stepStatusListeners = new ArrayList<>();
     private Map<String, Object> stepConfig = new HashMap<>();
     private HubClient hubClient;
     private Thread runningThread = null;
@@ -132,6 +134,12 @@ public StepRunner onItemFailed(StepItemFailureListener listener) {
         return this;
     }
 
+    @Override
+    public StepRunner onStatusChanged(StepStatusListener listener) {
+        this.stepStatusListeners.add(listener);
+        return this;
+    }
+
     @Override
     public void awaitCompletion() {
         try {
@@ -243,18 +251,32 @@ public int getBatchSize(){
 
     private DiskQueue<String> runCollector(String sourceDatabase) {
         CollectorImpl collector = new CollectorImpl(hubClient, sourceDatabase);
+
+        stepStatusListeners.forEach((StepStatusListener listener) -> {
+            listener.onStatusChange(this.jobId, 0, JobStatus.RUNNING_PREFIX + step, 0, 0,  "running collector");
+        });
+
         return !isStopped.get() ? collector.run(this.flow.getName(), step, combinedOptions) : null;
     }
 
     private RunStepResponse runHarmonizer(RunStepResponse runStepResponse, Collection<String> uris) {
         StepMetrics stepMetrics = new StepMetrics();
         final int urisCount = uris != null ? uris.size() : 0;
+
+        stepStatusListeners.forEach((StepStatusListener listener) -> {
+            listener.onStatusChange(runStepResponse.getJobId(), 0, JobStatus.RUNNING_PREFIX + step, 0,0, "starting step execution");
+        });
+
         if (urisCount == 0) {
             logger.info("No items found to process");
             final String stepStatus = isStopped.get() ?
                 JobStatus.CANCELED_PREFIX + step :
                 JobStatus.COMPLETED_PREFIX + step;
 
+            stepStatusListeners.forEach((StepStatusListener listener) -> {
+                listener.onStatusChange(runStepResponse.getJobId(), 100, stepStatus, 0, 0,
+                    (stepStatus.contains(JobStatus.COMPLETED_PREFIX) ? "collector returned 0 items" : "job was stopped"));
+            });
             runStepResponse.setCounts(0,0,0,0,0);
             runStepResponse.withStatus(stepStatus);
 
@@ -346,6 +368,15 @@ private RunStepResponse runHarmonizer(RunStepResponse runStepResponse, Collectio
                         stepMetrics.getFailedBatches().addAndGet(1);
                     }
 
+                    int percentComplete = (int) (((double) stepMetrics.getSuccessfulBatchesCount() / batchCount) * 100.0);
+
+                    if (percentComplete != previousPercentComplete && (percentComplete % 5 == 0)) {
+                        previousPercentComplete = percentComplete;
+                        stepStatusListeners.forEach((StepStatusListener listener) -> {
+                            listener.onStatusChange(runStepResponse.getJobId(), percentComplete, JobStatus.RUNNING_PREFIX + step, stepMetrics.getSuccessfulEventsCount(), stepMetrics.getFailedEventsCount(), "");
+                        });
+                    }
+
                     if (stepItemCompleteListeners.size() > 0) {
                         response.completedItems.forEach((String item) -> {
                             stepItemCompleteListeners.forEach((StepItemCompleteListener listener) -> {
@@ -412,6 +443,10 @@ private RunStepResponse runHarmonizer(RunStepResponse runStepResponse, Collectio
 
             String stepStatus = determineStepStatus(stepMetrics);
 
+            stepStatusListeners.forEach((StepStatusListener listener) -> {
+                listener.onStatusChange(runStepResponse.getJobId(), 100, stepStatus, stepMetrics.getSuccessfulEventsCount(), stepMetrics.getFailedEventsCount(), "");
+            });
+
             dataMovementManager.stopJob(queryBatcher);
 
             runStepResponse.setCounts(urisCount, stepMetrics.getSuccessfulEventsCount(), stepMetrics.getFailedEventsCount(), stepMetrics.getSuccessfulBatchesCount(), stepMetrics.getFailedBatchesCount());