import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
@Override
public void execute(ActivityExecution context) throws Exception {
  Order order = orderRepository.getOrder( //
      (String)context.getVariable("orderId")); 
  String traceId = context.getProcessBusinessKey();

  // publish
  messageSender.send(new Message<FetchGoodsCommandPayload>( //
          "FetchGoodsCommand", //
          traceId, //
          new FetchGoodsCommandPayload() //
            .setRefId(order.getId()) //
            .setItems(order.getItems())));
  
  addMessageSubscription(context, "GoodsFetchedEvent");
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
public void execute(ActivityExecution execution) throws Exception {
  PvmTransition more = execution.getActivity().findOutgoingTransition("more");
  PvmTransition done = execution.getActivity().findOutgoingTransition("done");

  Integer value = (Integer) execution.getVariable(variableName);

  if (value==null) {
    execution.setVariable(variableName, from);
    execution.leaveActivityViaTransition(more);

  } else {
    value = value+1;

    if (value<to) {
      execution.setVariable(variableName, value);
      execution.leaveActivityViaTransition(more);

    } else {
      execution.leaveActivityViaTransition(done);
    }
  }
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
protected void signalCompensationDone(ActivityExecution execution) {
  // default behavior is to join compensating executions and propagate the signal if all executions have compensated

  // only wait for non-event-scope executions cause a compensation event subprocess consume the compensation event and
  // do not have to compensate embedded subprocesses (which are still non-event-scope executions)

  if(((PvmExecutionImpl) execution).getNonEventScopeExecutions().isEmpty()) {
    if(execution.getParent() != null) {
      ActivityExecution parent = execution.getParent();
      execution.remove();
      parent.signal(SIGNAL_COMPENSATION_DONE, null);
    }
  } else {
    ((ExecutionEntity)execution).forceUpdate();
  }

}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
@Override
public void onParseMigratingInstance(MigratingInstanceParseContext parseContext, MigratingActivityInstance migratingInstance) {
  ExecutionEntity scopeExecution = migratingInstance.resolveRepresentativeExecution();

  List<ActivityExecution> concurrentInActiveExecutions =
      scopeExecution.findInactiveChildExecutions(getInnerActivity((ActivityImpl) migratingInstance.getSourceScope()));

  // variables on ended inner instance executions need not be migrated anywhere
  // since they are also not represented in the tree of migrating instances, we remove
  // them from the parse context here to avoid a validation exception
  for (ActivityExecution execution : concurrentInActiveExecutions) {
    for (VariableInstanceEntity variable : ((ExecutionEntity) execution).getVariablesInternal()) {
      parseContext.consume(variable);
    }
  }

}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
/**
 * Takes an {@link ActivityExecution} and an {@link Callable} and wraps
 * the call to the Callable with the proper error propagation. This method
 * also makes sure that exceptions not caught by following activities in the
 * process will be thrown and not propagated.
 *
 * @param execution
 * @param toExecute
 * @throws Exception
 */
protected void executeWithErrorPropagation(ActivityExecution execution, Callable<Void> toExecute) throws Exception {
  String activityInstanceId = execution.getActivityInstanceId();
  try {
    toExecute.call();
  } catch (Exception ex) {
    if (activityInstanceId.equals(execution.getActivityInstanceId())) {

      try {
        propagateException(execution, ex);
      }
      catch (ErrorPropagationException e) {
        LOG.errorPropagationException(activityInstanceId, e.getCause());
        // re-throw the original exception so that it is logged
        // and set as cause of the failure
        throw ex;
      }

    }
    else {
      throw ex;
    }
  }
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
@Override
public ActivityExecution createInnerInstance(ActivityExecution scopeExecution) {

  if (hasLoopVariable(scopeExecution, NUMBER_OF_ACTIVE_INSTANCES) && getLoopVariable(scopeExecution, NUMBER_OF_ACTIVE_INSTANCES) > 0) {
    throw LOG.unsupportedConcurrencyException(scopeExecution.toString(), this.getClass().getSimpleName());
  }
  else {
    int nrOfInstances = getLoopVariable(scopeExecution, NUMBER_OF_INSTANCES);

    setLoopVariable(scopeExecution, LOOP_COUNTER, nrOfInstances);
    setLoopVariable(scopeExecution, NUMBER_OF_INSTANCES, nrOfInstances + 1);
    setLoopVariable(scopeExecution, NUMBER_OF_ACTIVE_INSTANCES, 1);
  }

  return scopeExecution;
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
private static String getSubscriptionActivityId(ActivityExecution execution, String activityRef) {
  ActivityImpl activityToCompensate = ((ExecutionEntity) execution).getProcessDefinition().findActivity(activityRef);

  if (activityToCompensate.isMultiInstance()) {

    ActivityImpl flowScope = (ActivityImpl) activityToCompensate.getFlowScope();
    return flowScope.getActivityId();
  } else {

    ActivityImpl compensationHandler = activityToCompensate.findCompensationHandler();
    if (compensationHandler != null) {
      return compensationHandler.getActivityId();
    } else {
      // if activityRef = subprocess and subprocess has no compensation handler
      return activityRef;
    }
  }
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
@Override
protected void createInstances(ActivityExecution execution, int nrOfInstances) throws Exception {
  PvmActivity innerActivity = getInnerActivity(execution.getActivity());

  // initialize the scope and create the desired number of child executions
  prepareScopeExecution(execution, nrOfInstances);

  List<ActivityExecution> concurrentExecutions = new ArrayList<ActivityExecution>();
  for (int i = 0; i < nrOfInstances; i++) {
    concurrentExecutions.add(createConcurrentExecution(execution));
  }

  // start the concurrent child executions
  // start executions in reverse order (order will be reversed again in command context with the effect that they are
  // actually be started in correct order :) )
  for (int i = (nrOfInstances - 1); i >= 0; i--) {
    ActivityExecution activityExecution = concurrentExecutions.get(i);
    performInstance(activityExecution, innerActivity, i);
  }
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
private void addActivityInstanceId(ActivityExecution execution, Map<String, List<ActivityInstance>> instanceMap) {

    String actId = execution.getActivity().getId();
    String actInstanceId = execution.getActivityInstanceId();
    String parentActInstanceId = execution.getParentActivityInstanceId();
    String executionId = String.valueOf(System.identityHashCode(execution));

    // add to instance map
    List<ActivityInstance> instancesForThisAct = instanceMap.get(actId);
    if(instancesForThisAct == null) {
      instancesForThisAct = new ArrayList<ActivityInstance>();
      instanceMap.put(actId, instancesForThisAct);
    }
    ActivityInstance activityInstance = new ActivityInstance(executionId, actInstanceId, parentActInstanceId, execution.isCompleteScope());
    instancesForThisAct.add(activityInstance);
  }
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
public void complete(ActivityExecution execution) {

    ActivityExecution outgoingExecution = execution.getParent().createExecution();
    outgoingExecution.setConcurrent(false);
    outgoingExecution.setActivity(execution.getActivity());

    // eventscope execution
    execution.setConcurrent(false);
    execution.setActive(false);
    ((PvmExecutionImpl)execution).setEventScope(true);

    List<PvmTransition> outgoingTransitions = execution.getActivity().getOutgoingTransitions();
    if(outgoingTransitions.isEmpty()) {
      outgoingExecution.end(true);
    }else {
      outgoingExecution.leaveActivityViaTransitions(outgoingTransitions, Collections.EMPTY_LIST);
    }
  }
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
@Override
public void execute(ActivityExecution execution) throws Exception {

  EnsureUtil
  .ensureNotNull("Could not find cancel boundary event for cancel end event " + execution.getActivity(), "cancelBoundaryEvent", cancelBoundaryEvent);

  List<EventSubscriptionEntity> compensateEventSubscriptions =
      CompensationUtil.collectCompensateEventSubscriptionsForScope(execution);

  if(compensateEventSubscriptions.isEmpty()) {
    leave(execution);
  }
  else {
    CompensationUtil.throwCompensationEvent(compensateEventSubscriptions, execution, false);
  }

}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
public void execute(ActivityExecution execution) throws Exception {

    // Join
    PvmActivity activity = execution.getActivity();
    List<PvmTransition> outgoingTransitions = execution.getActivity().getOutgoingTransitions();

    execution.inactivate();
    lockConcurrentRoot(execution);

    List<ActivityExecution> joinedExecutions = execution.findInactiveConcurrentExecutions(activity);
    int nbrOfExecutionsToJoin = execution.getActivity().getIncomingTransitions().size();
    int nbrOfExecutionsJoined = joinedExecutions.size();

    if (nbrOfExecutionsJoined==nbrOfExecutionsToJoin) {

      // Fork
      LOG.activityActivation(activity.getId(), nbrOfExecutionsJoined, nbrOfExecutionsToJoin);
      execution.leaveActivityViaTransitions(outgoingTransitions, joinedExecutions);

    } else {
      LOG.noActivityActivation(activity.getId(), nbrOfExecutionsJoined, nbrOfExecutionsToJoin);
    }
  }
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
@Override
public void passOutputVariables(final ActivityExecution execution, final VariableScope subInstance) {
  // only data. no control flow available on this execution.
  VariableMap variables = filterVariables(getOutputVariables(subInstance));
  VariableMap localVariables = getOutputVariablesLocal(subInstance);

  execution.setVariables(variables);
  execution.setVariablesLocal(localVariables);

  final DelegateVariableMapping varMapping = resolveDelegation(execution);
  if (varMapping != null) {
    invokeVarMappingDelegation(new DelegateInvocation(execution, null) {
      @Override
      protected void invoke() throws Exception {
        varMapping.mapOutputVariables(execution, subInstance);
      }
    });
  }
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
public List<ActivityExecution> initializeScope(ActivityExecution scopeExecution, int nrOfInstances) {
  if (nrOfInstances > 1) {
    LOG.unsupportedConcurrencyException(scopeExecution.toString(), this.getClass().getSimpleName());
  }

  List<ActivityExecution> executions = new ArrayList<ActivityExecution>();

  prepareScope(scopeExecution, nrOfInstances);
  setLoopVariable(scopeExecution, NUMBER_OF_ACTIVE_INSTANCES, nrOfInstances);

  if (nrOfInstances > 0) {
    setLoopVariable(scopeExecution, LOOP_COUNTER, 0);
    executions.add(scopeExecution);
  }

  return executions;
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
protected boolean activatesGateway(ActivityExecution execution, PvmActivity gatewayActivity) {
  int numExecutionsGuaranteedToActivate = gatewayActivity.getIncomingTransitions().size();
  ActivityExecution scopeExecution = execution.isScope() ? execution : execution.getParent();

  List<ActivityExecution> executionsAtGateway = execution.findInactiveConcurrentExecutions(gatewayActivity);

  if (executionsAtGateway.size() >= numExecutionsGuaranteedToActivate) {
    return true;
  }
  else {
    Collection<ActivityExecution> executionsNotAtGateway = getLeafExecutions(scopeExecution);
    executionsNotAtGateway.removeAll(executionsAtGateway);

    for (ActivityExecution executionNotAtGateway : executionsNotAtGateway) {
      if (canReachActivity(executionNotAtGateway, gatewayActivity)) {
        return false;
      }
    }

    // if no more token may arrive, then activate
    return true;
  }

}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
@Override
public void signal(ActivityExecution execution, String signalName, Object signalData) {
    log.info("Received signal {} with data {} for execution ID {}", signalName, signalData, execution.getId());
    execution.setVariable("echoResponse", signalData);
    // leave the service task activity:
    leave(execution);
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
protected void addMessageSubscription(final ActivityExecution context, String eventName) {
  ExecutionEntity executionEntity = (ExecutionEntity)context;
  EventSubscriptionEntity eventSubscriptionEntity = new EventSubscriptionEntity(executionEntity, EventType.MESSAGE);
  eventSubscriptionEntity.setEventName(eventName);
  eventSubscriptionEntity.setActivity(executionEntity.getActivity());
  eventSubscriptionEntity.insert();
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
@Override
public void execute(ActivityExecution execution) throws Exception {
  Order order = OrderRepository.instance.getOrder((String)execution.getVariable("orderId")); 
  String transactionId = (String)execution.getVariable("transactionId");    

  // create event payload and send message
  JsonObjectBuilder payload = eventProducer.createPayloadJson("Command", "RetrievePayment", transactionId)
      .add("refId", order.getId()) //
      .add("reason", "CustomerOrder") //
      .add("amount", order.getTotalSum());
  eventProducer.send(payload);

  addMessageSubscription(execution, "PaymentReceived");
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
@Override
public void execute(ActivityExecution execution) throws Exception {
  OrderEventProducer eventProducer = new OrderEventProducer();
  
  Order order = orderRepository.getOrder((String)execution.getVariable("orderId"));
  String transactionId = (String)execution.getVariable("transactionId");
  
  addMessageSubscription(execution, "GoodsFetched");    
  
  eventProducer.publishCommandFetchGoods(transactionId, order);   
}
 

import org.camunda.bpm.engine.impl.pvm.delegate.ActivityExecution; //导入依赖的package包/类
@Override
public void execute(ActivityExecution execution) throws Exception {
  OrderEventProducer eventProducer = new OrderEventProducer();
  
  Order order = orderRepository.getOrder((String)execution.getVariable("orderId"));    
  String pickId = (String)execution.getVariable("pickId");
  String transactionId = (String)execution.getVariable("transactionId");

  addMessageSubscription(execution, "GoodsShipped");

  eventProducer.publishCommandShipGoods(transactionId, order, pickId);    
}