JBoss Fuseを使い倒す その3：デザインパターン詳細編

集約戦略の実装

メッセージの集約戦略を実装するには、org.apache.camel.processor.aggregate.AggregationStrategyインターフェースを実装するクラスを作成し、その唯一のメソッドであるaggregate(Exchange, Exchange)メソッドを定義します。

以下のクラスは、個々のメッセージの中身（Body）を単純にリストにして集約する戦略の実装例です。

import org.apache.camel.processor.aggregate.AggregationStrategy;

public class ListAggregationStrategy implements AggregationStrategy {

    @Override
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        if (oldExchange == null) {
            List<String> body = new ArrayList<>();
            body.add(newExchange.getIn().getBody(String.class));
            newExchange.getIn().setBody(body);
            return newExchange;
        }

        List<String> body = oldExchange.getIn().getBody(List.class);
        body.add(newExchange.getIn().getBody(String.class));
        return oldExchange;
    }

}

aggregateメソッド引数のoldExchangeが集約中のメッセージ、newExchangeが新たに届いたメッセージを表します。新しい相関IDを持つ最初のメッセージが届いたときはoldExchangeがnullになっているので、oldExchangeがnullかどうかで処理を条件分岐します。メソッドの戻り値には、集約後のメッセージを返します。

複数のパターンを組み合わせた高度なルーティング

CamelのEIPサポートでは、これまで見てきたRecipient ListやAggregatorのようにDSL上に単一のノードを置くだけで実装が終わってしまうパターンもあれば、Composed Message ProcessorやScatter-Gatherのように本来、複数のパターンを組み合わせて実現すべきより高次のパターンもあります。

最後の締めくくりに、そのような高次のパターンの実装例を1つ取り上げます。

Scatter-Gatherパターン

Scatter-Gatherパターンのアイデアは、これまでに登場した2つのパターンRecipient ListとAggregatorでそれぞれ実現した、メッセージのブロードキャストと集約を組み合わせるというものです。メッセージのブロードキャストと集約を組み合わせると、1つのメッセージを複数の外部サービスに同時に並列処理させ、それぞれの結果を最後にまとめて集計結果を得るといったような、高度なルーティングを実現できます。

Recipient Listパターンのところで議論したように、EIPでメッセージのブロードキャストを実現するにはPublish-Subscribe ChannelとRecipient Listの2通りの方法があり、したがってScatter-Gatherの実現方法も以下の2通りが考えられます。

1. Publish-Subscribe Channel + Aggregator
2. Recipient List + Aggregator

この記事では、これまでに登場したRecipient ListとAggregatorを組み合わせる方法を取り上げます。Scatter-Gatherを実装するには、メッセージをブロードキャストするルート（Scatterルート）と、それを集約するルート（Gatherルート）の2つのCamelルートが必要です。それぞれRecipientListノードとAggregateノードを使います。

図7：Scatter-Gatherパターンのブロードキャスト側（Scatter）ルート実装例

図8：Scatter-Gather パターンの集約側（Gather）ルート実装例

RecipientListとAggregateの各ノードに設定するプロパティは、それぞれRecipient ListパターンとAggregatorパターンのところで説明した内容がそのまま当てはまります。

Scatter-Gather パターン全体の動作は、次の通りです。まず、第一のScatterルートでRecipientListノードから複数のエンドポイントへメッセージが送られます。メッセージを送られた先の個々のエンドポイントは、それがSOAPやRESTのWebサービスであっても、JMSメッセージコンシューマとなるJava EEアプリケーションでも、あるいは別のCamelルートであっても構いません。そのエンドポイントが、受信したメッセージに対して何らかの処理を実行し、その結果が今度はGatherルートのエンドポイント（図8の「seda:gather/in」）へ入力メッセージとして送り返されます。Scatterルートからメッセージをブロードキャストされたどのエンドポイントも、最終的にGatherルートの一点へメッセージを集めるところがポイントです。最後に、GatherルートはAggregateノードに設定されたメッセージ集約のポリシー（Aggregatorパターンの説明を参照）に基づき、メッセージを1つに集約します。

XML DSLによる定義は、以下のようになります。

<route id="EIP-Scatter">
  <from uri="direct:scatter/in" />
  ...
  <recipientList>
    <header>addresses</header>
  </recipientList>
  ...
</route>

<route id="EIP-Gather">
  <from uri="seda:gather/in" />
  ...
  <aggregate strategyRef="aggregationStrategy">
    <correlationExpression>
      <header>aggregateId</header>
    </correlationExpression>
    <completionTimeout>
      <constant>1000</constant>
    </completionTimeout>
    ...
    <to uri="direct:gather/out" />
  </aggregate>
</route>

おわりに

本記事では4つのルーティングパターンだけを取り上げましたが、それだけでもいかに簡単に、かつ保守しやすい形でEIPを実装できるのか、そのJBoss Fuseのパワーを実感いただけたかと思います。実際JBoss Fuseは、60以上あるEIPのほとんどをカバーしています。繰り返しになりますが、それらのEIPを適材適所で使い分けられることが、システム間統合ソリューションの構築を成功させる上で大変重要になります。

その他のEIPの実装方法については、以下のドキュメントを参照してください。
Apache Camel Development Guide - Chapter 3. Introducing Enterprise Integration Patterns

本記事では複数のEIPを一度に取り上げた都合上、これまでの記事のように、1つ1つのパターンについて、ステップ・バイ・ステップでサンプルプロジェクトの作成はしませんでした。その代わりに、紹介した4つのパターンを含んだサンプルプロジェクトをGitHub上に公開いたしました。

https://github.com/tadayosi/thinkit-fuse-eip

以下のように、プロジェクトをローカルにクローンした後、Mavenのtestコマンドを実行することでサンプルコードの動作を確認できます。

$ git clone https://github.com/tadayosi/thinkit-fuse-eip.git
$ cd thinkit-fuse-eip/
$ mvn test

また、クローンしたプロジェクトをJBDSにインポートして閲覧することも可能です。プロジェクトをJBDS上にMavenプロジェクトとしてインポートし（［Fileメニュー］→［Import］→［Maven］→［Existing Maven Projects］）、その後インポートしたプロジェクトを右クリックして［Enable Fuse Camel Nature］を実行することで、Camelルートの閲覧が可能になります。