Merge pull request #2463 from marklogic/bugfix/dhfprod-2461

aebadirad · web-flow · commit 19b7f0a95f40 · 2019-05-09T11:45:55.000-07:00
DHFPROD-2461: In Manage Flow page first flow's info gets replaced by second flow's info when running second
diff --git a/marklogic-data-hub/src/main/java/com/marklogic/hub/flow/impl/FlowRunnerImpl.java b/marklogic-data-hub/src/main/java/com/marklogic/hub/flow/impl/FlowRunnerImpl.java
@@ -384,6 +384,7 @@ else if (!isJobSuccess.get()) {
                 stepsMap.remove(jobId);
                 flowMap.remove(jobId);
                 flowResp.remove(runningJobId);
+                runningFlow = null;
                 if (!jobQueue.isEmpty()) {
                     initializeFlow((String) jobQueue.peek());
                 } else {
@@ -435,6 +436,7 @@ public void afterExecute(Runnable r, Throwable t) {
                 //Run the next queued flow if stop-on-error is set or if the step queue is empty
                 if(((FlowRunnerTask)r).getStepQueue().isEmpty() || runningFlow.isStopOnError()) {
                     jobQueue.remove();
+                    runningFlow = null;
                     if (!jobQueue.isEmpty()) {
                         initializeFlow((String) jobQueue.peek());
                     } else {
diff --git a/marklogic-data-hub/src/main/resources/hub-internal-config/databases/job-database.json b/marklogic-data-hub/src/main/resources/hub-internal-config/databases/job-database.json
@@ -17,21 +17,45 @@
       "range-value-positions": false,
       "invalid-values": "reject"
     },
+    {
+      "scalar-type": "string",
+      "namespace-uri": "",
+      "localname": "flow",
+      "collation": "http://marklogic.com/collation/codepoint",
+      "range-value-positions": false,
+      "invalid-values": "ignore"
+    },
     {
       "scalar-type": "dateTime",
       "namespace-uri": "",
       "localname": "startTime",
       "collation": "",
       "range-value-positions": false,
-      "invalid-values": "reject"
+      "invalid-values": "ignore"
     },
     {
       "scalar-type": "dateTime",
       "namespace-uri": "",
       "localname": "endTime",
       "collation": "",
       "range-value-positions": false,
-      "invalid-values": "reject"
+      "invalid-values": "ignore"
+    },
+    {
+      "scalar-type": "dateTime",
+      "namespace-uri": "",
+      "localname": "timeStarted",
+      "collation": "",
+      "range-value-positions": false,
+      "invalid-values": "ignore"
+    },
+    {
+      "scalar-type": "dateTime",
+      "namespace-uri": "",
+      "localname": "timeEnded",
+      "collation": "",
+      "range-value-positions": false,
+      "invalid-values": "ignore"
     },
     {
       "scalar-type": "string",
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/5/extensions/jobs.sjs b/marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/5/extensions/jobs.sjs
@@ -21,6 +21,7 @@ function get(context, params) {
   let jobId = params["jobid"];
   let status = params["status"];
   let flow = params["flow-name"];
+  let latest = params["latest"];
 
   let resp = null;
 
@@ -36,6 +37,9 @@ function get(context, params) {
   else if (fn.exists(flow)) {
     resp = datahub.jobs.getJobDocsByFlow(flow);
   }
+  else if (fn.exists(latest)) {
+    resp = datahub.jobs.getLastestJobDocPerFlow();
+  }
   else{
     fn.error(null,"RESTAPI-SRVEXERR",  Sequence.from([400, "Bad Request", "Incorrect options"]));
   }
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/5/impl/jobs.sjs b/marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/5/impl/jobs.sjs
@@ -132,6 +132,21 @@ class Jobs {
     return docs;
   }
 
+  getLastestJobDocPerFlow() {
+    return this.hubutils.queryLatest(function(){
+      let flowNames = cts.values(cts.jsonPropertyReference("flow"));
+      let timeQuery = [];
+      for(let flowName of flowNames) {
+        let time = cts.values(cts.jsonPropertyReference("timeStarted"), null, ["descending","limit=1"], cts.jsonPropertyRangeQuery("flow", "=", flowName));
+        timeQuery.push(cts.rangeQuery(cts.jsonPropertyReference("timeStarted"), "=", time));
+      }
+      let results = cts.search(cts.orQuery(timeQuery));
+      if(results) {
+       return results.toArray();
+      }
+    }, this.config.JOBDATABASE);
+  }
+
   getJobDocsByFlow(flowName) {
     return this.hubutils.queryLatest(function() {
       let query = [cts.collectionQuery('Job'),  cts.jsonPropertyValueQuery('flow', flowName, "case-insensitive")];
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/5/impl/step.sjs b/marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/5/impl/step.sjs
@@ -77,20 +77,20 @@ class Step {
   }
 
   getStepsByType(type = 'custom') {
-    let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('type', type)];
+    let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('type', type, 'case-insensitive')];
     return cts.search(cts.andQuery(query)).toArray();
   }
 
   getStepByNameAndType(name, type = 'custom') {
-    let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('name', name), cts.jsonPropertyValueQuery('type', type)];
+    let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('name', name, 'case-insensitive'), cts.jsonPropertyValueQuery('type', type, 'case-insensitive')];
     let doc = fn.head(cts.search(cts.andQuery(query)));
     if(doc) {
       return doc.toObject();
     }
   }
 
   getStepProcessor(flow, name, type = 'custom') {
-    let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('name', name), cts.jsonPropertyValueQuery('type', type)];
+    let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('name', name, 'case-insensitive'), cts.jsonPropertyValueQuery('type', type, 'case-insensitive')];
     let doc = fn.head(cts.search(cts.andQuery(query)));
     if(doc){
       doc = doc.toObject();
diff --git a/ml-data-hub-plugin/src/main/groovy/com/marklogic/gradle/task/RunFlowTask.groovy b/ml-data-hub-plugin/src/main/groovy/com/marklogic/gradle/task/RunFlowTask.groovy
@@ -134,7 +134,7 @@ class RunFlowTask extends HubTask {
         Map<String, Object> options = new HashMap<>()
         def optionsString;
         if(project.ext.properties.containsKey("optionsFile")){
-            def jsonFile = new File(project.ext.optionsFile)
+            def jsonFile = new File(project.ext.optionsFile.trim())
             optionsString = jsonFile.text
         }
         else if(project.ext.properties.containsKey("options")) {
diff --git a/web/src/main/ui/app/components/flows-new/edit-flow/edit-flow.component.ts b/web/src/main/ui/app/components/flows-new/edit-flow/edit-flow.component.ts
@@ -66,7 +66,7 @@ export class EditFlowComponent implements OnInit, OnDestroy {
   }
 
   ngOnDestroy(): void {
-    this.runningJobService.stopPolling();
+    this.runningJobService.stopPolling(this.flow.id);
   }
 
   getFlow() {
@@ -155,7 +155,7 @@ export class EditFlowComponent implements OnInit, OnDestroy {
       console.log('stop flow response', resp);
       this.flow = Flow.fromJSON(resp);
       this.getSteps();
-      this.runningJobService.stopPolling();
+      this.runningJobService.stopPollingAll();
     });
   }
   createStep(stepObject) {
diff --git a/web/src/main/ui/app/components/flows-new/edit-flow/mapping/mapping.component.ts b/web/src/main/ui/app/components/flows-new/edit-flow/mapping/mapping.component.ts
@@ -94,6 +94,9 @@ export class MappingComponent implements OnInit {
       } else if (this.step.options.sourceDatabase === this.envService.settings.finalDbName) {
         this.sourceDbType = 'FINAL';
       }
+      if(!this.step.options.collections || this.step.options.collections.length === 0){
+        this.step.options.collections = [`${this.step.name}`, 'mdm-content'];
+      }
       this.loadEntity();
       this.loadMap();
     }
diff --git a/web/src/main/ui/app/components/flows-new/manage-flows/manage-flows.component.ts b/web/src/main/ui/app/components/flows-new/manage-flows/manage-flows.component.ts
@@ -39,7 +39,7 @@ export class ManageFlowsComponent implements OnInit, OnDestroy {
     this.getFlows();
   }
   ngOnDestroy(): void {
-    this.runningJobService.stopPolling();
+    this.runningJobService.stopPollingAll();
   }
 
   createFlow(newFlow) {
@@ -84,6 +84,7 @@ export class ManageFlowsComponent implements OnInit, OnDestroy {
 
   runFlow(runObject): void {
     this.manageFlowsService.runFlow(runObject).subscribe(resp => {
+      // console.log('run enpoint', resp);
       // TODO add response check
       const flowIndex = this.flows.findIndex(flow => flow.id === runObject.id);
       this.pollFlow(flowIndex, runObject.id);
@@ -101,7 +102,7 @@ export class ManageFlowsComponent implements OnInit, OnDestroy {
     this.manageFlowsService.stopFlow(flowId).subscribe(resp => {
       console.log('stop flow response', resp);
       this.getFlows();
-      this.runningJobService.stopPolling();
+      this.runningJobService.stopPolling(flowId);
     });
   }
 
diff --git a/web/src/main/ui/app/components/flows-new/manage-flows/ui/manage-flows-ui.component.html b/web/src/main/ui/app/components/flows-new/manage-flows/ui/manage-flows-ui.component.html
@@ -121,7 +121,7 @@ <h1>Manage Flows</h1>
                     color="primary"
                     class="run-flow-button"
                     *ngIf="checkRunStatus(flow)"
-                    [disabled]="flow.steps.length ? false : true"
+                    [disabled]="checkRunDisabled(flow)"
                     (click)="openRunDialog(flow)"
             >
               <span>RUN</span>
diff --git a/web/src/main/ui/app/components/flows-new/manage-flows/ui/manage-flows-ui.component.ts b/web/src/main/ui/app/components/flows-new/manage-flows/ui/manage-flows-ui.component.ts
@@ -39,6 +39,7 @@ export class ManageFlowsUiComponent implements OnInit, AfterViewInit {
   }
 
   ngOnInit() {
+    console.log('manage flows', this.flows);
     this.dataSource = new MatTableDataSource<Flow>(this.flows);
   }
 
@@ -48,7 +49,7 @@ export class ManageFlowsUiComponent implements OnInit, AfterViewInit {
   }
 
   updateDataSource() {
-    this.dataSource.data = this.flows
+    this.dataSource.data = this.flows;
   }
 
   openRunDialog(flow: Flow) {
@@ -154,4 +155,15 @@ export class ManageFlowsUiComponent implements OnInit, AfterViewInit {
     }
   }
 
+  checkRunDisabled(flow: Flow): boolean {
+    if (flow.steps.length) {
+      if (flow.latestJob === null) {
+        return true;
+      }
+      return false;
+    } else {
+      return true;
+    }
+  }
+
 }
diff --git a/web/src/main/ui/app/components/jobs-new/services/running-job-service.ts b/web/src/main/ui/app/components/jobs-new/services/running-job-service.ts
@@ -7,43 +7,56 @@ import { Flow } from '../../flows-new/models/flow.model';
 
 @Injectable()
 export class RunningJobService {
-  private running: Subscription;
-  private flowRunning = new Subject<Flow>();
+  private subscriptions: any = {};
+  private subjects: any = {};
 
   constructor(
     private manageFlowsService: ManageFlowsService
   ) {}
 
-  stopPolling() {
-    if (this.running) {
-      this.running.unsubscribe();
+  stopPollingAll() {
+    if (this.subscriptions) {
+      Object.keys(this.subscriptions).forEach(flowId => {
+        this.subscriptions[flowId].unsubscribe();
+      });
+    }
+  }
+
+  stopPolling(flowId: string) {
+    if (this.subscriptions[flowId]) {
+      this.subscriptions[flowId].unsubscribe();
     }
   }
 
   pollFlowById(id: string) {
-    this.running = timer(0, 1000).pipe(
+    this.subscriptions[id] = new Subscription();
+    this.subjects[id] = new Subject<Flow>();
+
+    this.subscriptions[id] = timer(0, 2000).pipe(
       concatMap( () => this.manageFlowsService.getFlowById(id))
     ).subscribe(data => {
       const flow = Flow.fromJSON(data);
       this.checkJobStatus(flow);
-      this.flowRunning.next(flow);
+      this.subjects[id].next(flow);
     });
-    return this.flowRunning.asObservable();
+    return this.subjects[id].asObservable();
   }
 
   checkJobStatus(flow: Flow): boolean {
-    if (flow.latestJob && flow.latestJob.status) {
+    if (this.subscriptions[flow.id] && flow.latestJob && flow.latestJob.status) {
       let runStatus = flow.latestJob.status.replace('_', ' ');
       runStatus = runStatus.replace('-', ' ');
       runStatus = runStatus.split(' ');
       if (runStatus[0] === 'finished' || runStatus[0] === 'canceled' || runStatus[0] === 'failed') {
-        if ( this.running) {
-          this.running.unsubscribe();
+        if ( this.subscriptions[flow.id]) {
+          this.subscriptions[flow.id].unsubscribe();
         }
         return false;
       } else {
         return true;
       }
+    } else if (this.subscriptions[flow.id] && flow.latestJob === null) {
+      return true;
     } else {
       return false;
     }

Original file line number	Diff line number	Diff line change
`@@ -77,20 +77,20 @@ class Step {`
`77`	`77`	`}`
`78`	`78`
`79`	`79`	`getStepsByType(type = 'custom') {`
`80`		`- let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('type', type)];`
	`80`	`+ let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('type', type, 'case-insensitive')];`
`81`	`81`	`return cts.search(cts.andQuery(query)).toArray();`
`82`	`82`	`}`
`83`	`83`
`84`	`84`	`getStepByNameAndType(name, type = 'custom') {`
`85`		`- let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('name', name), cts.jsonPropertyValueQuery('type', type)];`
	`85`	`+ let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('name', name, 'case-insensitive'), cts.jsonPropertyValueQuery('type', type, 'case-insensitive')];`
`86`	`86`	`let doc = fn.head(cts.search(cts.andQuery(query)));`
`87`	`87`	`if(doc) {`
`88`	`88`	`return doc.toObject();`
`89`	`89`	`}`
`90`	`90`	`}`
`91`	`91`
`92`	`92`	`getStepProcessor(flow, name, type = 'custom') {`
`93`		`- let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('name', name), cts.jsonPropertyValueQuery('type', type)];`
	`93`	`+ let query = [cts.collectionQuery('http://marklogic.com/data-hub/step-definition'), cts.jsonPropertyValueQuery('name', name, 'case-insensitive'), cts.jsonPropertyValueQuery('type', type, 'case-insensitive')];`
`94`	`94`	`let doc = fn.head(cts.search(cts.andQuery(query)));`
`95`	`95`	`if(doc){`
`96`	`96`	`doc = doc.toObject();`
Original file line number	Diff line number	Diff line change
`@@ -134,7 +134,7 @@ class RunFlowTask extends HubTask {`
`134`	`134`	`Map<String, Object> options = new HashMap<>()`
`135`	`135`	`def optionsString;`
`136`	`136`	`if(project.ext.properties.containsKey("optionsFile")){`
`137`		`- def jsonFile = new File(project.ext.optionsFile)`
	`137`	`+ def jsonFile = new File(project.ext.optionsFile.trim())`
`138`	`138`	`optionsString = jsonFile.text`
`139`	`139`	`}`
`140`	`140`	`else if(project.ext.properties.containsKey("options")) {`
Original file line number	Diff line number	Diff line change
`@@ -66,7 +66,7 @@ export class EditFlowComponent implements OnInit, OnDestroy {`
`66`	`66`	`}`
`67`	`67`
`68`	`68`	`ngOnDestroy(): void {`
`69`		`- this.runningJobService.stopPolling();`
	`69`	`+ this.runningJobService.stopPolling(this.flow.id);`
`70`	`70`	`}`
`71`	`71`
`72`	`72`	`getFlow() {`
`@@ -155,7 +155,7 @@ export class EditFlowComponent implements OnInit, OnDestroy {`
`155`	`155`	`console.log('stop flow response', resp);`
`156`	`156`	`this.flow = Flow.fromJSON(resp);`
`157`	`157`	`this.getSteps();`
`158`		`- this.runningJobService.stopPolling();`
	`158`	`+ this.runningJobService.stopPollingAll();`
`159`	`159`	`});`
`160`	`160`	`}`
`161`	`161`	`createStep(stepObject) {`
Original file line number	Diff line number	Diff line change
`@@ -94,6 +94,9 @@ export class MappingComponent implements OnInit {`
`94`	`94`	`} else if (this.step.options.sourceDatabase === this.envService.settings.finalDbName) {`
`95`	`95`	`this.sourceDbType = 'FINAL';`
`96`	`96`	`}`
	`97`	`+ if(!this.step.options.collections \|\| this.step.options.collections.length === 0){`
	`98`	+ this.step.options.collections = [`${this.step.name}`, 'mdm-content'];
	`99`	`+ }`
`97`	`100`	`this.loadEntity();`
`98`	`101`	`this.loadMap();`
`99`	`102`	`}`
Original file line number	Diff line number	Diff line change
`@@ -121,7 +121,7 @@ <h1>Manage Flows</h1>`
`121`	`121`	`color="primary"`
`122`	`122`	`class="run-flow-button"`
`123`	`123`	`*ngIf="checkRunStatus(flow)"`
`124`		`- [disabled]="flow.steps.length ? false : true"`
	`124`	`+ [disabled]="checkRunDisabled(flow)"`
`125`	`125`	`(click)="openRunDialog(flow)"`
`126`	`126`	`>`
`127`	`127`	`<span>RUN</span>`