ACT-1103 adding support for a single message start event

modules/activiti-engine/src/main/java/org/activiti/engine/RuntimeService.java

@@ -29,6 +29,7 @@ import org.activiti.engine.runtime.ProcessInstanceQuery;
  * 
  * @author Tom Baeyens
  * @author Joram Barrez
+ * @author Daniel Meyer
  */
 public interface RuntimeService {
   
@@ -149,6 +150,53 @@ public interface RuntimeService {
    * @throws ActivitiException when no process definition is deployed with the given key. 
    */
   ProcessInstance startProcessInstanceById(String processDefinitionId, String businessKey, Map<String, Object> variables);
+  
+  /**
+   * <p>Signals the process engine that a message is received and starts a new 
+   * {@link ProcessInstance}.</p>
+   * 
+   * <p>Calling this method can have two different outcomes:
+   * <ul>
+   * <li>If the message name is associated with a message start event, a new
+   * process instance is started.</li>
+   * <li>If no subscription to a message with the given name exists, {@link ActivitiException}
+   * is thrown</li>
+   * </ul>
+   * </p>
+   *  
+   * @param messageName
+   *          the 'name' of the message as specified as an attribute on the
+   *          bpmn20 {@code <message name="messageName" />} element.
+   * @param processVariables
+   *          the 'payload' of the message. The variables are added as processes
+   *          variables to the started process instance.
+   * @return the {@link ProcessInstance} object representing the started process instance
+   * 
+   * @throws ActivitiExeception if no subscription to a message with the given name exists
+   * 
+   * @since 5.9
+   */
+  ProcessInstance startProcessInstanceByMessage(String messageName, Map<String, Object> processVariables);
+  
+  /**
+   * See {@link #startProcessInstanceByMessage(String, Map)}. In addition, this method allows 
+   * specifying a business key.
+   *  
+   * @param messageName
+   *          the 'name' of the message as specified as an attribute on the
+   *          bpmn20 {@code <message name="messageName" />} element.
+   * @param businessKey
+   *          the business key which is added to the started process instance 
+   * @param processVariables
+   *          the 'payload' of the message. The variables are added as processes
+   *          variables to the started process instance.
+   * @return the {@link ProcessInstance} object representing the started process instance
+   * 
+   * @throws ActivitiExeception if no subscription to a message with the given name exists
+   * 
+   * @since 5.9
+   */
+  ProcessInstance startProcessInstanceByMessage(String messageName, String businessKey, Map<String, Object> processVariables);
 
   /** Delete an existing runtime process instance.
    * @param processInstanceId id of process instance to delete, cannot be null.
@@ -286,6 +334,5 @@ public interface RuntimeService {
    * @throws ActivitiException if no such processInstance can be found or if the process instance is already in state active.
    */
   void activateProcessInstanceById(String processInstanceId);
-  
-  
+    
 }
\ No newline at end of file

modules/activiti-engine/src/main/java/org/activiti/engine/impl/EventSubscriptionQueryImpl.java

@@ -105,7 +105,7 @@ public class EventSubscriptionQueryImpl
   }
   
   public EventSubscriptionQuery orderByCreated() {
-    return this;    
+    return orderBy(EventSubscriptionQueryProperty.CREATED);
   }
   
   //results //////////////////////////////////////////

modules/activiti-engine/src/main/java/org/activiti/engine/impl/EventSubscriptionQueryProperty.java

+/* Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.activiti.engine.impl;
+
+import org.activiti.engine.query.QueryProperty;
+
+
+/**
+ * @author Daniel Meyer
+ */
+public class EventSubscriptionQueryProperty implements QueryProperty {
+
+  private static final long serialVersionUID = 1L;
+  
+  // properties used in event subscription queries:
+  
+  public final static EventSubscriptionQueryProperty CREATED = new EventSubscriptionQueryProperty("E.CREATED_");
+  
+  ///////////////////////////////////////////////// 
+  
+  private final String propertyName;
+  
+  public EventSubscriptionQueryProperty(String propertyName) {
+    this.propertyName = propertyName;
+  }
+
+  public String getName() {
+    return propertyName;
+  }
+
+}

modules/activiti-engine/src/main/java/org/activiti/engine/impl/RuntimeServiceImpl.java

@@ -28,6 +28,7 @@ import org.activiti.engine.impl.cmd.GetExecutionVariablesCmd;
 import org.activiti.engine.impl.cmd.GetStartFormCmd;
 import org.activiti.engine.impl.cmd.SetExecutionVariablesCmd;
 import org.activiti.engine.impl.cmd.SignalCmd;
+import org.activiti.engine.impl.cmd.StartProcessInstanceByMessageCmd;
 import org.activiti.engine.impl.cmd.StartProcessInstanceCmd;
 import org.activiti.engine.impl.cmd.SuspendProcessInstanceCmd;
 import org.activiti.engine.runtime.EventSubscriptionQuery;
@@ -158,5 +159,13 @@ public class RuntimeServiceImpl extends ServiceImpl implements RuntimeService {
   public void activateProcessInstanceById(String processInstanceId) {
     commandExecutor.execute(new ActivateProcessInstanceCmd(processInstanceId));
   }
+  
+  public ProcessInstance startProcessInstanceByMessage(String messageName, Map<String, Object> processVariables) {
+    return commandExecutor.execute(new StartProcessInstanceByMessageCmd(messageName, null, processVariables));
+  }
+  
+  public ProcessInstance startProcessInstanceByMessage(String messageName, String businessKey, Map<String, Object> processVariables) {
+    return commandExecutor.execute(new StartProcessInstanceByMessageCmd(messageName, businessKey, processVariables));
+  }
    
 }

modules/activiti-engine/src/main/java/org/activiti/engine/impl/bpmn/deployer/BpmnDeployer.java

@@ -20,20 +20,25 @@ import java.util.Map;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
+import org.activiti.engine.ActivitiException;
 import org.activiti.engine.impl.bpmn.diagram.ProcessDiagramGenerator;
 import org.activiti.engine.impl.bpmn.parser.BpmnParse;
 import org.activiti.engine.impl.bpmn.parser.BpmnParser;
+import org.activiti.engine.impl.bpmn.parser.MessageEventDefinition;
 import org.activiti.engine.impl.cfg.IdGenerator;
 import org.activiti.engine.impl.cmd.DeleteJobsCmd;
 import org.activiti.engine.impl.context.Context;
 import org.activiti.engine.impl.db.DbSqlSession;
 import org.activiti.engine.impl.el.ExpressionManager;
+import org.activiti.engine.impl.event.MessageEventHandler;
 import org.activiti.engine.impl.interceptor.CommandContext;
 import org.activiti.engine.impl.jobexecutor.TimerDeclarationImpl;
 import org.activiti.engine.impl.jobexecutor.TimerStartEventJobHandler;
 import org.activiti.engine.impl.persistence.deploy.Deployer;
 import org.activiti.engine.impl.persistence.deploy.DeploymentCache;
 import org.activiti.engine.impl.persistence.entity.DeploymentEntity;
+import org.activiti.engine.impl.persistence.entity.EventSubscriptionEntity;
+import org.activiti.engine.impl.persistence.entity.MessageEventSubscriptionEntity;
 import org.activiti.engine.impl.persistence.entity.ProcessDefinitionEntity;
 import org.activiti.engine.impl.persistence.entity.ProcessDefinitionManager;
 import org.activiti.engine.impl.persistence.entity.ResourceEntity;
@@ -136,6 +141,9 @@ public class BpmnDeployer implements Deployer {
 
         removeObsoleteTimers(processDefinition);
         addTimerDeclarations(processDefinition);
+        
+        removeObsoleteMessageEventSubscriptions(processDefinition);
+        addMessageEventSubscriptions(processDefinition);
 
         dbSqlSession.insert(processDefinition);
         deploymentCache.addProcessDefinition(processDefinition);
@@ -180,6 +188,52 @@ public class BpmnDeployer implements Deployer {
         new DeleteJobsCmd(job.getId()).execute(Context.getCommandContext());
     }
   }
+  
+  protected void removeObsoleteMessageEventSubscriptions(ProcessDefinitionEntity processDefinition) {
+    // delete message event subscriptions:
+    List<EventSubscriptionEntity> subscriptionsToDelete = Context.getCommandContext()
+      .getEventSubscriptionManager()
+      .findEventSubscriptionsByConfiguration(MessageEventHandler.TYPE, processDefinition.getKey());
+    for (EventSubscriptionEntity eventSubscriptionEntity : subscriptionsToDelete) {
+      eventSubscriptionEntity.delete();      
+    }
+  }
+  
+  @SuppressWarnings("unchecked")
+  protected void addMessageEventSubscriptions(ProcessDefinitionEntity processDefinition) {
+    List<MessageEventDefinition> messageEventDefinitions = (List<MessageEventDefinition>) processDefinition.getProperty(BpmnParse.PROPERTYNAME_MESSAGE_EVENT_DEFINITIONS);
+    if(messageEventDefinitions == null) {
+      // this process has no subscriptions
+      return;
+    }
+    for (MessageEventDefinition messageEventDefinition : messageEventDefinitions) {
+      List<EventSubscriptionEntity> subscriptionsForSameMessageName = Context.getCommandContext()
+        .getEventSubscriptionManager()
+        .findEventSubscriptionByName(MessageEventHandler.TYPE, messageEventDefinition.getName());
+      List<MessageEventSubscriptionEntity> cachedSubscriptions = Context.getCommandContext()
+        .getDbSqlSession()
+        .findInCache(MessageEventSubscriptionEntity.class);
+      // also look for subscriptions created in the same command:
+      for (MessageEventSubscriptionEntity cachedSubscription : cachedSubscriptions) {
+        if(messageEventDefinition.getName().equals(cachedSubscription.getEventName())) {
+          subscriptionsForSameMessageName.add(cachedSubscription);
+        }
+      }
+      
+      if(!subscriptionsForSameMessageName.isEmpty()) {
+        throw new ActivitiException("Cannot deploy process definition '" + processDefinition.getDiagramResourceName()
+                + "': there already is a message event subscription for the message with name '" + messageEventDefinition.getName() + "'.");
+      }
+      
+      MessageEventSubscriptionEntity newSubscription = new MessageEventSubscriptionEntity();
+      newSubscription.setEventName(messageEventDefinition.getName());
+      newSubscription.setActivityId(messageEventDefinition.getActivityId());
+      newSubscription.setConfiguration(processDefinition.getKey());
+      
+      newSubscription.insert();      
+    }
+      
+  }
 
 
   /**

modules/activiti-engine/src/main/java/org/activiti/engine/impl/bpmn/parser/BpmnParse.java

@@ -138,6 +138,7 @@ public class BpmnParse extends Parse {
   public static final String PROPERTYNAME_COMPENSATION_HANDLER_ID = "compensationHandler";
   public static final String PROPERTYNAME_IS_FOR_COMPENSATION = "isForCompensation";
   public static final String PROPERTYNAME_ERROR_EVENT_DEFINITIONS = "errorEventDefinitions";
+  public static final String PROPERTYNAME_MESSAGE_EVENT_DEFINITIONS = "messageEventDefinitions";
 
   /** The deployment to which the parsed process definitions will be added. */
   protected DeploymentEntity deployment;
@@ -364,14 +365,20 @@ public class BpmnParse extends Parse {
     for (Element messageElement : rootElement.elements("message")) {
       String id = messageElement.attribute("id");
       String itemRef = this.resolveName(messageElement.attribute("itemRef"));
-
-      if (!this.itemDefinitions.containsKey(itemRef)) {
-        addError(itemRef + " does not exist", messageElement);
-      } else {
+      String name = messageElement.attribute("name");
+      
+      MessageDefinition messageDefinition = new MessageDefinition(id, name);
+      
+      if(itemRef != null) {
         ItemDefinition itemDefinition = this.itemDefinitions.get(itemRef);
-        MessageDefinition message = new MessageDefinition(this.targetNamespace + ":" + id, itemDefinition);
-        this.messages.put(message.getId(), message);
+        if(itemDefinition == null) {
+          addError(itemRef + " does not exist", messageElement);          
+        } else {
+          messageDefinition.setItemDefinition(itemDefinition);
+        }
       }
+      
+      this.messages.put(messageDefinition.getId(), messageDefinition);
     }
   }
   
@@ -683,15 +690,15 @@ public class BpmnParse extends Parse {
    */
   public void parseStartEvents(Element parentElement, ScopeImpl scope) {
     List<Element> startEventElements = parentElement.elements("startEvent");
-    if (startEventElements.size() > 1) {
-      addError("Multiple start events are currently unsupported", parentElement);
-    } else if (startEventElements.size() > 0) {
+    
+    if (startEventElements.size() > 0) {
 
       Element startEventElement = startEventElements.get(0);
     
       ActivityImpl startEventActivity = createActivityOnScope(startEventElement, scope);
 
       if (scope instanceof ProcessDefinitionEntity) {
+        
         ProcessDefinitionEntity processDefinition = (ProcessDefinitionEntity) scope;
         if (processDefinition.getInitial() != null) {
           // in order to support this, the initial should here be replaced with
@@ -718,8 +725,11 @@ public class BpmnParse extends Parse {
         }
         // only allowed for process...
         Element timerEventDefinition = startEventElement.element("timerEventDefinition");
+        Element messageEventDefinition = startEventElement.element("messageEventDefinition");        
         if (timerEventDefinition != null) {
           parseTimerStartEventDefinition(timerEventDefinition, startEventActivity, processDefinition);
+        } else if(messageEventDefinition != null) {
+          parseMessageStartEventDefinition(messageEventDefinition, startEventActivity, processDefinition);
         }
       } else {
         scope.setProperty(PROPERTYNAME_INITIAL, startEventActivity);
@@ -755,6 +765,33 @@ public class BpmnParse extends Parse {
     }
   }
 
+  protected void parseMessageStartEventDefinition(Element messageEventDefinition, ActivityImpl startEventActivity, ProcessDefinitionEntity processDefinition) {
+    String messageRef = messageEventDefinition.attribute("messageRef");
+    if(messageRef == null) {
+      addError("attriute 'messageRef' is required", messageEventDefinition);
+    }
+    MessageDefinition messageDefinition = messages.get(messageRef);
+    if(messageDefinition == null) {
+      addError("Invalid 'messageRef': no message with id '"+messageRef+"' found.", messageEventDefinition);
+    }
+    
+    startEventActivity.setProperty("type", "messageStartEvent");
+    
+    // create message event subscription:
+    MessageEventDefinition subscription = new MessageEventDefinition(messageDefinition.getId(), messageDefinition.getName(), startEventActivity.getId());
+    addMessageEventDefinition(subscription, processDefinition);    
+  }
+
+  @SuppressWarnings("unchecked")
+  protected void addMessageEventDefinition(MessageEventDefinition subscription, ScopeImpl scope) {
+    List<MessageEventDefinition> messageEventDefinitions = (List<MessageEventDefinition>) scope.getProperty(PROPERTYNAME_MESSAGE_EVENT_DEFINITIONS);
+    if(messageEventDefinitions == null) {
+      messageEventDefinitions = new ArrayList<MessageEventDefinition>();
+      scope.setProperty(PROPERTYNAME_MESSAGE_EVENT_DEFINITIONS, messageEventDefinitions);
+    }
+    messageEventDefinitions.add(subscription);
+  }
+
   /**
    * Parses the activities of a certain level in the process (process,
    * subprocess or another scope).

modules/activiti-engine/src/main/java/org/activiti/engine/impl/bpmn/parser/MessageEventDefinition.java

+/* Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.activiti.engine.impl.bpmn.parser;
+
+import java.io.Serializable;
+
+
+/**
+ * A subscription to a message event persisted in the process definition
+ * 
+ * @author Daniel Meyer
+ */
+public class MessageEventDefinition implements Serializable {
+  
+  private static final long serialVersionUID = 1L;
+  
+  protected final String name;
+  protected final String id;
+  protected final String activityId;
+
+  public MessageEventDefinition(String id, String name, String activityId) {
+    this.id = id;
+    this.name = name;
+    this.activityId = activityId;    
+  }
+  
+  public String getName() {
+    return name;
+  }
+  
+  public String getId() {
+    return id;
+  }
+  
+  public String getActivityId() {
+    return activityId;
+  }
+
+}

modules/activiti-engine/src/main/java/org/activiti/engine/impl/bpmn/webservice/MessageDefinition.java

@@ -25,10 +25,12 @@ public class MessageDefinition {
   protected String id;
   
   protected ItemDefinition itemDefinition;
+
+  protected String name;
   
-  public MessageDefinition(String id, ItemDefinition itemDefinition) {
+  public MessageDefinition(String id, String name) {
     this.id = id;
-    this.itemDefinition = itemDefinition;
+    this.name = name;
   }
   
   public MessageInstance createInstance() {
@@ -42,8 +44,20 @@ public class MessageDefinition {
   public StructureDefinition getStructureDefinition() {
     return this.itemDefinition.getStructureDefinition();
   }
+  
+  public void setItemDefinition(ItemDefinition itemDefinition) {
+    this.itemDefinition = itemDefinition;
+  }
 
   public String getId() {
     return this.id;
   }
+  
+  public String getName() {
+    return name;
+  }
+  
+  public void setName(String name) {
+    this.name = name;
+  }
 }

modules/activiti-engine/src/main/java/org/activiti/engine/impl/cmd/StartProcessInstanceByMessageCmd.java

+/* Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.activiti.engine.impl.cmd;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.activiti.engine.ActivitiException;
+import org.activiti.engine.impl.context.Context;
+import org.activiti.engine.impl.interceptor.Command;
+import org.activiti.engine.impl.interceptor.CommandContext;
+import org.activiti.engine.impl.persistence.deploy.DeploymentCache;
+import org.activiti.engine.impl.persistence.entity.ExecutionEntity;
+import org.activiti.engine.impl.persistence.entity.MessageEventSubscriptionEntity;
+import org.activiti.engine.impl.persistence.entity.ProcessDefinitionEntity;
+import org.activiti.engine.impl.pvm.process.ActivityImpl;
+import org.activiti.engine.runtime.ProcessInstance;
+
+
+/**
+ * @author Daniel Meyer
+ */
+public class StartProcessInstanceByMessageCmd implements Command<ProcessInstance> {
+
+  protected final String messageName;
+  protected final String businessKey;
+  protected final Map<String, Object> processVariables;
+
+  public StartProcessInstanceByMessageCmd(String messageName, String businessKey, Map<String, Object> processVariables) {
+    this.messageName = messageName;
+    this.businessKey = businessKey;
+    this.processVariables = processVariables;    
+  }
+
+  public ProcessInstance execute(CommandContext commandContext) {
+    
+    if(messageName == null) {
+      throw new ActivitiException("Cannot start process instance by message: message name is null");
+    }
+    
+    MessageEventSubscriptionEntity messageEventSubscription = commandContext.getEventSubscriptionManager()
+      .findMessageStartEventSubscriptionByName(messageName);
+    
+    if(messageEventSubscription == null) {
+      throw new ActivitiException("Cannot start process instance by message: no subscription to message with name '"+messageName+"' found.");
+    }
+    
+    String processDefinitionKey = messageEventSubscription.getConfiguration();
+    if(processDefinitionKey == null) {
+      throw new ActivitiException("Cannot start process instance by message: subscription to message with name '"+messageName+"' is not a message start event.");
+    }
+        
+    DeploymentCache deploymentCache = Context
+            .getProcessEngineConfiguration()
+            .getDeploymentCache();
+          
+    ProcessDefinitionEntity processDefinition = deploymentCache.findDeployedLatestProcessDefinitionByKey(processDefinitionKey);
+    if (processDefinition == null) {
+        throw new ActivitiException("No process definition found for key '" + processDefinitionKey + "'");
+    }
+  
+    ActivityImpl startActivity = processDefinition.findActivity(messageEventSubscription.getActivityId());
+    ExecutionEntity processInstance = processDefinition.createProcessInstance(businessKey, startActivity);
+
+    if (processVariables != null) {
+      processInstance.setVariables(processVariables);
+    }
+    
+    processInstance.start();
+    
+    return processInstance;
+  }
+
+}

modules/activiti-engine/src/main/java/org/activiti/engine/impl/event/MessageEventHandler.java

+/* Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.activiti.engine.impl.event;
+
+import org.activiti.engine.impl.interceptor.CommandContext;
+import org.activiti.engine.impl.persistence.entity.EventSubscriptionEntity;
+
+
+/**
+ * @author Daniel Meyer
+ */
+public class MessageEventHandler implements EventHandler {
+  
+  public final static String TYPE = "message";
+
+  public String getEventHandlerType() {
+    return TYPE;
+  }
+
+  public void handleEvent(EventSubscriptionEntity eventSubscription, Object payload, CommandContext commandContext) {
+  }
+
+}

modules/activiti-engine/src/main/java/org/activiti/engine/impl/persistence/entity/DeploymentManager.java

@@ -18,9 +18,12 @@ import java.util.List;
 import org.activiti.engine.impl.DeploymentQueryImpl;
 import org.activiti.engine.impl.Page;
 import org.activiti.engine.impl.context.Context;
+import org.activiti.engine.impl.event.MessageEventHandler;
+import org.activiti.engine.impl.jobexecutor.TimerStartEventJobHandler;
 import org.activiti.engine.impl.persistence.AbstractManager;
 import org.activiti.engine.repository.Deployment;
 import org.activiti.engine.repository.ProcessDefinition;
+import org.activiti.engine.runtime.Job;
 
 
 /**
@@ -43,28 +46,54 @@ public class DeploymentManager extends AbstractManager {
   }
   
   public void deleteDeployment(String deploymentId, boolean cascade) {
+    List<ProcessDefinition> processDefinitions = getDbSqlSession()
+            .createProcessDefinitionQuery()
+            .deploymentId(deploymentId)
+            .list();
+
     if (cascade) {
-      List<ProcessDefinition> processDefinitions = getDbSqlSession()
-        .createProcessDefinitionQuery()
-        .deploymentId(deploymentId)
-        .list();
 
+      // delete process instances
       for (ProcessDefinition processDefinition: processDefinitions) {
         String processDefinitionId = processDefinition.getId();
         
         getProcessInstanceManager()
           .deleteProcessInstancesByProcessDefinition(processDefinitionId, "deleted deployment", cascade);
-
-        Context
-          .getProcessEngineConfiguration()
-          .getDeploymentCache()
-          .removeProcessDefinition(processDefinitionId);
+    
       }
     }
     
+    // delete process definitions from db
     getProcessDefinitionManager()
       .deleteProcessDefinitionsByDeploymentId(deploymentId);
     
+    for (ProcessDefinition processDefinition : processDefinitions) {
+      String processDefinitionId = processDefinition.getId();
+      
+      // remove process definitions from cache:
+      Context
+        .getProcessEngineConfiguration()
+        .getDeploymentCache()
+        .removeProcessDefinition(processDefinitionId);
+      
+      // remove timer start events:
+      List<Job> timerStartJobs = Context.getCommandContext()
+        .getJobManager()
+        .findJobsByConfiguration(TimerStartEventJobHandler.TYPE, processDefinition.getKey());
+      for (Job job : timerStartJobs) {
+        ((JobEntity)job).delete();        
+      }
+      
+      // remove message event subscriptions:
+      List<EventSubscriptionEntity> findEventSubscriptionsByConfiguration = Context
+        .getCommandContext()
+        .getEventSubscriptionManager()
+        .findEventSubscriptionsByConfiguration(MessageEventHandler.TYPE, processDefinition.getKey());
+      for (EventSubscriptionEntity eventSubscriptionEntity : findEventSubscriptionsByConfiguration) {
+        eventSubscriptionEntity.delete();        
+      }
+    }
+    
     getResourceManager()
       .deleteResourcesByDeploymentId(deploymentId);
     

modules/activiti-engine/src/main/java/org/activiti/engine/impl/persistence/entity/EventSubscriptionEntity.java

@@ -50,13 +50,15 @@ public abstract class EventSubscriptionEntity implements PersistentObject, Event
   
   /////////////////////////////////////////////
   
-  public EventSubscriptionEntity() { }
+  public EventSubscriptionEntity() { 
+    this.created = ClockUtil.getCurrentTime();
+  }
 
   public EventSubscriptionEntity(ExecutionEntity executionEntity) {
+    this();
     setExecution(executionEntity);
     setActivity(execution.getActivity());
-    this.processInstanceId = executionEntity.getProcessDefinitionId();
-    this.created = ClockUtil.getCurrentTime();
+    this.processInstanceId = executionEntity.getProcessInstanceId();
   }
   
   // processing /////////////////////////////

modules/activiti-engine/src/main/java/org/activiti/engine/impl/persistence/entity/EventSubscriptionManager.java

@@ -22,7 +22,6 @@ import java.util.Set;
 
 import org.activiti.engine.impl.EventSubscriptionQueryImpl;
 import org.activiti.engine.impl.Page;
-import org.activiti.engine.impl.db.PersistentObject;
 import org.activiti.engine.impl.persistence.AbstractManager;
 import org.activiti.engine.runtime.EventSubscription;
 
@@ -111,5 +110,26 @@ public class EventSubscriptionManager extends AbstractManager {
     params.put("activityId", activityId);
     return getDbSqlSession().selectList(query, params);            
   }
+
+  public List<EventSubscriptionEntity> findEventSubscriptionsByConfiguration(String type, String configuration) {
+    final String query = "selectEventSubscriptionsByConfiguration";    
+    Map<String,String> params = new HashMap<String, String>();
+    params.put("eventType", type);
+    params.put("configuration", configuration);
+    return getDbSqlSession().selectList(query, params);            
+  }
+
+  public List<EventSubscriptionEntity> findEventSubscriptionByName(String type, String eventName) {
+    final String query = "selectEventSubscriptionsByName";    
+    Map<String,String> params = new HashMap<String, String>();
+    params.put("eventType", type);
+    params.put("eventName", eventName);    
+    return getDbSqlSession().selectList(query, params);            
+  }
+
+  public MessageEventSubscriptionEntity findMessageStartEventSubscriptionByName(String messageName) {
+    MessageEventSubscriptionEntity entity = (MessageEventSubscriptionEntity) getDbSqlSession().selectOne("selectMessageStartEventSubscriptionByName", messageName);
+    return entity;
+  }
    
 }

modules/activiti-engine/src/main/java/org/activiti/engine/impl/persistence/entity/ExecutionEntity.java

@@ -842,7 +842,7 @@ public class ExecutionEntity extends VariableScopeImpl implements ActivityExecut
               
     for (EventSubscriptionEntity eventSubscription : eventSubscriptions) {
       if (replacedBy!=null) {
-        eventSubscription.setExecutionId(replacedBy.getId());
+        eventSubscription.setExecution(replacedBy);
       } else {
         eventSubscription.delete();
       }        
@@ -850,7 +850,7 @@ public class ExecutionEntity extends VariableScopeImpl implements ActivityExecut
 
     for (CompensateEventSubscriptionEntity compensateEventSubscription : getCompensateEventSubscriptions()) {
       if (replacedBy!=null) {
-        compensateEventSubscription.setExecutionId(replacedBy.getId());
+        compensateEventSubscription.setExecution(replacedBy);
       } else {
         removeCompensateEventSubscription(compensateEventSubscription);
         compensateEventSubscription.delete();

modules/activiti-engine/src/main/java/org/activiti/engine/impl/persistence/entity/MessageEventSubscriptionEntity.java

@@ -13,6 +13,7 @@
 
 package org.activiti.engine.impl.persistence.entity;
 
+
 /**
  * @author Daniel Meyer
  */

modules/activiti-engine/src/main/java/org/activiti/engine/impl/persistence/entity/ProcessDefinitionEntity.java

@@ -26,6 +26,7 @@ import org.activiti.engine.impl.db.PersistentObject;
 import org.activiti.engine.impl.form.StartFormHandler;
 import org.activiti.engine.impl.identity.Authentication;
 import org.activiti.engine.impl.interceptor.CommandContext;
+import org.activiti.engine.impl.pvm.process.ActivityImpl;
 import org.activiti.engine.impl.pvm.process.ProcessDefinitionImpl;
 import org.activiti.engine.impl.pvm.runtime.InterpretableExecution;
 import org.activiti.engine.impl.task.TaskDefinition;
@@ -59,8 +60,14 @@ public class ProcessDefinitionEntity extends ProcessDefinitionImpl implements Pr
     super(null);
   }
   
-  public ExecutionEntity createProcessInstance(String businessKey) {
-	  ExecutionEntity processInstance = (ExecutionEntity) super.createProcessInstance();
+  public ExecutionEntity createProcessInstance(String businessKey, ActivityImpl initial) {
+    ExecutionEntity processInstance = null;
+  
+    if(initial == null) {
+      processInstance = (ExecutionEntity) super.createProcessInstance();
+    }else {
+      processInstance = (ExecutionEntity) super.createProcessInstanceForInitial(initial);
+    }
 
     CommandContext commandContext = Context.getCommandContext();
   
@@ -115,11 +122,15 @@ public class ProcessDefinitionEntity extends ProcessDefinitionImpl implements Pr
 
     return processInstance;
   }
+  public ExecutionEntity createProcessInstance(String businessKey) {
+    return createProcessInstance(businessKey, null);
+  }
 
   public ExecutionEntity createProcessInstance() {
     return createProcessInstance(null);
   }
   
+  
   @Override
   protected InterpretableExecution newProcessInstance() {
     ExecutionEntity processInstance = new ExecutionEntity();

modules/activiti-engine/src/main/java/org/activiti/engine/impl/pvm/process/ProcessDefinitionImpl.java

@@ -62,6 +62,37 @@ public class ProcessDefinitionImpl extends ScopeImpl implements PvmProcessDefini
 
     return processInstance;
   }
+  
+  /** creates a process instance using the provided activity as initial */
+  public PvmProcessInstance createProcessInstanceForInitial(ActivityImpl startActivity) {
+    InterpretableExecution processInstance = newProcessInstance();
+    processInstance.setProcessDefinition(this);
+    processInstance.setProcessInstance(processInstance);
+    processInstance.initialize();
+
+    InterpretableExecution scopeInstance = processInstance;
+    
+    ArrayList<ActivityImpl> initialActivityStack = new ArrayList<ActivityImpl>();
+    ActivityImpl activity = startActivity;
+    while (activity!=null) {
+      initialActivityStack.add(0, activity);
+      activity = activity.getParentActivity();
+    }
+    
+    for (ActivityImpl initialActivity: initialActivityStack) {
+      if (initialActivity.isScope()) {
+        scopeInstance = (InterpretableExecution) scopeInstance.createExecution();
+        scopeInstance.setActivity(initialActivity);
+        if (initialActivity.isScope()) {
+          scopeInstance.initialize();
+        }
+      }
+    }
+    
+    scopeInstance.setActivity(startActivity);
+
+    return processInstance;
+  }
 
   public synchronized List<ActivityImpl> getInitialActivityStack() {
     if (initialActivityStack==null) {

modules/activiti-engine/src/main/resources/org/activiti/db/mapping/entity/EventSubscription.xml

@@ -14,6 +14,10 @@
     delete from ACT_RU_EVENT_SUBSCR where ID_ = #{id}
   </delete>
   
+  <delete id="deleteMessageEventSubscription" parameterType="string">
+    delete from ACT_RU_EVENT_SUBSCR where ID_ = #{id}
+  </delete>
+  
   <!-- RESULTMAP -->
 
   <resultMap id="eventSubscriptionResultMap" type="org.activiti.engine.impl.persistence.entity.EventSubscriptionEntity">
@@ -115,6 +119,29 @@
     	and (ACTIVITY_ID_ = #{activityId})
   </select>
   
+  <select id="selectEventSubscriptionsByConfiguration" resultMap="eventSubscriptionResultMap" parameterType="java.util.Map">
+    select * 
+    from ACT_RU_EVENT_SUBSCR
+    where (EVENT_TYPE_ = #{eventType})
+    	and (CONFIGURATION_ = #{configuration})
+  </select>
+  
+  <select id="selectEventSubscriptionsByName" resultMap="eventSubscriptionResultMap" parameterType="java.util.Map">
+    select * 
+    from ACT_RU_EVENT_SUBSCR
+    where (EVENT_TYPE_ = #{eventType})
+    	and (EVENT_NAME_ = #{eventName})
+  </select>
+  
+   <select id="selectMessageStartEventSubscriptionByName" resultMap="eventSubscriptionResultMap" parameterType="java.lang.String">
+    select * 
+    from ACT_RU_EVENT_SUBSCR
+    where (EVENT_TYPE_ = 'message')
+    	and (EVENT_NAME_ = #{eventName})
+    	and EXECUTION_ID_ is null
+  </select>
+  
+  
   <!-- MESSAGE INSERT -->
   
   <insert id="insertMessageEventSubscription" parameterType="org.activiti.engine.impl.persistence.entity.MessageEventSubscriptionEntity">

modules/activiti-engine/src/test/java/org/activiti/engine/test/bpmn/event/message/MessageStartEventTest.java

+/* Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.activiti.engine.test.bpmn.event.message;
+
+import org.activiti.engine.impl.test.PluggableActivitiTestCase;
+import org.activiti.engine.runtime.ProcessInstance;
+import org.activiti.engine.task.Task;
+import org.activiti.engine.test.Deployment;
+
+
+/**
+ * @author Daniel Meyer
+ */
+public class MessageStartEventTest extends PluggableActivitiTestCase {
+  
+  @Deployment
+  public void testSingleMessageStartEvent() {
+    
+    ProcessInstance processInstance = runtimeService.startProcessInstanceByMessage("newInvoiceMessage", null);
+    
+    assertFalse(processInstance.isEnded());
+    
+    Task task = taskService.createTaskQuery().singleResult();
+    assertNotNull(task);
+    
+    taskService.complete(task.getId());
+            
+  }
+
+}

modules/activiti-engine/src/test/resources/org/activiti/engine/test/bpmn/event/message/MessageStartEventTest.testSingleMessageStartEvent.bpmn20.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<definitions id="definitions" 
+  xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL"
+  xmlns:activiti="http://activiti.org/bpmn"
+  targetNamespace="Examples">
+  
+  <message id="newInvoice" name="newInvoiceMessage" />
+  
+  <process id="singleMessageStartEvent">
+  
+    <startEvent id="theStart" >
+    	<messageEventDefinition messageRef="newInvoice" />
+    </startEvent>
+    
+    <sequenceFlow id="flow1" sourceRef="theStart" targetRef="task" />
+  
+    <userTask id="task" name="Task" />
+    <sequenceFlow id="flow2" sourceRef="task" targetRef="theEnd" />
+    
+    <endEvent id="theEnd" />
+    
+  </process>
+
+</definitions>
\ No newline at end of file