Merge branch 'feature/threading' into develop

Author: paxtonhare

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

@@ -15,12 +15,11 @@
  */
 package com.marklogic.hub.collector;
 
-import java.util.List;
+import com.marklogic.hub.plugin.PluginType;
 
 import javax.xml.stream.XMLStreamException;
 import javax.xml.stream.XMLStreamWriter;
-
-import com.marklogic.hub.plugin.PluginType;
+import java.util.Vector;
 
 public abstract class AbstractCollector implements Collector {
 
@@ -36,7 +35,7 @@ public PluginType getType() {
     }
 
     @Override
-    public abstract List<String> run();
+    public abstract Vector<String> run();
 
     @Override
     public abstract void serialize(XMLStreamWriter serializer) throws XMLStreamException;

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

@@ -15,16 +15,15 @@
  */
 package com.marklogic.hub.collector;
 
-import java.util.List;
+import com.marklogic.hub.plugin.PluginType;
 
 import javax.xml.stream.XMLStreamException;
 import javax.xml.stream.XMLStreamWriter;
-
-import com.marklogic.hub.plugin.PluginType;
+import java.util.Vector;
 
 public interface Collector {
 
     PluginType getType();
-    List<String> run();
+    Vector<String> run();
     void serialize(XMLStreamWriter serializer) throws XMLStreamException;
 }

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

@@ -15,10 +15,10 @@
  */
 package com.marklogic.hub.collector;
 
-import java.util.List;
-
 import com.marklogic.hub.plugin.PluginType;
 
+import java.util.Vector;
+
 public class QueryCollector extends ServerCollector {
 
     public static final String MODULE = "/com.marklogic.hub/collectors/query.xqy";
@@ -28,7 +28,7 @@ public QueryCollector() {
     }
 
     @Override
-    public List<String> run() {
+    public Vector<String> run() {
         return null;
     }
 

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

@@ -15,12 +15,6 @@
  */
 package com.marklogic.hub.collector;
 
-import java.util.ArrayList;
-import java.util.List;
-
-import javax.xml.stream.XMLStreamException;
-import javax.xml.stream.XMLStreamWriter;
-
 import com.marklogic.client.DatabaseClient;
 import com.marklogic.client.extensions.ResourceManager;
 import com.marklogic.client.extensions.ResourceServices.ServiceResult;
@@ -29,6 +23,10 @@
 import com.marklogic.client.util.RequestParameters;
 import com.marklogic.hub.plugin.PluginType;
 
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.XMLStreamWriter;
+import java.util.Vector;
+
 public class ServerCollector extends AbstractCollector {
 
     private DatabaseClient client = null;
@@ -61,24 +59,24 @@ public void serialize(XMLStreamWriter serializer) throws XMLStreamException {
     }
 
     @Override
-    public List<String> run() {
+    public Vector<String> run() {
         CollectorModule cm = new CollectorModule(client);
         return cm.run(getModule());
     }
 
-    static class CollectorModule  extends ResourceManager {
+    static class CollectorModule extends ResourceManager {
         static final public String NAME = "collector";
 
         public CollectorModule(DatabaseClient client) {
             super();
             client.init(NAME, this);
         }
 
-        public List<String> run(String moduleUri) {
+        public Vector<String> run(String moduleUri) {
             RequestParameters params = new RequestParameters();
             params.add("module-uri", moduleUri);
 
-            ServiceResultIterator resultItr = null;
+            ServiceResultIterator resultItr;
 
             resultItr = this.getServices().get(params);
 
@@ -87,7 +85,7 @@ public List<String> run(String moduleUri) {
             }
 
             ServiceResult res = resultItr.next();
-            JacksonDatabindHandle<ArrayList<String>> handle = new JacksonDatabindHandle<ArrayList<String>>(new ArrayList<String>());
+            JacksonDatabindHandle<Vector<String>> handle = new JacksonDatabindHandle<>(new Vector<String>());
             handle.getMapper().disableDefaultTyping();
             return res.getContent(handle).get();
         }

marklogic-data-hub/src/main/java/com/marklogic/spring/batch/hub/CollectorReader.java

@@ -1,22 +1,17 @@
 package com.marklogic.spring.batch.hub;
 
-import java.util.ArrayList;
-import java.util.List;
-
+import com.marklogic.hub.collector.Collector;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemStreamException;
-import org.springframework.batch.item.ItemStreamReader;
+import org.springframework.batch.item.support.SynchronizedItemStreamReader;
 
-import com.marklogic.client.helper.LoggingObject;
-import com.marklogic.hub.collector.Collector;
+import java.util.Vector;
 
-public class CollectorReader extends LoggingObject implements ItemStreamReader<String> {
+public class CollectorReader extends SynchronizedItemStreamReader<String> {
 
     private Collector collector;
 
-    private List<String> results;
-
-    private int index = 0;
+    private Vector<String> results;
 
     public CollectorReader(Collector collector) {
         this.collector = collector;
@@ -28,7 +23,7 @@ public void open(ExecutionContext executionContext) {
             this.results = collector.run();
         }
         else {
-            this.results = new ArrayList<>();
+            this.results = new Vector<>();
         }
 
         executionContext.putInt("totalItems", this.results.size());
@@ -38,9 +33,11 @@ public void open(ExecutionContext executionContext) {
     public String read() {
         String result = null;
 
-        if (results.size() > this.index) {
-            result = this.results.get(this.index);
-            index++;
+        try {
+            result = this.results.remove(0);
+        }
+        catch(ArrayIndexOutOfBoundsException e) {
+
         }
 
         return result;

marklogic-data-hub/src/main/java/com/marklogic/spring/batch/hub/FlowWriter.java

@@ -21,7 +21,6 @@ public class FlowWriter extends ResourceManager  implements ItemWriter<String> {
     private DatabaseClient client;
     private Flow flow;
     StringHandle handle;
-    RequestParameters params = new RequestParameters();
 
     public FlowWriter(DatabaseClient client, Flow flow) {
         super();
@@ -36,6 +35,7 @@ public FlowWriter(DatabaseClient client, Flow flow) {
     public void write(List<? extends String> items) {
 
         try {
+            RequestParameters params = new RequestParameters();
             params.put("identifier", items.toArray(new String[items.size()]));
             this.getServices().post(params, handle);
         }

marklogic-data-hub/src/main/java/com/marklogic/spring/batch/hub/RunHarmonizeFlowConfig.java

@@ -33,7 +33,10 @@ public class RunHarmonizeFlowConfig extends AbstractMarkLogicBatchConfig {
 
     @Bean
     public Job job(@Qualifier("step1") Step step1) {
-        return jobBuilderFactory.get("Harmonize").start(step1).listener(new JobExecutionListener() {
+        return jobBuilderFactory.get("Harmonize")
+            .start(step1)
+            .preventRestart()
+            .listener(new JobExecutionListener() {
             @Override
             public void beforeJob(JobExecution jobExecution) {
                 jobId = jobExecution.getJobId();

quick-start/src/main/java/com/marklogic/quickstart/service/FlowManagerService.java

@@ -38,7 +38,7 @@
 import java.util.UUID;
 
 @Service
-public class FlowManagerService extends LoggingObject{
+public class FlowManagerService extends LoggingObject {
 
     private static final String PROJECT_TMP_FOLDER = ".tmp";
     @Autowired

quick-start/src/main/ui/app/dashboard/dashboard.component.ts

@@ -24,9 +24,7 @@ export class DashboardComponent implements OnInit, OnDestroy {
   constructor(private projectService: ProjectService) {}
 
   getStatus() {
-    if (this.statsInterval) {
-      clearInterval(this.statsInterval);
-    }
+    this.ngOnDestroy();
 
     this.projectService.getStatus().subscribe((stats) => {
       this.stats = stats;
@@ -46,6 +44,7 @@ export class DashboardComponent implements OnInit, OnDestroy {
   ngOnDestroy() {
     if (this.statsInterval) {
       clearInterval(this.statsInterval);
+      delete this.statsInterval;
     }
   }
 

quick-start/src/main/ui/app/mlcp-ui/mlcp-ui.ts

@@ -481,7 +481,7 @@ export class MlcpUiComponent {
           if (setting.value) {
             const key = setting.field;
             let value = setting.value;
-            if (setting.type !== 'boolean') {
+            if (setting.type !== 'boolean' && setting.type !== 'number') {
               value = '"' + setting.value + '"';
             }
             this.addMlcpOption(options, key, value, true, true);