fixed #352

fixed #351
fixed #347
fixed #334
fixed #319

Author: paxtonhare

examples/spring-batch/marklogic-config/databases/trace-database.json

@@ -28,24 +28,40 @@
     {
       "scalar-type": "string",
       "namespace-uri": "",
-      "localname": "traceId",
+      "localname": "hasError",
       "collation": "http://marklogic.com/collation/codepoint",
       "range-value-positions": false,
       "invalid-values": "reject"
     },
     {
       "scalar-type": "string",
       "namespace-uri": "",
-      "localname": "identifier",
+      "localname": "flowType",
       "collation": "http://marklogic.com/collation/codepoint",
       "range-value-positions": false,
       "invalid-values": "reject"
     },
     {
-      "scalar-type": "unsignedInt",
-      "namespace-uri": "http://marklogic.com/data-hub/debug",
-      "localname": "is-debugging-enabled",
-      "collation": "",
+      "scalar-type": "string",
+      "namespace-uri": "",
+      "localname": "jobId",
+      "collation": "http://marklogic.com/collation/codepoint",
+      "range-value-positions": false,
+      "invalid-values": "reject"
+    },
+    {
+      "scalar-type": "string",
+      "namespace-uri": "",
+      "localname": "traceId",
+      "collation": "http://marklogic.com/collation/codepoint",
+      "range-value-positions": false,
+      "invalid-values": "reject"
+    },
+    {
+      "scalar-type": "string",
+      "namespace-uri": "",
+      "localname": "identifier",
+      "collation": "http://marklogic.com/collation/codepoint",
       "range-value-positions": false,
       "invalid-values": "reject"
     },

marklogic-data-hub/build.gradle

@@ -19,14 +19,10 @@ sourceCompatibility = 1.8
 targetCompatibility = 1.8
 
 dependencies {
-    compile 'org.springframework.batch:spring-batch-core:3.0.6.RELEASE'
-    compile 'org.springframework:spring-jdbc:4.2.6.RELEASE'
-    compile 'com.marklogic:java-client-api:3.0.5'
+    compile 'com.marklogic:marklogic-client-api:4.0-SNAPSHOT'
     compile 'com.marklogic:ml-javaclient-util:2.10.0'
     compile 'com.marklogic:ml-app-deployer:2.4.0'
-    compile 'com.marklogic:marklogic-spring-batch-core:0.7.0'
     compile 'commons-io:commons-io:2.4'
-    testCompile 'org.springframework.batch:spring-batch-test:3.0.6.RELEASE'
     testCompile 'junit:junit:4.12'
     testCompile 'xmlunit:xmlunit:1.3'
     testCompile 'org.skyscreamer:jsonassert:1.3.0'

marklogic-data-hub/src/main/java/com/marklogic/hub/DataHub.java

@@ -516,6 +516,15 @@ public void install(HubDeployStatusListener listener) {
         deployer.deploy(config);
     }
 
+    public void updateIndexes() {
+        AppConfig config = getAppConfig();
+        HubAppDeployer deployer = new HubAppDeployer(client, adminManager, null);
+        List<Command> commands = new ArrayList<>();
+        commands.add(new DeployHubDatabasesCommand(hubConfig));
+        deployer.setCommands(commands);
+        deployer.deploy(config);
+    }
+
     /**
      * Uninstalls the data hub configuration and server-side modules from MarkLogic
      */

marklogic-data-hub/src/main/java/com/marklogic/hub/FlowManager.java

@@ -16,47 +16,49 @@
 package com.marklogic.hub;
 
 import com.marklogic.client.DatabaseClient;
-import com.marklogic.client.DatabaseClientFactory;
+import com.marklogic.client.datamovement.DataMovementManager;
+import com.marklogic.client.datamovement.JobTicket;
+import com.marklogic.client.datamovement.QueryBatcher;
 import com.marklogic.client.extensions.ResourceManager;
 import com.marklogic.client.extensions.ResourceServices.ServiceResult;
 import com.marklogic.client.extensions.ResourceServices.ServiceResultIterator;
 import com.marklogic.client.io.DOMHandle;
 import com.marklogic.client.util.RequestParameters;
-import com.marklogic.hub.flow.Flow;
-import com.marklogic.hub.flow.FlowComplexity;
-import com.marklogic.hub.flow.FlowType;
-import com.marklogic.hub.flow.SimpleFlow;
-import com.marklogic.spring.batch.hub.FlowConfig;
-import com.marklogic.spring.batch.hub.RunHarmonizeFlowConfig;
-import com.marklogic.spring.batch.hub.StagingConfig;
-import org.springframework.batch.core.Job;
-import org.springframework.batch.core.JobExecution;
-import org.springframework.batch.core.JobParameters;
-import org.springframework.batch.core.JobParametersBuilder;
-import org.springframework.batch.core.launch.JobLauncher;
-import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import com.marklogic.hub.collector.Collector;
+import com.marklogic.hub.collector.ServerCollector;
+import com.marklogic.hub.flow.*;
+import com.marklogic.hub.job.Job;
+import com.marklogic.hub.job.JobManager;
 import org.w3c.dom.Document;
 import org.w3c.dom.Element;
 import org.w3c.dom.Node;
 import org.w3c.dom.NodeList;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 public class FlowManager extends ResourceManager {
     private static final String HUB_NS = "http://marklogic.com/data-hub";
     private static final String NAME = "flow";
 
-    private DatabaseClient client;
+    private DatabaseClient stagingClient;
+    private DatabaseClient finalClient;
+    private DatabaseClient jobClient;
     private HubConfig hubConfig;
+    private JobManager jobManager;
+
+    private DataMovementManager dataMovementManager;
 
     public FlowManager(HubConfig hubConfig) {
         super();
         this.hubConfig = hubConfig;
-        this.client = hubConfig.newStagingClient();
-        this.client.init(NAME, this);
+        this.stagingClient = hubConfig.newStagingClient();
+        this.finalClient = hubConfig.newFinalClient();
+        this.jobClient = hubConfig.newJobDbClient();
+        this.jobManager = new JobManager(this.jobClient);
+        this.dataMovementManager = this.stagingClient.newDataMovementManager();
+        this.stagingClient.init(NAME, this);
     }
 
     /**
@@ -103,8 +105,7 @@ public List<Flow> getFlows(String entityName) {
      * @return the flow
      */
     public Flow getFlow(String entityName, String flowName) {
-        Flow flow = getFlow(entityName, flowName, null);
-        return flow;
+        return getFlow(entityName, flowName, null);
     }
 
     public Flow getFlow(String entityName, String flowName, FlowType flowType) {
@@ -124,27 +125,8 @@ public Flow getFlow(String entityName, String flowName, FlowType flowType) {
         return flowFromXml(parent.getDocumentElement());
     }
 
-    private ConfigurableApplicationContext buildApplicationContext(Flow flow, JobStatusListener statusListener) {
-        AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
-        ctx.register(StagingConfig.class);
-        ctx.register(FlowConfig.class);
-        ctx.register(RunHarmonizeFlowConfig.class);
-        ctx.getBeanFactory().registerSingleton("hubConfig", hubConfig);
-        ctx.getBeanFactory().registerSingleton("flow", flow);
-        ctx.getBeanFactory().registerSingleton("statusListener", statusListener);
-        ctx.refresh();
-        return ctx;
-    }
-
-    private JobParameters buildJobParameters(Flow flow, int batchSize, int threadCount) {
-        JobParametersBuilder jpb = new JobParametersBuilder();
-        jpb.addLong("batchSize", Integer.toUnsignedLong(batchSize));
-        jpb.addLong("threadCount", Integer.toUnsignedLong(threadCount));
-        jpb.addString("uid", UUID.randomUUID().toString());
-        jpb.addString("flowType", flow.getType().toString());
-        jpb.addString("entity", flow.getEntityName());
-        jpb.addString("flow", flow.getName());
-        return jpb.toJobParameters();
+    public JobTicket runFlow(Flow flow, int batchSize, int threadCount, JobStatusListener statusListener) {
+        return runFlow(flow, batchSize, threadCount, HubDatabase.STAGING, HubDatabase.FINAL, null, statusListener);
     }
 
     /**
@@ -155,20 +137,83 @@ private JobParameters buildJobParameters(Flow flow, int batchSize, int threadCou
      * @param statusListener - the callback to receive job status updates
      * @return a JobExecution instance
      */
-    public JobExecution runFlow(Flow flow, int batchSize, int threadCount, JobStatusListener statusListener) {
-        JobExecution result = null;
-        try {
-            ConfigurableApplicationContext ctx = buildApplicationContext(flow, statusListener);
-
-            JobParameters params = buildJobParameters(flow, batchSize, threadCount);
-            JobLauncher launcher = ctx.getBean(JobLauncher.class);
-            Job job = ctx.getBean(Job.class);
-            result = launcher.run(job, params);
-        } catch (Exception e) {
-            e.printStackTrace();
+    public JobTicket runFlow(Flow flow, int batchSize, int threadCount, HubDatabase srcDb, HubDatabase destDb, Map<String, Object> options, JobStatusListener statusListener) {
+
+        Collector c = flow.getCollector();
+        if (c instanceof ServerCollector) {
+            ((ServerCollector)c).setClient(stagingClient);
+        }
+
+        AtomicLong successfulEvents = new AtomicLong(0);
+        AtomicLong failedEvents = new AtomicLong(0);
+        AtomicLong successfulBatches = new AtomicLong(0);
+        AtomicLong failedBatches = new AtomicLong(0);
+
+        Vector<String> uris = c.run(options);
+
+        DatabaseClient srcClient;
+        if (srcDb.equals(HubDatabase.STAGING)) {
+            srcClient = this.stagingClient;
         }
+        else {
+            srcClient = this.finalClient;
+        }
+        String targetDatabase;
+        if (destDb.equals(HubDatabase.STAGING)) {
+            targetDatabase = hubConfig.stagingDbName;
+        }
+        else {
+            targetDatabase = hubConfig.finalDbName;
+        }
+
+        FlowRunner flowRunner = new FlowRunner(srcClient, targetDatabase, flow);
+        AtomicInteger count = new AtomicInteger(0);
+        ArrayList<String> errorMessages = new ArrayList<>();
+        QueryBatcher queryBatcher = dataMovementManager.newQueryBatcher(uris.iterator())
+            .withBatchSize(batchSize)
+            .withThreadCount(threadCount)
+            .onUrisReady(batch -> {
+                try {
+                    RunFlowResponse response = flowRunner.run(batch.getJobTicket().getJobId(), batch.getItems(), options);
+                    failedEvents.addAndGet(response.errorCount);
+                    successfulEvents.addAndGet(response.totalCount - response.errorCount);
+                    successfulBatches.addAndGet(1);
+                    count.addAndGet(1);
+                }
+                catch(Exception e) {
+                    errorMessages.add(e.toString());
+                }
+            })
+            .onQueryFailure(failure -> {
+               failedBatches.addAndGet(1);
+               failedEvents.addAndGet(batchSize);
+            });
+        JobTicket jobTicket = dataMovementManager.startJob(queryBatcher);
+        jobManager.saveJob(Job.withFlow(flow)
+            .withJobId(jobTicket.getJobId())
+        );
+
+        new Thread(() -> {
+            queryBatcher.awaitCompletion();
+
+            if (statusListener != null) {
+                statusListener.onJobFinished();
+            }
+            dataMovementManager.stopJob(queryBatcher);
+
+            // store the thing in MarkLogic
+            Job job = Job.withFlow(flow)
+                .withJobId(jobTicket.getJobId())
+                .setCounts(successfulEvents.get(), failedEvents.get(), successfulBatches.get(), failedBatches.get())
+                .withEndTime(new Date());
+
+            if (errorMessages.size() > 0) {
+                job.withJobOutput(String.join("\n", errorMessages));
+            }
+            jobManager.saveJob(job);
+        }).start();
 
-        return result;
+        return jobTicket;
     }
 
     /**
@@ -185,7 +230,7 @@ public static Flow flowFromXml(Element doc) {
             complexity = elements.item(0).getTextContent();
         }
 
-        if (complexity.equals(FlowComplexity.SIMPLE.toString())) {
+        if (complexity != null && complexity.equals(FlowComplexity.SIMPLE.toString())) {
             f = new SimpleFlow(doc);
         }
 

marklogic-data-hub/src/main/java/com/marklogic/hub/HubConfig.java

@@ -35,6 +35,7 @@ public class HubConfig extends LoggingObject {
     public static final String OLD_HUB_CONFIG_DIR = "marklogic-config";
     public static final String HUB_CONFIG_DIR = "hub-internal-config";
     public static final String USER_CONFIG_DIR = "user-config";
+    public static final String SEARCH_OPTIONS_FILE = "all-search-options.xml";
 
     public static final String DEFAULT_HOST = "localhost";
 

marklogic-data-hub/src/main/java/com/marklogic/hub/HubDatabase.java

@@ -0,0 +1,24 @@
+package com.marklogic.hub;
+
+public enum HubDatabase {
+    STAGING("staging"),
+    FINAL("final");
+
+    private String type;
+    HubDatabase(String type) {
+        this.type = type;
+    }
+
+    public static HubDatabase getHubDatabase(String status) {
+        for (HubDatabase hubDatabase : HubDatabase.values()) {
+            if (hubDatabase.toString().equals(status)) {
+                return hubDatabase;
+            }
+        }
+        return null;
+    }
+
+    public String toString() {
+        return this.type;
+    }
+}

marklogic-data-hub/src/main/java/com/marklogic/hub/JobStatusListener.java

@@ -1,6 +1,6 @@
 package com.marklogic.hub;
 
 public interface JobStatusListener {
-    void onStatusChange(long jobId, int percentComplete, String message);
+    void onStatusChange(String jobId, int percentComplete, String message);
     void onJobFinished();
 }

marklogic-data-hub/src/main/java/com/marklogic/hub/collector/AbstractCollector.java

@@ -19,6 +19,7 @@
 
 import javax.xml.stream.XMLStreamException;
 import javax.xml.stream.XMLStreamWriter;
+import java.util.Map;
 import java.util.Vector;
 
 public abstract class AbstractCollector implements Collector {
@@ -35,7 +36,7 @@ public PluginType getType() {
     }
 
     @Override
-    public abstract Vector<String> run();
+    public abstract Vector<String> run(Map<String, Object> options);
 
     @Override
     public abstract void serialize(XMLStreamWriter serializer) throws XMLStreamException;

marklogic-data-hub/src/main/java/com/marklogic/hub/collector/Collector.java

@@ -19,11 +19,12 @@
 
 import javax.xml.stream.XMLStreamException;
 import javax.xml.stream.XMLStreamWriter;
+import java.util.Map;
 import java.util.Vector;
 
 public interface Collector {
 
     PluginType getType();
-    Vector<String> run();
+    Vector<String> run(Map<String, Object> options);
     void serialize(XMLStreamWriter serializer) throws XMLStreamException;
 }

marklogic-data-hub/src/main/java/com/marklogic/hub/collector/QueryCollector.java

@@ -17,6 +17,7 @@
 
 import com.marklogic.hub.plugin.PluginType;
 
+import java.util.Map;
 import java.util.Vector;
 
 public class QueryCollector extends ServerCollector {
@@ -28,7 +29,7 @@ public QueryCollector() {
     }
 
     @Override
-    public Vector<String> run() {
+    public Vector<String> run(Map<String, Object> options) {
         return null;
     }