jBPM 4.1／workflowpattern／Discriminator

jBPM では、Discriminator パターンが直接サポートされていないので、ちょっと試作してみた。 Discriminator は Join の一種で、パラレルに分岐した実行経路のうちで、最初に 合流点に到達したものをすぐに後ろに通して、後から来たものは無視するワークフローパターンで、一言で言うと「早い者勝ち」といった感じ。 これを org.jbpm.jpdl.internal.activity.JoinActivity を参考にして、<custom> アクティビティで実装してみる。

　　┌→ activityA →→┐
　　│　　　　　　　　 ↓
○→┼→ activityB → discriminator→activityD→ activityE
　　│　　　　　　　　 ↑
　　└→ activityC →→┘

上ののようなフローを定義して activityA を 実行すると、フローの先端が discriminator を通過して activityD に到達する。これが、ただの <join> だと、activityB,C が実行されるまで activityD には行かない。 続いてactivityB を実行すると、discriminatorに届いたところで経路が終わる（エラーにはならない）。ここで、discriminator を使わず、単に activityDに activityA,B,Cをつないでいるだけだと、activityD 上で AからのものとBからのものとで、実行経路が重複する事になる。 以下のようなコードでやってみた。 ※試作なので Session.lock() とか、プロセス定義の正当性チェックは、ここでは省略した。

package test.trial;

import static org.jbpm.api.Execution.STATE_ACTIVE_ROOT;
import java.util.List;
import java.util.Map;
import org.jbpm.api.activity.ActivityExecution;
import org.jbpm.api.activity.ExternalActivityBehaviour;
import org.jbpm.pvm.internal.model.Activity;
import org.jbpm.pvm.internal.model.ExecutionImpl;
import org.jbpm.pvm.internal.model.Transition;

public class Discriminator implements ExternalActivityBehaviour {
   private static final long serialVersionUID = 1L;
   private static final String IS_SUCCEEDING = "__IS_SUCCEEDING__";

   @Override
   public void signal(ActivityExecution execution, String signalName,
         Map<String, ?> parameters) throws Exception {
      assert false;
   }

   @Override
   public void execute(ActivityExecution e) throws Exception {
      assert e instanceof ExecutionImpl; 
      execute((ExecutionImpl)e);
   }

   private void execute(ExecutionImpl exec) throws Exception {
      if (Boolean.TRUE.equals(exec.getVariable(IS_SUCCEEDING))) {
         exec.end();
         return;
      }
      exec.waitForSignal();

      ExecutionImpl root = exec.getParent();
      Activity activity = exec.getActivity();

      markSucceedingExecution(root, activity);

      root.setState(STATE_ACTIVE_ROOT);
      exec.setActivity(activity, root);

      Transition transition = activity.getDefaultOutgoingTransition();
      root.take(transition);
   }

   private void markSucceedingExecution(
         ExecutionImpl root, Activity activity) {
      for (ExecutionImpl concurrent : root.getExecutions()) {
         if (concurrent.getActivity() == activity) continue;
         concurrent.setVariable(IS_SUCCEEDING, true);
      }
   }
}

開始直後、activityA 実行直後、activityB 実行直後、activityD 実行直後の各段階で、それぞれのアクティビティ上の Execution の様子を調べてみたが、想定している部分についてはほぼ期待通り。（JUnit4 と mysql コンソールを使って、<task> と <state> で同じ事を試行。） ただし、上記コードだけだと、プロセス全体の終了（<end>到達）後にも activityC が未終了のまま残ってしまうらしいので、何か工夫が必要かも。 まとめると、本番で使うには以下の追加作業が必要になるか

• 排他制御
• プロセス定義のチェック
• 取り残されたアクティビティをどうするかの決めと実装

これを応用したら、N-out-of-M Join (Partial Join)パターンとか、BPMNの Complex Gatewayなんかも、jBPM で 実装できそうな気がする。