JBPM-4595 - add support for correlation key into jbpm services

jbpm-installer/db/ddl-scripts/db2/db2-jbpm-schema.sql

@@ -243,6 +243,7 @@
         parentProcessInstanceId bigint,
         processId varchar(255),
         processInstanceDescription varchar(255),
+        correlationKey varchar(255),
         processInstanceId bigint not null,
         processName varchar(255),
         processVersion varchar(255),

jbpm-installer/db/ddl-scripts/derby/derby-jbpm-schema.sql

@@ -243,6 +243,7 @@
         parentProcessInstanceId bigint,
         processId varchar(255),
         processInstanceDescription varchar(255),
+        correlationKey varchar(255),
         processInstanceId bigint not null,
         processName varchar(255),
         processVersion varchar(255),

jbpm-installer/db/ddl-scripts/h2/h2-jbpm-schema.sql

@@ -243,6 +243,7 @@
         parentProcessInstanceId bigint,
         processId varchar(255),
         processInstanceDescription varchar(255),
+        correlationKey varchar(255),
         processInstanceId bigint not null,
         processName varchar(255),
         processVersion varchar(255),

jbpm-installer/db/ddl-scripts/hsqldb/hsqldb-jbpm-schema.sql

@@ -243,6 +243,7 @@
         parentProcessInstanceId bigint,
         processId varchar(255),
         processInstanceDescription varchar(255),
+        correlationKey varchar(255),
         processInstanceId bigint not null,
         processName varchar(255),
         processVersion varchar(255),

jbpm-installer/db/ddl-scripts/mysql5/mysql5-jbpm-schema.sql

@@ -243,6 +243,7 @@
         parentProcessInstanceId bigint,
         processId varchar(255),
         processInstanceDescription varchar(255),
+        correlationKey varchar(255),
         processInstanceId bigint not null,
         processName varchar(255),
         processVersion varchar(255),

jbpm-installer/db/ddl-scripts/mysqlinnodb/mysql-innodb-jbpm-schema.sql

@@ -243,6 +243,7 @@
         parentProcessInstanceId bigint,
         processId varchar(255),
         processInstanceDescription varchar(255),
+        correlationKey varchar(255),
         processInstanceId bigint not null,
         processName varchar(255),
         processVersion varchar(255),

jbpm-installer/db/ddl-scripts/oracle/oracle-jbpm-schema.sql

@@ -243,6 +243,7 @@
         parentProcessInstanceId number(19,0),
         processId varchar2(255 char),
         processInstanceDescription varchar2(255 char),
+        correlationKey varchar2(255 char),
         processInstanceId number(19,0) not null,
         processName varchar2(255 char),
         processVersion varchar2(255 char),

jbpm-installer/db/ddl-scripts/postgresql/postgresql-jbpm-schema.sql

@@ -243,6 +243,7 @@
         parentProcessInstanceId int8,
         processId varchar(255),
         processInstanceDescription varchar(255),
+        correlationKey varchar(255),
         processInstanceId int8 not null,
         processName varchar(255),
         processVersion varchar(255),

jbpm-installer/db/ddl-scripts/sqlserver/sqlserver-jbpm-schema.sql

@@ -243,6 +243,7 @@
         parentProcessInstanceId numeric(19,0),
         processId varchar(255),
         processInstanceDescription varchar(255),
+        correlationKey varchar(255),
         processInstanceId numeric(19,0) not null,
         processName varchar(255),
         processVersion varchar(255),

jbpm-installer/db/ddl-scripts/sqlserver2008/sqlserver2008-jbpm-schema.sql

@@ -243,6 +243,7 @@
         parentProcessInstanceId bigint,
         processId varchar(255),
         processInstanceDescription varchar(255),
+        correlationKey varchar(255),
         processInstanceId bigint not null,
         processName varchar(255),
         processVersion varchar(255),

jbpm-installer/db/upgrade-scripts/jbpm-6.2-to-6.3.sql

@@ -0,0 +1,38 @@
+--db2
+
+alter table ProcessInstanceLog add correlationKey varchar(255); 
+
+
+--derby
+alter table ProcessInstanceLog add column correlationKey varchar(255); 
+
+
+-- h2
+
+ALTER TABLE ProcessInstanceLog ADD correlationKey varchar(255);
+
+
+-- hsqldb
+
+ALTER TABLE ProcessInstanceLog ADD COLUMN correlationKey VARCHAR(255);
+
+
+-- mysql
+
+ALTER TABLE ProcessInstanceLog ADD COLUMN correlationKey VARCHAR(255);
+
+
+-- oracle
+
+alter table ProcessInstanceLog add correlationKey varchar2(255 char);
+
+
+-- postgres
+
+ALTER TABLE ProcessInstanceLog ADD COLUMN correlationKey varchar(255);
+
+
+--sql server
+
+ALTER TABLE ProcessInstanceLog ADD correlationKey varchar(255);
+

jbpm-services/jbpm-kie-services/src/main/java/org/jbpm/kie/services/impl/ProcessServiceImpl.java

@@ -48,7 +48,10 @@
 import org.kie.api.runtime.process.ProcessInstance;
 import org.kie.api.runtime.process.WorkItem;
 import org.kie.api.runtime.process.WorkflowProcessInstance;
+import org.kie.internal.process.CorrelationAwareProcessRuntime;
+import org.kie.internal.process.CorrelationKey;
 import org.kie.internal.runtime.manager.InternalRuntimeManager;
+import org.kie.internal.runtime.manager.context.CorrelationKeyContext;
 import org.kie.internal.runtime.manager.context.ProcessInstanceIdContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -97,6 +100,35 @@ public Long startProcess(String deploymentId, String processId, Map<String, Obje
         	disposeRuntimeEngine(manager, engine);
         }
 	}
+
+	@Override
+	public Long startProcess(String deploymentId, String processId, CorrelationKey correlationKey) {
+		return startProcess(deploymentId, processId, correlationKey, new HashMap<String, Object>());
+	}
+
+	@Override
+	public Long startProcess(String deploymentId, String processId, CorrelationKey correlationKey, Map<String, Object> params) {
+		DeployedUnit deployedUnit = deploymentService.getDeployedUnit(deploymentId);
+		if (deployedUnit == null) {
+			throw new DeploymentNotFoundException("No deployments available for " + deploymentId);
+		}		
+		if (!deployedUnit.isActive()) {
+			throw new DeploymentNotFoundException("Deployments " + deploymentId + " is not active");
+		}
+
+		RuntimeManager manager = deployedUnit.getRuntimeManager();
+
+		params = process(params, ((InternalRuntimeManager) manager).getEnvironment().getClassLoader());
+        RuntimeEngine engine = manager.getRuntimeEngine(ProcessInstanceIdContext.get());
+        KieSession ksession = engine.getKieSession();
+        ProcessInstance pi = null;
+        try {
+            pi = ((CorrelationAwareProcessRuntime)ksession).startProcess(processId, correlationKey, params);
+            return pi.getId();
+        } finally {
+        	disposeRuntimeEngine(manager, engine);
+        }
+	}
 
 	@Override
 	public void abortProcessInstance(Long processInstanceId) {
@@ -173,6 +205,26 @@ public ProcessInstance getProcessInstance(Long processInstanceId) {
         	disposeRuntimeEngine(manager, engine);
         }
 	}
+
+	@Override
+	public ProcessInstance getProcessInstance(CorrelationKey key) {
+		ProcessInstanceDesc piDesc = dataService.getProcessInstanceByCorrelationKey(key);
+		if (piDesc == null) {
+			return null;
+		}
+		DeployedUnit deployedUnit = deploymentService.getDeployedUnit(piDesc.getDeploymentId());
+		if (deployedUnit == null) {
+			throw new DeploymentNotFoundException("No deployments available for " + piDesc.getDeploymentId());
+		}
+		RuntimeManager manager = deployedUnit.getRuntimeManager();
+        RuntimeEngine engine = manager.getRuntimeEngine(CorrelationKeyContext.get(key));
+        KieSession ksession = engine.getKieSession();
+        try {
+        	return ((CorrelationAwareProcessRuntime)ksession).getProcessInstance(key);
+        } finally {
+        	disposeRuntimeEngine(manager, engine);
+        }
+	}
 
 	@Override
 	public void setProcessVariable(Long processInstanceId, String variableId, Object value) {

jbpm-services/jbpm-kie-services/src/main/java/org/jbpm/kie/services/impl/RuntimeDataServiceImpl.java

@@ -49,6 +49,7 @@
 import org.kie.api.task.model.Status;
 import org.kie.api.task.model.TaskSummary;
 import org.kie.internal.identity.IdentityProvider;
+import org.kie.internal.process.CorrelationKey;
 import org.kie.internal.query.QueryContext;
 import org.kie.internal.query.QueryFilter;
 import org.kie.internal.task.api.AuditTask;
@@ -425,6 +426,34 @@ public ProcessInstanceDesc getProcessInstanceById(long processId) {
         }
         return null;
    }
+
+	@Override
+	public ProcessInstanceDesc getProcessInstanceByCorrelationKey(CorrelationKey correlationKey) {
+	   	Map<String, Object> params = new HashMap<String, Object>();
+        params.put("correlationKey", correlationKey.toExternalForm());
+        params.put("maxResults", 1);
+
+        List<ProcessInstanceDesc> processInstances = commandService.execute(
+				new QueryNameCommand<List<ProcessInstanceDesc>>("getProcessInstanceByCorrelationKey", 
+                params));
+
+        if (!processInstances.isEmpty()) {
+        	ProcessInstanceDesc desc = processInstances.iterator().next();
+        	List<String> statuses = new ArrayList<String>();
+        	statuses.add(Status.Ready.name());
+        	statuses.add(Status.Reserved.name());
+        	statuses.add(Status.InProgress.name());
+
+        	params = new HashMap<String, Object>();
+            params.put("processInstanceId", desc.getId());
+            params.put("statuses", statuses);
+            List<UserTaskInstanceDesc> tasks = commandService.execute(
+    				new QueryNameCommand<List<UserTaskInstanceDesc>>("getTaskInstancesByProcessInstanceId", params));
+        	((org.jbpm.kie.services.impl.model.ProcessInstanceDesc)desc).setActiveTasks(tasks);
+        	return desc;
+        }
+        return null;
+	}
 
 
     @Override

jbpm-services/jbpm-kie-services/src/main/java/org/jbpm/kie/services/impl/model/ProcessInstanceDesc.java

@@ -36,14 +36,17 @@ public class ProcessInstanceDesc implements org.jbpm.services.api.model.ProcessI
 
     private String processInstanceDescription;
 
+    private String correlationKey;
+
     private List<org.jbpm.services.api.model.UserTaskInstanceDesc> activeTasks;
 
     public ProcessInstanceDesc() {
 
     }
 
     public ProcessInstanceDesc(long id, String processId, String processName, String processVersion, 
-                                int state, String deploymentId, Date dataTimeStamp, String initiator,String processInstanceDescription) {
+                                int state, String deploymentId, Date dataTimeStamp, String initiator,
+                                String processInstanceDescription, String correlationKey) {
         this.id = id;
         this.processId = processId;
         this.processName = processName;
@@ -53,10 +56,11 @@ public ProcessInstanceDesc(long id, String processId, String processName, String
         this.dataTimeStamp = dataTimeStamp;
         this.initiator = initiator;
         this.processInstanceDescription = processInstanceDescription;
+        this.correlationKey = correlationKey==null?"":correlationKey;
     }
 
     public ProcessInstanceDesc(long id, String processId, String processName, String processVersion, 
-                               int state, String deploymentId, Date dataTimeStamp, String initiator) {
+                               int state, String deploymentId, Date dataTimeStamp, String initiator, String correlationKey) {
        this.id = id;
        this.processId = processId;
        this.processName = processName;
@@ -65,6 +69,7 @@ public ProcessInstanceDesc(long id, String processId, String processName, String
        this.deploymentId = deploymentId;
        this.dataTimeStamp = dataTimeStamp;
        this.initiator = initiator;
+       this.correlationKey = correlationKey==null?"":correlationKey;
    }
 
     public String getProcessId() {
@@ -93,7 +98,7 @@ public Date getDataTimeStamp() {
 
     @Override
     public String toString() {
-        return "ProcessInstanceDesc{" + "id=" + id + ", processId=" + processId + ", processName=" + processName + ", processVersion=" + processVersion + ", state=" + state + ", deploymentId=" + deploymentId + ", initiator=" + initiator + ", dataTimeStamp=" + dataTimeStamp + '}';
+        return "ProcessInstanceDesc{" + "id=" + id + ", processId=" + processId + ", processName=" + processName + ", processVersion=" + processVersion + "correlationKey=" + correlationKey + ", state=" + state + ", deploymentId=" + deploymentId + ", initiator=" + initiator + ", dataTimeStamp=" + dataTimeStamp + '}';
     }
 
     public String getProcessVersion() {
@@ -132,4 +137,12 @@ public void setActiveTasks(List<org.jbpm.services.api.model.UserTaskInstanceDesc
 		this.activeTasks = activeTasks;
 	}
 
+	public String getCorrelationKey() {
+		return correlationKey;
+	}
+
+	public void setCorrelationKey(String correlationKey) {
+		this.correlationKey = correlationKey;
+	}
+
 }