Spring Integration 勉強の備忘録 ~Spring Integration Sampleの理解 2. JMS Gateway~ #Java

目的

Spring Integrationを学ぶために、Spring Projectが提供しているサンプルを理解して、実装し直す + アレンジしてみた。本記事は、その備忘録として。なので色々間違いがあるとは思いますが、その点はご指摘いただければ幸いです。
サンプルは以下のGitに登録してあるもの。
spring-integration-sample

今回はbasicのjmsの内の1つであるGatewayについて。
ちなみにSpring Integrationのバージョンは4.3.10.RELEASEである。

概要

JMSでは3つの機能が実装されており、Mainクラスを実行すると対話形式で1~3のどのモードを試すかを決定するようになっている。今回は1のGatewayについてのみ実装を行った。
1. Gateway
2. Channel Adapter
3. Aggrigation

まだ全然わからないことだらけで、一辺に実装しても理解しにくいので同様の形式で実装するのではなく、個々に実装して理解していくこととする。

実装

Gateway

全体像

全体のフローはたぶん以下のような感じ。（間違っていたらどなたかご指摘ください）
標準入力(stdin) -> stdinToJmsoutChannel -> RequestQueue -> ServiceActivator -> RequestChannel -> ReplyQueue -> ReplyChannel -> 標準出力(stdout)

コンソールから入力した値を、outbound-gatewayがstdin-channel-adapterを介してqueue(RequestQueue)に詰める。次に、inbound-gatewayがRequestQueueから値を取得してServiceActivatorで処理したものをRequestChannelに流す。それをまた、outbound-gatewayが今度はReplyQueueから取得して、ReplyChannelに中継して、最終的に標準出力を行う。

Reference Manualのoutbound-gatewayの説明だと、ReplyQueue(reply-destination属性)は指定しないこともでき、指定しないとTemporaryQueueが使われるとのこと。本サンプルでも未指定のため、TemporaryQueueから値をReplyChannelに流していると思われれる。

bean定義ファイル

本サンプルはcommon.xml, outboundGateway.xml, inboundGateway.xmlの3構成からなっている。

私の実装したものは基本サンプルプロジェクトの使い回しであるが、理解のために不要そうな記述は全部抜いてある。

common.xml

ここではaggregationとかchannelAdapterとか他の機能でも必要な設定を共通設定として定義してたが、今回はまずgatewayに必要なもの設定のみ残す形とした。

Gatewayのアプリに必要であるのは以下の設定。1のconnectionFactoryは内部的にJmsOutboundGatewayクラスがコネクションを作成するために必要。2はstdin-channel-adapterが標準入力をポーリングするために必要っぽい。
そして、requestQueueはoutboundGatewayもinboundGatewayでも使っているため定義が必要である。ここでは実装クラスとして、OSSのActiveMQのQueueを指定している。

他の設定は今回は不要そうだったので削除した。

    <!-- 1 -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <property name="targetConnectionFactory">
            <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                <property name="brokerURL" value="vm://localhost"/>
            </bean>
        </property>
        <property name="sessionCacheSize" value="10"/>
    </bean>

    <!-- 2 -->
    <bean id="requestQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="queue.demo"/>
    </bean>

    <!-- 3 -->
    <integration:poller id="poller" default="true" fixed-delay="1000"/>

outboundGateway.xml

ここも必要最小限の設定だけ残している。test用のprofile情報の設定等もあったが、まずは不要なので消した。
ここでは、JmsOutboundGatewayクラスが、その名の通り処理フローを色々と中継する役割のため、中継するために必要な設定を記述している。
1,3がコンソールへの入力と出力について定義している箇所。2は、request-channel属性から取得した入力情報を、requestQueue(MessageChannel)のsendメソッドに渡している。その戻り値は、今回は省略しているTemporaryQueueに入るので、それを内部的にreply-channel属性のオブジェクトへと渡すという設定になっている模様。

     <!-- 1 -->
    <stream:stdin-channel-adapter id="stdin" channel="stdinToJmsoutChannel"/>
    <channel id="stdinToJmsoutChannel"/>

    <!-- 2 -->
    <jms:outbound-gateway request-channel="stdinToJmsoutChannel"
                          request-destination="requestQueue"
                          reply-channel="jmsReplyChannel"/>

    <!-- 3 -->
    <channel id="jmsReplyChannel" />    
    <stream:stdout-channel-adapter channel="jmsReplyChannel" append-newline="true"/>

inboundGateway.xml

最後はinboundの設定になるが、元々ServiceActivatorの設定があったが、Hello Worldの記事でも実装したので、アノテーションで実装することにした。アノテーションにすることでxmlがスッキリしてやりたいことが分かり易くなった。
（Spring Integrationはxmlが複雑になりがちなので、JavaConfigとかアノテーション主体で実装した方がいいのではと一瞬思った。が、アプリの規模が大きくなると、突如ソースコードが追いにくくなるようにも思う。webシステムよりそこら辺のさじ加減が難しそう。）

本題に戻ると、1はコンポーネントスキャンをしている。これにて、ServiceActivatorを実装しているクラスをbean定義不要となる。次に2はinbound-gatewayタグで内部的な処理フローを定義している。outboundとは逆に、request-destination属性から値を取得し、request-channel属性に値を伝搬させていく。

    <!-- 1 -->
    <context:component-scan base-package="org.ek.sample.jms.gateway"/>

    <!-- 2 -->
    <jms:inbound-gateway id="jmsin"
                         request-destination="requestQueue"
                         request-channel="requestChannel"/>

    <channel id="requestChannel"/>

ServiceActivator

既述の通り、ServiceActivatorはアノテーションを使った。そのために(1) @EnableIntegrationアノテーションと、(2) @MessageEndpoint アノテーションを使っている。
(1)を使わない状態で@ServiceActivatorを使うと、Dispatch先が見つからないという類のエラーが出る。Endpoint系のオブジェクトでアノテーション使うには一緒に付与する必要があるのかなーという簡単な認識だけしておいた。後々はもう少し中身を見たいが、今のところは我慢して先に進むことにする。

(2)は単にコンポーネントスキャン対象とするためにアノテーションである。webで言うところの@Controllerとか@Serviceと一緒である。

(3)はまさにbean定義を置き換えたところである。

@EnableIntegration // (1)
@MessageEndpoint // (2)
public class GatewayEndpoint {

    // (3)
    @ServiceActivator(inputChannel="requestChannel")
    public String upperCase(String input) {
        return "JMS response: " + input.toUpperCase();
    }
}

Mainクラス

Mainクラスのコードは以下の通り。

xmlを配列で定義して、それをApplicationContextのインスタンスとして読み込むだけ。
これだけで、コンソールとアプリとの対話を実現できる。

Main

    private final static String[] configFilesGatewayDemo = {
            "/META-INF/spring/integration/common/common.xml",
            "/META-INF/spring/integration/gateway/inboundGateway.xml",
            "/META-INF/spring/integration/gateway/outboundGateway.xml"
        };

    @SuppressWarnings("resource")
    public static void main(String[] args) {
        new ClassPathXmlApplicationContext(configFilesGatewayDemo,GatewayMain.class);
    }

最後に

今回はinbound-gatewayとかoutbound-gateway等を扱ったが、こいつらはinとoutのやり取りを後ろ側でよしなにやってくれるので結構使い勝手良さそうと思った。あとは、個人的に感覚的にxmlが読めないのでそこの可読性をどう上げるかを考えないと、自分で実装してても後で忘れそうという印象を持った。

コードはコチラ。Jmsサンプルをいじったもの