今回はProcessorを書いてみたいと思います。Processorとは?Binderにメッセージを送り込むのがSource、メッセージを受け取るのがSinkですが、この両方を併せ持ったのがProcessorみたいです。Streamのドキュメントを見ると、ProcessorはSourceとSinkを継承していることになっています。{}http://docs.spring.io/spring-cloud-stream/docs/Brooklyn.SR3/reference/htmlsingle/#__literal_source_literal_literal_sink_literal_and_literal_processor_literalSourceとSinkの間に挟み込むことで、メッセージを変換したり、フィルタリングしたり、集約などをする、といった目的で使われるみたいですね。前置きはこのあたりにして、書いていってみましょう。今回はProcessorが焦点ですが、一応SourceとSinkも作成します。Kafkaを使用し、Source、Processor、Sinkで同じBinderを使用するで、ProcessorはSourceが受け取ったメッセージの一部項目を、ちょっと装飾する感じにしてみましょう。構成は、Mavenのマルチプロジェクトとします。</build>以降のSource、Sink、そしてProcessorのpomは、この親pomを継承して作成するものとします。簡単にSoruceとSinkを作成します。Sourceのpom。</build>Sourceのソースコードは、こちら。}}HTTPボディで受け取った内容を、そのままBinderに送る単純な内容。設定は、こんな感じです。api-source/src/main/resources/application.propertiesspring.cloud.stream.bindings.output.destination=message-sourcespring.cloud.stream.bindings.output.contentType=application/jsonspring.cloud.stream.kafka.binder.brokers=172.17.0.2spring.cloud.stream.kafka.binder.defaultBrokerPort=9092spring.cloud.stream.kafka.binder.zkNodes=172.17.0.2spring.cloud.stream.kafka.binder.defaultZkPort=2181Binderへ向けたTopicの名前は、「message-source」としました。ZooKeeperが稼働しているホストのIPアドレスは、「172.17.0.2」とします。pomの設定は、Sourceと同じです。</build>Sinkのソースコード。}}受け取ったメッセージを、コンソール出力しているだけです。src/main/resources/application.propertiesspring.cloud.stream.bindings.input.destination=message-decoratedspring.cloud.stream.bindings.input.group=group-decoratedspring.cloud.stream.kafka.binder.brokers=172.17.0.2spring.cloud.stream.kafka.binder.defaultBrokerPort=9092spring.cloud.stream.kafka.binder.zkNodes=172.17.0.2spring.cloud.stream.kafka.binder.defaultZkPort=2181server.port=10080こちらでは、Binderからメッセージを受け取る際のTopicの名前を「message-decorated」としました。Sourceでは、「message-source」でした。今回は、この間を埋めるのがProcessorということになります。Kafkaはひとつなのでメッセージがぐるぐると回るので...。Sink側のリッスンポートは、「10080」としました。Sourceは8080を使うことにします。ここまでで、SourceとSinkの作成はおしまい。Processorを作るでは、Processorを作ってみましょう。}こう変換するようにしてみましょう。}pomは、Source、Sinkと同じです。</build>作成したProcessorのソースコードは、こちら。}}まず、@EnableBindingに、Processorクラスを指定します。{実装する処理ですが、今回はメッセージの変換を行うので@Transformerを使用しました。メッセージはStringで来るみたい(?)なので、Jacksonでパースしました...。Integrationのアノテーションのようです。@Transformerは、メッセージを変換できることを意味します。API)@Transformerの他に、@Filterや@Aggregatorなどがあります。Streamが提供している、@Transformer(というかProcessor)のサンプルはこちら。StreamのSourceやSinkと同じような方法で実装すればいいみたいです。すれば、今回のアプリケーションではMap形式でメッセージを受け取れるようになります。キューへの送信は@SendToアノテーションを使用して戻り値としています。