]> git.argeo.org Git - gpl/argeo-slc.git/blob - runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ProcessThread.java
Introduce JMX agent
[gpl/argeo-slc.git] / runtime / org.argeo.slc.core / src / main / java / org / argeo / slc / core / execution / ProcessThread.java
1 package org.argeo.slc.core.execution;
2
3 import java.util.ArrayList;
4 import java.util.Iterator;
5 import java.util.List;
6
7 import org.apache.commons.logging.Log;
8 import org.apache.commons.logging.LogFactory;
9 import org.argeo.slc.process.RealizedFlow;
10 import org.argeo.slc.process.SlcExecution;
11 import org.argeo.slc.process.SlcExecutionNotifier;
12
13 /** Thread of the SLC Process, starting the sub executions. */
14 public class ProcessThread extends Thread {
15 private final static Log log = LogFactory.getLog(ProcessThread.class);
16
17 private final AbstractExecutionModulesManager executionModulesManager;
18 private final SlcExecution slcProcess;
19 private final ThreadGroup processThreadGroup;
20 private final List<RealizedFlow> flowsToProcess = new ArrayList<RealizedFlow>();
21
22 private Boolean hadAnError = false;
23
24 public ProcessThread(
25 AbstractExecutionModulesManager executionModulesManager,
26 SlcExecution slcExecution) {
27 super(executionModulesManager.getProcessesThreadGroup(),
28 "SLC Process #" + slcExecution.getUuid());
29 this.executionModulesManager = executionModulesManager;
30 this.slcProcess = slcExecution;
31 processThreadGroup = new ThreadGroup("SLC Process #"
32 + slcExecution.getUuid() + " thread group");
33 }
34
35 public void run() {
36 log.info("\n##\n## Process SLC Execution " + slcProcess + "\n##\n");
37
38 slcProcess.setStatus(SlcExecution.STATUS_RUNNING);
39 dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_SCHEDULED,
40 SlcExecution.STATUS_RUNNING);
41
42 flowsToProcess.addAll(slcProcess.getRealizedFlows());
43
44 while (flowsToProcess.size() > 0) {
45 RealizedFlow flow = flowsToProcess.remove(0);
46 ExecutionThread thread = new ExecutionThread(this, flow);
47 thread.start();
48
49 synchronized (this) {
50 try {
51 wait();
52 } catch (InterruptedException e) {
53 // silent
54 }
55 }
56 }
57
58 if (hadAnError)
59 slcProcess.setStatus(SlcExecution.STATUS_ERROR);
60 else
61 slcProcess.setStatus(SlcExecution.STATUS_FINISHED);
62 dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_RUNNING,
63 slcProcess.getStatus());
64 }
65
66 protected void dispatchUpdateStatus(SlcExecution slcExecution,
67 String oldStatus, String newStatus) {
68 for (Iterator<SlcExecutionNotifier> it = executionModulesManager
69 .getSlcExecutionNotifiers().iterator(); it.hasNext();) {
70 it.next().updateStatus(slcExecution, oldStatus, newStatus);
71 }
72 }
73
74 public void notifyError() {
75 hadAnError = true;
76 }
77
78 public synchronized void flowCompleted() {
79 notifyAll();
80 }
81
82 public SlcExecution getSlcProcess() {
83 return slcProcess;
84 }
85
86 public ThreadGroup getProcessThreadGroup() {
87 return processThreadGroup;
88 }
89
90 public AbstractExecutionModulesManager getExecutionModulesManager() {
91 return executionModulesManager;
92 }
93 }