TCP および UDP のサポート

Spring Integration は、インターネットプロトコルを介してメッセージを送受信するためのチャネルアダプターを提供します。UDP（ユーザーデータグラムプロトコル）と TCP（伝送制御プロトコル）アダプターの両方が提供されます。各アダプターは、基礎となるプロトコルを介した一方向通信を提供します。さらに、Spring Integration はシンプルな受信および送信 TCP ゲートウェイを提供します。これらは、双方向通信が必要なときに使用されます。

この依存関係をプロジェクトに含める必要があります。

Maven
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-ip</artifactId>
    <version>5.3.4.RELEASE</version>
</dependency>
Gradle
compile "org.springframework.integration:spring-integration-ip:5.3.4.RELEASE"

導入

各 UDP 受信および送信チャネルアダプターの 2 つのフレーバーが提供されます。

  • UnicastSendingMessageHandler は、データグラムパケットを単一の宛先に送信します。

  • UnicastReceivingChannelAdapter は、受信データグラムパケットを受信します。

  • MulticastSendingMessageHandler は、データグラムパケットをマルチキャストアドレスに送信（ブロードキャスト）します。

  • MulticastReceivingChannelAdapter は、マルチキャストアドレスに参加することにより、受信データグラムパケットを受信します。

TCP 受信および送信チャネルアダプターが提供されます。

  • TcpSendingMessageHandler は、TCP を介してメッセージを送信します。

  • TcpReceivingChannelAdapter は、TCP 経由でメッセージを受信します。

受信 TCP ゲートウェイが提供されます。これにより、単純なリクエスト / レスポンス処理が可能になります。ゲートウェイは任意の数の接続をサポートできますが、各接続はシリアルでしか処理できません。ソケットから読み取るスレッドは、再度読み取る前にレスポンスを待機し、送信します。接続ファクトリが単一使用接続用に構成されている場合、接続はソケットがタイムアウトした後に閉じられます。

送信 TCP ゲートウェイが提供されます。これにより、単純なリクエスト / レスポンス処理が可能になります。関連付けられた接続ファクトリが使い捨て接続用に構成されている場合、新しいリクエストごとに新しい接続がすぐに作成されます。それ以外の場合、接続が使用中の場合、レスポンスが受信されるか、タイムアウトまたは I/O エラーが発生するまで、呼び出しスレッドは接続をブロックします。

TCP および UDP 受信チャネルアダプターと TCP 受信ゲートウェイは、error-channel 属性をサポートしています。これにより、GatewayProxyFactoryBean を入力してくださいで説明したものと同じ基本機能が提供されます。

UDP アダプター

このセクションでは、UDP アダプターを構成および使用する方法について説明します。

送信 UDP アダプター (XML 構成)

次の例では、UDP 送信チャネルアダプターを構成します。

<int-ip:udp-outbound-channel-adapter id="udpOut"
    host="somehost"
    port="11111"
    multicast="false"
    socket-customizer="udpCustomizer"
    channel="exampleChannel"/>
multicast を true に設定する場合、ホスト属性にマルチキャストアドレスも指定する必要があります。

UDP は効率的ですが、信頼性の低いプロトコルです。Spring Integration は、信頼性を向上させるために、check-length と acknowledge の 2 つの属性を追加します。check-length が true に設定されている場合、アダプターはメッセージデータの前に長さフィールド（ネットワークバイト順で 4 バイト）を付けます。これにより、受信側は受信したパケットの長さを確認できます。受信システムが、パケットを格納するには短すぎるバッファを使用する場合、パケットは切り捨てられる可能性があります。length ヘッダーは、これを検出するメカニズムを提供します。

バージョン 4.3 以降、port を 0 に設定できます。この場合、オペレーティングシステムがポートを選択します。選択されたポートは、アダプターが開始され、isListening() が true を返した後に getPort() を呼び出すことで検出できます。

バージョン 5.3.3 以降では、SocketCustomizer Bean を追加して、作成後に DatagramSocket を変更できます（たとえば、setTrafficClass(0x10) を呼び出します）。

次の例は、データグラムパケットに長さチェックを追加する送信チャネルアダプターを示しています。

<int-ip:udp-outbound-channel-adapter id="udpOut"
    host="somehost"
    port="11111"
    multicast="false"
    check-length="true"
    channel="exampleChannel"/>
パケットの受信者は、実際のデータに先行する長さを期待するように構成する必要もあります。Spring Integration UDP 受信チャネルアダプターの場合、check-length 属性を設定します。

2 番目の信頼性の向上により、アプリケーションレベルの確認応答プロトコルを使用できます。受信者は、指定された時間内に送信者に確認応答を送信する必要があります。

次の例は、データグラムパケットに長さチェックを追加し、確認応答を待機する送信チャネルアダプターを示しています。

<int-ip:udp-outbound-channel-adapter id="udpOut"
    host="somehost"
    port="11111"
    multicast="false"
    check-length="true"
    acknowledge="true"
    ack-host="thishost"
    ack-port="22222"
    ack-timeout="10000"
    channel="exampleChannel"/>
acknowledge を true に設定すると、パケットの受信者は、確認応答データ（ホストおよびポート）を含むパケットに追加されたヘッダーを解釈できることを意味します。ほとんどの場合、受信者は Spring Integration 受信チャネルアダプターです。
マルチキャストが真である場合、追加の属性（min-acks-for-success）は、ack-timeout 内で受信する必要がある確認応答の数を指定します。

バージョン 4.3 以降、ackPort を 0 に設定できます。この場合、オペレーティングシステムがポートを選択します。

送信 UDP アダプター (Java 構成)

次の例は、Java で送信 UDP アダプターを構成する方法を示しています。

@Bean
@ServiceActivator(inputChannel = "udpOut")
public UnicastSendingMessageHandler handler() {
    return new UnicastSendingMessageHandler("localhost", 11111);
}

（またはマルチキャストの場合は MulticastSendingChannelAdapter）。

送信 UDP アダプター (Java DSL 設定)

次の例は、Java DSL を使用して送信 UDP アダプターを構成する方法を示しています。

@Bean
public IntegrationFlow udpOutFlow() {
    return f -> f.handle(Udp.outboundAdapter("localhost", 1234)
                    .configureSocket(socket -> socket.setTrafficClass(0x10)))
                .get();
}

受信 UDP アダプター (XML 構成)

次の例は、基本的なユニキャスト受信 udp チャネルアダプターを構成する方法を示しています。

<int-ip:udp-inbound-channel-adapter id="udpReceiver"
    channel="udpOutChannel"
    port="11111"
    receive-buffer-size="500"
    multicast="false"
    socket-customizer="udpCustomizer"
    check-length="true"/>

次の例は、基本的なマルチキャスト受信 udp チャネルアダプターを構成する方法を示しています。

<int-ip:udp-inbound-channel-adapter id="udpReceiver"
    channel="udpOutChannel"
    port="11111"
    receive-buffer-size="500"
    multicast="true"
    multicast-address="225.6.7.8"
    check-length="true"/>

デフォルトでは、メッセージヘッダーで使用するために IP アドレスをホスト名に変換するために、受信パケットで逆 DNS ルックアップが行われます。DNS が構成されていない環境では、これにより遅延が発生する可能性があります。lookup-host 属性を false に設定することにより、このデフォルトの動作をオーバーライドできます。

バージョン 5.3.3 以降では、SocketCustomizer Bean を追加して、作成後に DatagramSocket を変更できます。これは、受信ソケットと送信用に作成されたソケットに対して呼び出されます。

受信 UDP アダプター (Java 構成)

次の例は、Java で受信 UDP アダプターを構成する方法を示しています。

@Bean
public UnicastReceivingChannelAdapter udpIn() {
    UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(11111);
    adapter.setOutputChannelName("udpChannel");
    return adapter;
}

次の例は、Java DSL で受信 UDP アダプターを構成する方法を示しています。

受信 UDP アダプター (Java DSL 設定)

@Bean
public IntegrationFlow udpIn() {
    return IntegrationFlows.from(Udp.inboundAdapter(11111))
            .channel("udpChannel")
            .get();
}

サーバーリスニングイベント

バージョン 5.0.2 以降、受信アダプターが開始されてリッスンを開始すると、UdpServerListeningEvent が発行されます。これは、ポート 0 でリッスンするようにアダプターが構成されている場合、つまりオペレーティングシステムがポートを選択する場合に役立ちます。ソケットに接続する他のプロセスを開始する前に待機する必要がある場合は、isListening() のポーリングの代わりに使用することもできます。

高度な送信設定

<int-ip:udp-outbound-channel-adapter> （UnicastSendingMessageHandler）には destination-expression および socket-expression オプションがあります。

ハードコードされた host-port ペアのランタイム代替として destination-expression を使用して、requestMessage に対する発信データグラムパケットの宛先アドレスを決定できます（評価コンテキストのルートオブジェクトを使用）。式は、URI、URI スタイルの String （RFC-2396 [IETF] (英語) を参照）、または SocketAddress に評価される必要があります。この式に受信 IpHeaders.PACKET_ADDRESS ヘッダーを使用することもできます。フレームワークでは、UnicastReceivingChannelAdapter でデータグラムを受信し、メッセージに変換すると、DatagramPacketMessageMapper がこのヘッダーにデータを入力します。ヘッダー値は、受信データグラムの DatagramPacket.getSocketAddress() の結果とまったく同じです。

socket-expression を使用すると、送信チャネルアダプターは（たとえば）受信チャネルアダプターソケットを使用して、受信したのと同じポートを介してデータグラムを送信できます。アプリケーションが UDP サーバーとして機能し、クライアントがネットワークアドレス変換（NAT）の背後で動作するシナリオで役立ちます。この式は DatagramSocket に評価される必要があります。requestMessage は、評価コンテキストのルートオブジェクトとして使用されます。socket-expression パラメーターを multicast および acknowledge パラメーターと一緒に使用することはできません。次の例は、大文字に変換し、ソケットを使用するトランスフォーマーで UDP 受信チャネルアダプターを構成する方法を示しています。

<int-ip:udp-inbound-channel-adapter id="inbound" port="0" channel="in" />

<int:channel id="in" />

<int:transformer expression="new String(payload).toUpperCase()"
                       input-channel="in" output-channel="out"/>

<int:channel id="out" />

<int-ip:udp-outbound-channel-adapter id="outbound"
                        socket-expression="@inbound.socket"
                        destination-expression="headers['ip_packetAddress']"
                        channel="out" />

次の例は、Java DSL を使用した同等の構成を示しています。

@Bean
public IntegrationFlow udpEchoUpcaseServer() {
    return IntegrationFlows.from(Udp.inboundAdapter(11111).id("udpIn"))
            .<byte[], String>transform(p -> new String(p).toUpperCase())
            .handle(Udp.outboundAdapter("headers['ip_packetAddress']")
                    .socketExpression("@udpIn.socket"))
            .get();
}

TCP 接続ファクトリ

概要

TCP の場合、基礎となる接続の構成は、接続ファクトリを使用して提供されます。クライアント接続ファクトリとサーバー接続ファクトリの 2 種類の接続ファクトリが提供されます。クライアント接続ファクトリは、発信接続を確立します。サーバー接続ファクトリは、受信接続をリッスンします。

送信チャネルアダプターはクライアント接続ファクトリを使用しますが、クライアント接続ファクトリへの参照を受信チャネルアダプターに提供することもできます。そのアダプターは、送信アダプターによって作成された接続で受信される受信メッセージを受信します。

受信チャネルアダプターまたはゲートウェイは、サーバー接続ファクトリを使用します。（実際、接続ファクトリは接続ファクトリなしでは機能できません）。サーバー接続ファクトリへの参照を送信アダプターに提供することもできます。その後、そのアダプターを使用して、同じ接続で受信メッセージに返信を送信できます。

返信メッセージは、接続ファクトリによって元のメッセージに挿入された ip_connectionId ヘッダーが返信に含まれている場合にのみ、接続にルーティングされます。
これは、受信アダプターと送信アダプター間で接続ファクトリを共有するときに実行されるメッセージ相関の範囲です。このような共有により、TCP を介した非同期の双方向通信が可能になります。デフォルトでは、ペイロード情報のみが TCP を使用して転送されます。メッセージの関連付けは、アグリゲーターまたはその他のエンドポイントなどのダウンストリームコンポーネントによって実行する必要があります。選択されたヘッダーの転送のサポートは、バージョン 3.0 で導入されました。詳細については、TCP メッセージ相関を参照してください。

接続ファクトリへの参照を、各型の最大 1 つのアダプターに提供できます。

Spring Integration は、java.net.Socket および java.nio.channel.SocketChannel を使用する接続ファクトリを提供します。

次の例は、java.net.Socket 接続を使用する単純なサーバー接続ファクトリを示しています。

<int-ip:tcp-connection-factory id="server"
    type="server"
    port="1234"/>

次の例は、java.nio.channel.SocketChannel 接続を使用する単純なサーバー接続ファクトリを示しています。

<int-ip:tcp-connection-factory id="server"
    type="server"
    port="1234"
    using-nio="true"/>
Spring Integration バージョン 4.2 以降、サーバーが（ポートを 0 に設定することにより）ランダムポートでリッスンするように構成されている場合、getPort() を使用して OS が選択した実際のポートを取得できます。また、getServerSocketAddress() を使用すると、完全な SocketAddress を取得できます。詳細については、TcpServerConnectionFactory インターフェースの Javadoc を参照してください。
<int-ip:tcp-connection-factory id="client"
    type="client"
    host="localhost"
    port="1234"
    single-use="true"
    so-timeout="10000"/>

次の例は、java.net.Socket 接続を使用し、各メッセージに新しい接続を作成するクライアント接続ファクトリを示しています。

<int-ip:tcp-connection-factory id="client"
    type="client"
    host="localhost"
    port="1234"
    single-use="true"
    so-timeout="10000"
    using-nio=true/>

バージョン 5.2 以降、クライアント接続ファクトリは、秒単位で指定されるプロパティ connectTimeout をサポートします。デフォルトは 60 です。

メッセージ境界 (シリアライザーとデシリアライザー)

TCP はストリーミングプロトコルです。これは、受信者がデータを個別のメッセージに区分できるように、TCP で転送されるデータに何らかの構造を提供する必要があることを意味します。接続ファクトリは、シリアライザーとデシリアライザーを使用して、メッセージペイロードと TCP を介して送信されるビットの間で変換するように構成されます。これは、受信メッセージと送信メッセージにそれぞれデシリアライザーとシリアライザーを提供することで実現されます。Spring Integration は、多くの標準シリアライザーとデシリアライザーを提供します。

ByteArrayCrlfSerializer* は、バイト配列をバイトストリームに変換し、その後にキャリッジリターンおよびラインフィード文字（\r\n）が続きます。これはデフォルトのシリアライザー（およびデシリアライザー）であり、クライアントとして telnet で（たとえば）使用できます。

ByteArraySingleTerminatorSerializer* は、バイト配列を、単一の終了文字が続くバイトのストリームに変換します（デフォルトは 0x00 です）。

ByteArrayLfSerializer* は、バイト配列をバイトのストリームに変換し、その後に単一の改行文字（0x0a）が続きます。

ByteArrayStxEtxSerializer* は、バイト配列を、STX（0x02）が先行し、ETX（0x03）が後続するバイトのストリームに変換します。

ByteArrayLengthHeaderSerializer は、バイト配列を、ネットワークバイト順のバイナリ長（ビッグエンディアン）が前に付くバイトのストリームに変換します。これは、終了文字シーケンスを探すためにすべてのバイトを解析する必要がないため、効率的なデシリアライザーです。バイナリデータを含むペイロードにも使用できます。上記のシリアライザーは、ペイロード内のテキストのみをサポートします。長さヘッダーのデフォルトサイズは 4 バイト（整数）で、最大（2 ^ 31-1）バイトのメッセージを許可します。ただし、length ヘッダーは、255 バイトまでのメッセージの場合は 1 バイト（符号なし）、（2 ^ 16-1）バイトまでのメッセージの場合は符号なしショート（2 バイト）にすることができます。ヘッダーに他の形式が必要な場合は、ByteArrayLengthHeaderSerializer をサブクラス化し、readHeader メソッドと writeHeader メソッドの実装を提供できます。絶対最大データサイズは（2 ^ 31-1）バイトです。バージョン 5.2 以降、ヘッダー値には、ペイロードに加えてヘッダーの長さを含めることができます。inclusive プロパティを設定して、そのメカニズムを有効にします（プロデューサーとコンシューマーで同じに設定する必要があります）。

ByteArrayRawSerializer* は、バイト配列をバイトのストリームに変換し、追加のメッセージ境界データを追加しません。このシリアライザ（およびデシリアライザ）を使用すると、メッセージの終わりは、クライアントが規則的にソケットを閉じることで示されます。このシリアライザーを使用すると、クライアントがソケットを閉じるかタイムアウトが発生するまで、メッセージの受信がハングします。タイムアウトはメッセージになりません。このシリアライザーが使用され、クライアントが Spring Integration アプリケーションである場合、クライアントは single-use="true" で構成された接続ファクトリを使用する必要があります。これを行うと、アダプターはメッセージの送信後にソケットを閉じます。シリアライザーは、それ自体では接続を閉じません。このシリアライザーは、チャネルアダプター（ゲートウェイではなく）で使用される接続ファクトリでのみ使用する必要があり、接続ファクトリは、受信または送信アダプターのいずれかで使用する必要があります。このセクションで後述する ByteArrayElasticRawDeserializer も参照してください。ただし、バージョン 5.2 以降、送信ゲートウェイには新しいプロパティ closeStreamAfterSend があります。これにより、応答を受信するために接続を開いたまま、EOF がサーバーに通知されるため、生のシリアライザー / デシリアライザーを使用できます。

バージョン 4.2.2 より前、ノンブロッキング I/O（NIO）を使用する場合、このシリアライザーは（読み取り中の）タイムアウトをファイルの終わりとして扱い、これまでに読み取られたデータはメッセージとして出力されました。これは信頼性が低く、メッセージの区切りには使用しないでください。現在では、このような条件を例外として扱います。この方法で使用することはほとんどありませんが、treatTimeoutAsEndOfMessage コンストラクター引数を true に設定することにより、以前の動作を復元できます。

これらはそれぞれ AbstractByteArraySerializer のサブクラスであり、org.springframework.core.serializer.Serializer と org.springframework.core.serializer.Deserializer の両方を実装しています。後方互換性のために、シリアライゼーションに AbstractByteArraySerializer のサブクラスを使用する接続も、最初にバイト配列に変換される String を受け入れます。これらのシリアライザーとデシリアライザーはそれぞれ、対応する形式を含む入力ストリームをバイト配列ペイロードに変換します。

動作が不適切なクライアント（構成されたシリアライザーのプロトコルに準拠していないクライアント）によるメモリの枯渇を回避するために、これらのシリアライザーは最大メッセージサイズを課します。受信メッセージがこのサイズを超えると、例外がスローされます。デフォルトの最大メッセージサイズは 2048 バイトです。maxMessageSize プロパティを設定することにより、これを増やすことができます。デフォルトのシリアライザーまたはデシリアライザーを使用して最大メッセージサイズを増やしたい場合、maxMessageSize プロパティセットを使用して明示的な Bean として最大メッセージサイズを宣言し、その Bean を使用するように接続ファクトリを構成する必要があります。

このセクションで前述の * マークが付いたクラスは、中間バッファーを使用し、デコードされたデータを正しいサイズの最終バッファーにコピーします。バージョン 4.3 から、poolSize プロパティを設定してこれらのバッファを設定できます。これらのバッファは、デフォルトの動作である各メッセージに割り当てられて破棄されるのではなく、再利用されます。プロパティを負の値に設定すると、境界のないプールが作成されます。プールが制限されている場合は、poolWaitTimeout プロパティ（ミリ秒単位）を設定することもできます。その後、使用可能なバッファーがなくなると例外がスローされます。デフォルトは無限です。このような例外により、ソケットが閉じられます。

カスタムデシリアライザーで同じメカニズムを使用する場合は、（スーパークラス AbstractByteArraySerializer の代わりに） AbstractPooledBufferByteArraySerializer を継承し、deserialize() の代わりに doDeserialize() を実装できます。バッファは自動的にプールに返されます。AbstractPooledBufferByteArraySerializer は、便利なユーティリティメソッド copyToSizedArray() も提供します。

バージョン 5.0 は ByteArrayElasticRawDeserializer を追加しました。これは、上記の ByteArrayRawSerializer のデシリアライザー側に似ていますが、maxMessageSize を設定する必要がない点が異なります。内部的には、必要に応じてバッファーを拡張できる ByteArrayOutputStream を使用します。クライアントは、メッセージの終わりを通知するために、適切な方法でソケットを閉じる必要があります。

このデシリアライザは、ピアが信頼されている場合にのみ使用してください。メモリ不足の状態により、DoS 接続の影響を受けやすくなります。

MapJsonSerializer は Jackson ObjectMapper を使用して、Map と JSON の間の変換を行います。このシリアライザーを MessageConvertingTcpMessageMapper および MapMessageConverter と組み合わせて使用して、選択したヘッダーとペイロードを JSON で転送できます。

Jackson ObjectMapper は、ストリーム内のメッセージを区別できません。MapJsonSerializer は、メッセージの境界を処理するために別のシリアライザーまたはデシリアライザーに委譲する必要があります。デフォルトでは、ByteArrayLfSerializer が使用され、その結果、メッセージは <json><LF> の形式でワイヤ上に表示されますが、他のメッセージを代わりに使用するように構成できます。(次の例は、その方法を示しています。)

最終的な標準シリアライザーは org.springframework.core.serializer.DefaultSerializer です。これを使用して、Java シリアライゼーションでシリアライズ可能オブジェクトを変換できます。org.springframework.core.serializer.DefaultDeserializer は、シリアライズ可能なオブジェクトを含むストリームの受信デシリアライゼーションのために提供されています。

デフォルトのシリアライザーおよびデシリアライザー（ByteArrayCrLfSerializer）を使用したくない場合は、接続ファクトリで serializer および deserializer 属性を設定する必要があります。次の例は、その方法を示しています。

<bean id="javaSerializer"
      class="org.springframework.core.serializer.DefaultSerializer" />
<bean id="javaDeserializer"
      class="org.springframework.core.serializer.DefaultDeserializer" />

<int-ip:tcp-connection-factory id="server"
    type="server"
    port="1234"
    deserializer="javaDeserializer"
    serializer="javaSerializer"/>

java.net.Socket 接続を使用し、ワイヤー上で Java 直列化を使用するサーバー接続ファクトリ。

接続ファクトリで使用可能な属性の詳細については、このセクションの最後にあるリファレンスを参照してください。

デフォルトでは、メッセージヘッダーで使用するために IP アドレスをホスト名に変換するために、受信パケットで逆 DNS ルックアップが行われます。DNS が構成されていない環境では、これにより接続が遅延する可能性があります。lookup-host 属性を false に設定することにより、このデフォルトの動作をオーバーライドできます。

ソケットおよびソケットファクトリの属性を変更することもできます。詳細については、SSL/TLS サポートを参照してください。そこに記載されているように、SSL が使用されているかどうかにかかわらず、このような変更は可能です。

カスタムシリアライザーとデシリアライザー

データが標準のデシリアライザのいずれかでサポートされている形式でない場合、独自のデシリアライザを実装できます。カスタムシリアライザーを実装することもできます。

カスタムシリアライザーとデシリアライザーのペアを実装するには、org.springframework.core.serializer.Deserializer および org.springframework.core.serializer.Serializer インターフェースを実装します。

デシリアライザは、メッセージ間で閉じた入力ストリームを検出すると、SoftEndOfStreamException をスローする必要があります。これは、クローズが「正常」だったことを示すフレームワークへのシグナルです。メッセージのデコード中にストリームが閉じられた場合、代わりに他の例外がスローされます。

バージョン 5.2 から、SoftEndOfStreamException は IOException を継承する代わりに RuntimeException になりました。

TCP キャッシングクライアント接続ファクトリ

前述のように、TCP ソケットは「使い捨て」（1 つのリクエストまたはレスポンス）または共有にできます。ソケットは一度に 1 つのリクエストまたはレスポンスしか処理できないため、大容量環境の送信ゲートウェイでは共有ソケットはうまく機能しません。

パフォーマンスを向上させるために、ゲートウェイの代わりに共同チャネルアダプターを使用できますが、これにはアプリケーションレベルのメッセージ相関が必要です。詳細については、TCP メッセージ相関を参照してください。

Spring Integration 2.2 は、共有ソケットのプールを使用するキャッシングクライアント接続ファクトリを導入しました。これにより、ゲートウェイは共有接続のプールを使用して複数の同時リクエストを処理できます。

TCP フェールオーバークライアント接続ファクトリ

1 つ以上の他のサーバーへのフェイルオーバーをサポートする TCP 接続ファクトリを構成できます。メッセージを送信する場合、ファクトリは、メッセージが送信されるか接続が見つからなくなるまで、構成されているすべてのファクトリを反復処理します。最初に、構成済みリストの最初のファクトリが使用されます。その後接続が失敗すると、次のファクトリが現在のファクトリになります。次の例は、フェールオーバークライアント接続ファクトリを構成する方法を示しています。

<bean id="failCF" class="o.s.i.ip.tcp.connection.FailoverClientConnectionFactory">
    <constructor-arg>
        <list>
            <ref bean="clientFactory1"/>
            <ref bean="clientFactory2"/>
        </list>
    </constructor-arg>
</bean>
フェイルオーバー接続ファクトリを使用する場合、singleUse プロパティは、ファクトリ自体と、使用するように構成されているファクトリのリストの間で一貫している必要があります。

接続ファクトリには、共有接続（singleUse=false）で使用した場合のフェイルバックに関連する 2 つのプロパティがあります。

  • refreshSharedInterval

  • closeOnRefresh

上記の構成に基づいて、次のシナリオを検討してください。clientFactory1 は接続を確立できないが、clientFactory2 は接続を確立できるとします。refreshSharedInterval が通過した後に failCF getConnection() メソッドが呼び出されると、clientFactory1 を使用して接続を再試行します。成功すると、clientFactory2 への接続が閉じられます。closeOnRefresh が false の場合、「古い」接続は開いたままになり、最初のファクトリがもう一度失敗した場合に将来再利用される可能性があります。

refreshSharedInterval を設定して、その時間が経過した後、最初のファクトリとの再接続のみを試みるようにします。現在の接続が失敗したときに最初のファクトリにのみフェールバックする場合は、Long.MAX_VALUE （デフォルト）に設定します。

リフレッシュによって実際に新しい接続が作成された後、「古い」接続を閉じるように closeOnRefresh を設定します。

これらのプロパティは、デリゲートファクトリのいずれかが CachingClientConnectionFactory である場合は適用されません。これは、接続キャッシュがそこで処理されるためです。その場合、接続を取得するために接続ファクトリのリストが常に参照されます。

バージョン 5.3 以降、これらはデフォルトで Long.MAX_VALUE および true になるため、ファクトリは現在の接続が失敗した場合にのみフェールバックを試みます。以前のバージョンのデフォルトの動作に戻すには、0 および false に設定します。

接続をテストも参照してください。

TCP スレッドアフィニティ接続ファクトリ

Spring Integration バージョン 5.0 は、この接続ファクトリを導入しました。接続は呼び出しスレッドにバインドされ、その接続はスレッドがメッセージを送信するたびに再利用されます。これは、接続が（サーバーまたはネットワークによって）閉じられるか、スレッドが releaseConnection() メソッドを呼び出すまで続きます。接続自体は、別のクライアントファクトリ実装によって提供されます。これは、各スレッドが接続を取得できるように、非共有（単一使用）接続を提供するように構成する必要があります。

次の例は、TCP スレッドアフィニティ接続ファクトリを構成する方法を示しています。

@Bean
public TcpNetClientConnectionFactory cf() {
    TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory("localhost",
            Integer.parseInt(System.getProperty(PORT)));
    cf.setSingleUse(true);
    return cf;
}

@Bean
public ThreadAffinityClientConnectionFactory tacf() {
    return new ThreadAffinityClientConnectionFactory(cf());
}

@Bean
@ServiceActivator(inputChannel = "out")
public TcpOutboundGateway outGate() {
    TcpOutboundGateway outGate = new TcpOutboundGateway();
    outGate.setConnectionFactory(tacf());
    outGate.setReplyChannelName("toString");
    return outGate;
}

接続をテスト

一部のシナリオでは、接続が最初に開かれたときに、ある種のヘルスチェックリクエストを送信すると便利な場合があります。そのようなシナリオの 1 つは、TCP フェールオーバークライアント接続ファクトリを使用していて、選択したサーバーが接続を開くことを許可したが、接続が正常ではないと報告した場合にフェイルオーバーできるようにする場合です。

この機能をサポートするには、クライアント接続ファクトリに connectionTest を追加します。

/**
 * Set a {@link Predicate} that will be invoked to test a new connection; return true
 * to accept the connection, false the reject.
 * @param connectionTest the predicate.
 * @since 5.3
 */
public void setConnectionTest(@Nullable Predicate<TcpConnectionSupport> connectionTest) {
    this.connectionTest = connectionTest;
}

接続をテストするには、テスト内の接続に一時リスナーを接続します。テストが失敗した場合、接続は閉じられ、例外がスローされます。TCP フェールオーバークライアント接続ファクトリと共に使用すると、次のサーバーの試行がトリガーされます。

サーバーからの最初の応答のみがテストリスナーに送られます。

次の例では、PING を送信するときにサーバーが PONG と応答した場合、サーバーは正常であると見なされます。

Message<String> ping = new GenericMessage<>("PING");
byte[] pong = "PONG".getBytes();
clientFactory.setConnectionTest(conn -> {
    CountDownLatch latch = new CountDownLatch(1);
    AtomicBoolean result = new AtomicBoolean();
    conn.registerTestListener(msg -> {
        if (Arrays.equals(pong, (byte[]) msg.getPayload())) {
            result.set(true);
        }
        latch.countDown();
        return false;
    });
    conn.send(ping);
    try {
        latch.await(10, TimeUnit.SECONDS);
    }
    catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return result.get();
});

TCP 接続インターセプター

TcpConnectionInterceptorFactoryChain への参照を使用して接続ファクトリを構成できます。インターセプターを使用して、ネゴシエーション、セキュリティ、その他のオプションなどの動作を接続に追加できます。現在、フレームワークによってインターセプターは提供されていませんが、例についてはソースリポジトリの InterceptedSharedConnectionTests  [GitHub] (英語) を参照してください。

テストケースで使用される HelloWorldInterceptor は、次のように機能します。

インターセプターは、最初にクライアント接続ファクトリで構成されます。最初のメッセージがインターセプトされた接続を介して送信されると、インターセプターは接続を介して "Hello" を送信し、"world!" を受信することを期待します。それが発生すると、ネゴシエーションが完了し、元のメッセージが送信されます。同じ接続を使用するメッセージは、追加のネゴシエーションなしで送信されます。

サーバー接続ファクトリで構成されている場合、インターセプターは最初のメッセージが "Hello" である必要があり、そうであれば "world!" を返します。それ以外の場合は、接続を閉じる例外をスローします。

すべての TcpConnection メソッドがインターセプトされます。インターセプターインスタンスは、インターセプターファクトリによって接続ごとに作成されます。インターセプターがステートフルの場合、ファクトリは接続ごとに新しいインスタンスを作成する必要があります。状態がない場合、同じインターセプターが各接続をラップできます。インターセプターファクトリは、インターセプターファクトリチェーンの構成に追加されます。チェーンは、interceptor-factory 属性を設定することで接続ファクトリに提供できます。インターセプターは TcpConnectionInterceptorSupport を継承する必要があります。ファクトリは TcpConnectionInterceptorFactory インターフェースを実装する必要があります。TcpConnectionInterceptorSupport にはパススルーメソッドがあります。このクラスを継承することにより、インターセプトするメソッドのみを実装する必要があります。

次の例は、接続インターセプターファクトリチェーンを構成する方法を示しています。

<bean id="helloWorldInterceptorFactory"
    class="o.s.i.ip.tcp.connection.TcpConnectionInterceptorFactoryChain">
    <property name="interceptors">
        <array>
            <bean class="o.s.i.ip.tcp.connection.HelloWorldInterceptorFactory"/>
        </array>
    </property>
</bean>

<int-ip:tcp-connection-factory id="server"
    type="server"
    port="12345"
    using-nio="true"
    single-use="true"
    interceptor-factory-chain="helloWorldInterceptorFactory"/>

<int-ip:tcp-connection-factory id="client"
    type="client"
    host="localhost"
    port="12345"
    single-use="true"
    so-timeout="100000"
    using-nio="true"
    interceptor-factory-chain="helloWorldInterceptorFactory"/>

TCP 接続イベント

バージョン 3.0 以降、TcpConnection インスタンスへの変更は TcpConnectionEvent インスタンスによって報告されます。TcpConnectionEvent は ApplicationEvent のサブクラスであるため、ApplicationContext で定義されている任意の ApplicationListener （イベント受信チャネルアダプターなど）で受信できます。

TcpConnectionEvents には次のプロパティがあります。

  • connectionId: 接続にデータを送信するためにメッセージヘッダーで使用できる接続識別子。

  • connectionFactoryName: 接続が属する接続ファクトリの Bean 名。

  • throwable: Throwable （TcpConnectionExceptionEvent イベントのみ）。

  • source: TcpConnection これを使用して、たとえば、getHostAddress() でリモート IP アドレスを判別できます（キャストが必要）。

さらに、バージョン 4.0 以降、TCP 接続ファクトリで説明した標準デシリアライザーは、データストリームのデコード中に問題が発生すると TcpDeserializationExceptionEvent インスタンスを発行するようになりました。これらのイベントには、例外、構築中のバッファー、例外が発生した時点でのバッファーへのオフセット（使用可能な場合）が含まれます。アプリケーションは、通常の ApplicationListener または ApplicationEventListeningMessageProducer （Spring アプリケーションイベントの受信を参照）を使用してこれらのイベントをキャプチャーし、問題の分析を可能にします。

バージョン 4.0.7 および 4.1.3 以降、サーバーソケットで予期しない例外が発生すると（サーバーソケットの使用中に BindException など）、TcpConnectionServerExceptionEvent インスタンスが公開されます。これらのイベントには、接続ファクトリと原因への参照があります。

バージョン 4.2 以降、TcpConnectionFailedCorrelationEvent インスタンスは、ip_connectionId ヘッダーが無効であるため接続にルーティングできないメッセージをエンドポイント（受信ゲートウェイまたはコラボレーション送信チャネルアダプター）が受信するたびに公開されます。送信ゲートウェイは、遅い応答が受信されたとき（送信者スレッドがタイムアウトしたとき）にもこのイベントを発行します。このイベントには、失敗したメッセージを含む cause プロパティの例外と接続 ID が含まれます。

バージョン 4.3 以降、サーバー接続ファクトリの開始時に TcpConnectionServerListeningEvent が発行されます。これは、オペレーティングシステムがポートを選択することを意味する、ポート 0 でリッスンするようにファクトリが構成されている場合に役立ちます。ソケットに接続する他のプロセスを開始する前に待機する必要がある場合は、isListening() のポーリングの代わりに使用することもできます。

リスニングスレッドが接続を受け入れるのを遅延させないために、イベントは別のスレッドで公開されます。

バージョン 4.3.2 から、クライアント接続を作成できない場合は常に TcpConnectionFailedEvent が発行されます。イベントのソースは接続ファクトリであり、これを使用して、接続を確立できなかったホストとポートを判別できます。

TCP アダプター

前述の接続ファクトリを使用する TCP 受信および送信チャネルアダプターが提供されます。これらのアダプターには、connection-factory および channel という 2 つの関連属性があります。connection-factory 属性は、アダプターの接続を管理するために使用される接続ファクトリを示します。channel 属性は、メッセージが送信アダプターに到着し、受信アダプターによってメッセージが配置されるチャネルを指定します。受信アダプターと送信アダプターの両方が接続ファクトリを共有できますが、サーバー接続ファクトリは常に受信アダプターによって「所有」されます。クライアント接続ファクトリは、常に送信アダプターによって「所有」されます。各型の 1 つのアダプターのみが接続ファクトリへの参照を取得できます。次の例は、クライアントとサーバーの TCP 接続ファクトリを定義する方法を示しています。

<bean id="javaSerializer"
      class="org.springframework.core.serializer.DefaultSerializer"/>
<bean id="javaDeserializer"
      class="org.springframework.core.serializer.DefaultDeserializer"/>

<int-ip:tcp-connection-factory id="server"
    type="server"
    port="1234"
    deserializer="javaDeserializer"
    serializer="javaSerializer"
    using-nio="true"
    single-use="true"/>

<int-ip:tcp-connection-factory id="client"
    type="client"
    host="localhost"
    port="#{server.port}"
    single-use="true"
    so-timeout="10000"
    deserializer="javaDeserializer"
    serializer="javaSerializer"/>

<int:channel id="input" />

<int:channel id="replies">
    <int:queue/>
</int:channel>

<int-ip:tcp-outbound-channel-adapter id="outboundClient"
    channel="input"
    connection-factory="client"/>

<int-ip:tcp-inbound-channel-adapter id="inboundClient"
    channel="replies"
    connection-factory="client"/>

<int-ip:tcp-inbound-channel-adapter id="inboundServer"
    channel="loop"
    connection-factory="server"/>

<int-ip:tcp-outbound-channel-adapter id="outboundServer"
    channel="loop"
    connection-factory="server"/>

<int:channel id="loop"/>

上記の構成では、input チャネルに到着したメッセージは、client 接続ファクトリによって作成された接続を介して直列化され、サーバーで受信され、loop チャネルに置かれます。loop は outboundServer の入力チャネルであるため、メッセージは同じ接続でループバックされ、inboundClient によって受信され、replies チャネルに格納されます。Java 直列化がワイヤ上で使用されます。

通常、受信アダプターは、受信接続リクエストをリッスンする type="server" 接続ファクトリを使用します。場合によっては、受信アダプターが外部サーバーに接続し、その接続で受信メッセージを待機するように、逆に接続を確立することができます。

このトポロジーは、受信アダプターで client-mode="true" を設定することでサポートされます。この場合、接続ファクトリの型は client でなければならず、single-use を false に設定する必要があります。

2 つの追加属性がこのメカニズムをサポートします。retry-interval は、接続障害後にフレームワークが再接続を試行する頻度（ミリ秒単位）を指定します。scheduler は TaskScheduler を提供して、接続試行をスケジュールし、接続がまだアクティブであることをテストします。

スケジューラを指定しない場合は、フレームワークのデフォルトの taskScheduler Bean が使用されます。

送信アダプターの場合、通常、最初のメッセージが送信されたときに接続が確立されます。送信アダプターの client-mode="true" により、アダプターの起動時に接続が確立されます。デフォルトでは、アダプターは自動的に開始されます。繰り返しますが、接続ファクトリは client 型であり、single-use="false" を持っている必要があります。retry-interval および scheduler もサポートされています。接続が失敗した場合、スケジューラによって、または次のメッセージが送信されたときに、接続が再確立されます。

受信と送信の両方で、アダプターが開始されている場合、<control-bus /> コマンド @adapter_id.retryConnection() を送信することにより、アダプターに強制的に接続を確立させることができます。その後、@adapter_id.isClientModeConnected() を使用して現在の状態を調べることができます。

TCP ゲートウェイ

受信 TCP ゲートウェイ TcpInboundGateway および送信 TCP ゲートウェイ TcpOutboundGateway は、それぞれサーバーおよびクライアント接続ファクトリを使用します。各接続は、一度に 1 つのリクエストまたはレスポンスを処理できます。

受信ゲートウェイは、受信ペイロードでメッセージを構築して requestChannel に送信した後、レスポンスを待機し、接続に書き込むことでレスポンスメッセージからペイロードを送信します。

受信ゲートウェイの場合、ip_connectionId ヘッダーは接続にメッセージを関連付けるために使用されるため、ip_connectionId ヘッダーを保持または追加する必要があります。ゲートウェイで発信されるメッセージには、ヘッダーが自動的に設定されます。返信が新しいメッセージとして作成される場合、ヘッダーを設定する必要があります。ヘッダー値は、受信メッセージから取得できます。

受信アダプターと同様に、受信ゲートウェイは通常、受信接続リクエストをリッスンする type="server" 接続ファクトリを使用します。場合によっては、受信ゲートウェイが外部サーバーに接続し、その接続で受信メッセージを待機して返信するように、逆に接続を確立することができます。

このトポロジは、受信ゲートウェイで client-mode="true" を使用することでサポートされます。この場合、接続ファクトリの型は client でなければならず、single-use を false に設定する必要があります。

2 つの追加属性がこのメカニズムをサポートします。retry-interval は、接続障害後にフレームワークが再接続を試行する頻度（ミリ秒単位）を指定します。scheduler は、TaskScheduler を提供して、接続試行をスケジュールし、接続がまだアクティブであることをテストします。

ゲートウェイが開始されている場合は、<control-bus/> コマンド @adapter_id.retryConnection() を送信してゲートウェイに接続を確立させ、@adapter_id.isClientModeConnected() で現在の状態を調べることができます。

送信ゲートウェイは、接続を介してメッセージを送信した後、レスポンスを待機し、レスポンスメッセージを構築して、レスポンスチャネルに配置します。接続を介した通信はシングルスレッドです。一度に処理できるメッセージは 1 つだけです。現在のレスポンスを受信する前に別のスレッドがメッセージを送信しようとすると、以前のリクエストが完了する（またはタイムアウトする）までブロックされます。ただし、クライアント接続ファクトリが使い捨て接続用に構成されている場合、新しいリクエストはそれぞれ独自の接続を取得し、すぐに処理されます。次の例は、受信 TCP ゲートウェイを構成します。

<int-ip:tcp-inbound-gateway id="inGateway"
    request-channel="tcpChannel"
    reply-channel="replyChannel"
    connection-factory="cfServer"
    reply-timeout="10000"/>

デフォルトのシリアライザーまたはデシリアライザーで構成された接続ファクトリが使用される場合、メッセージは \r\n で区切られたデータであり、telnet などの単純なクライアントがゲートウェイを使用できます。

次の例は、送信 TCP ゲートウェイを示しています。

<int-ip:tcp-outbound-gateway id="outGateway"
    request-channel="tcpChannel"
    reply-channel="replyChannel"
    connection-factory="cfClient"
    request-timeout="10000"
    remote-timeout="10000"/> <!-- or e.g.
remote-timeout-expression="headers['timeout']" -->

client-mode は現在、送信ゲートウェイでは使用できません。

バージョン 5.2 以降、発信ゲートウェイはプロパティ closeStreamAfterSend で構成できます。接続ファクトリが single-use （各リクエスト / 応答の新しい接続）用に構成されている場合、ゲートウェイは出力ストリームを閉じます。これは、サーバーに EOF を通知します。これは、サーバーがストリーム内の区切り文字ではなく EOF を使用してメッセージの終わりを判別する場合に役立ちますが、応答を受信するために接続を開いたままにします。

通常、呼び出しスレッドはゲートウェイでブロックされ、応答（またはタイムアウト）を待ちます。バージョン 5.3 以降、ゲートウェイで async プロパティを設定でき、送信スレッドが解放されて他の作業を行うことができます。返信（またはエラー）は受信スレッドで送信されます。これは、TcpNetClientConnectionFactory を使用する場合にのみ適用されます。NIO を使用する場合は無視されます。これは、応答の受信後に発生するソケットエラーが、応答の前にゲートウェイに渡される可能性があるためです。

共有接続（singleUse=false）を使用している場合、新しいリクエストは、別のリクエストが処理されている間、現在の応答が受信されるまでブロックされます。長期間有効な接続のプールでの同時リクエストをサポートする場合は、CachingClientConnectionFactory の使用を検討してください。

TCP メッセージ相関

IP エンドポイントの 1 つのゴールは、Spring Integration アプリケーション以外のシステムとの通信を提供することです。このため、デフォルトではメッセージペイロードのみが送受信されます。3.0 以降、JSON、Java 直列化、カスタムシリアライザーとデシリアライザーを使用してヘッダーを転送できます。詳細については、ヘッダーの転送を参照してください。フレームワーク（ゲートウェイを使用する場合を除く）またはサーバー側のチャネルアダプターのコラボレーションでは、メッセージの相関は提供されません。このドキュメントの後半では、アプリケーションで利用可能なさまざまな相関手法について説明します。ほとんどの場合、これには、メッセージペイロードに自然な相関データ（オーダー番号など）が含まれている場合でも、メッセージの特定のアプリケーションレベルの相関が必要です。

ゲートウェイ

ゲートウェイはメッセージを自動的に関連付けます。ただし、比較的少量のアプリケーションには送信ゲートウェイを使用する必要があります。すべてのメッセージペアに対して単一の共有接続を使用するように接続ファクトリを構成すると（'single-use="false" '）、一度に処理できるメッセージは 1 つだけです。新しいメッセージは、前のメッセージへの応答が受信されるまで待機する必要があります。新しいメッセージを使用して新しい接続を使用するように接続ファクトリを構成する場合（'single-use="true" '）、この制限は適用されません。この設定は、共有接続環境よりも高いスループットを提供できますが、メッセージペアごとに新しい接続を開いたり閉じたりするオーバーヘッドが伴います。

大量のメッセージについては、チャネルアダプターのコラボレーションペアの使用を検討してください。ただし、そのためには、コラボレーションロジックを提供する必要があります。

Spring Integration 2.2 で導入された別のソリューションは、CachingClientConnectionFactory を使用することです。これにより、共有接続のプールを使用できます。

送信および受信チャネルアダプターのコラボレーション

大容量のスループットを実現するために（前述のようにゲートウェイを使用する場合の落とし穴を回避するために）、連携する送信および受信チャネルアダプターのペアを構成できます。また、完全に非同期の通信のために、コラボレーションアダプター（サーバー側またはクライアント側）を使用できます（リクエスト - 応答セマンティクスではなく）。サーバー側では、受信アダプターが応答メッセージの送信時に使用する接続を送信アダプターが決定できるヘッダーを追加するため、メッセージ相関はアダプターによって自動的に処理されます。

サーバー側では、ip_connectionId ヘッダーを設定する必要があります。これは、メッセージと接続を関連付けるために使用されるためです。受信アダプターから発信されるメッセージには、自動的にヘッダーが設定されます。送信する他のメッセージを作成する場合は、ヘッダーを設定する必要があります。受信メッセージからヘッダー値を取得できます。

クライアント側では、アプリケーションは必要に応じて独自の相関ロジックを提供する必要があります。これにはいくつかの方法があります。

メッセージペイロードに自然な相関データ（トランザクション ID やオーダー番号など）があり、元の送信メッセージからの情報（応答チャネルヘッダーなど）を保持する必要がない場合、相関は単純で、いずれにしても、アプリケーションレベルで行われます。

メッセージペイロードに自然な相関データ（トランザクション ID やオーダー番号など）が含まれているが、元の送信メッセージからの情報（応答チャネルヘッダーなど）を保持する必要がある場合、元の送信メッセージのコピーを保持できます。発信メッセージ（パブリッシュ / サブスクライブチャネルを使用するなど）を使用し、アグリゲーターを使用して必要なデータを再結合します。

前の 2 つのシナリオのいずれかで、ペイロードに自然な相関データがない場合、送信チャネルアダプターの上流にトランスフォーマーを提供して、そのようなデータでペイロードを強化できます。このようなトランスフォーマーは、元のペイロードを、元のペイロードとメッセージヘッダーのサブセットの両方を含む新しいオブジェクトに変換できます。もちろん、ヘッダーからのライブオブジェクト（応答チャネルなど）は、変換されたペイロードに含めることはできません。

このような戦略を選択する場合は、接続ファクトリに適切なシリアライザーとデシリアライザーのペアがあり、そのようなペイロードを処理する必要があります（java 直列化を使用する DefaultSerializer や DefaultDeserializer、またはカスタムシリアライザーとデシリアライザー）。デフォルト ByteArrayCrLfSerializer を含む TCP 接続ファクトリでメンションされた ByteArray*Serializer オプションは、変換されたペイロードが String または byte[] でない限り、そのようなペイロードをサポートしません。

2.2 リリースの前は、コラボレーションチャネルアダプターがクライアント接続ファクトリを使用していた場合、so-timeout 属性はデフォルトの応答タイムアウト（10 秒）にデフォルト設定されていました。これは、この時間内に受信アダプターがデータを受信しなかった場合、ソケットが閉じられたことを意味します。

このデフォルトの動作は、真の非同期環境では適切ではなかったため、デフォルトで無限タイムアウトになりました。クライアント接続ファクトリの so-timeout 属性を 10000 ミリ秒に設定することにより、以前のデフォルトの動作に戻すことができます。

ヘッダーの転送

TCP はストリーミングプロトコルです。Serializers および Deserializers は、ストリーム内のメッセージを区別します。3.0 より前は、メッセージペイロード（String または byte[]）のみが TCP を介して転送できました。3.0 からは、選択したヘッダーとペイロードを転送できます。ただし、replyChannel ヘッダーなどの「ライブ」オブジェクトは直列化できません。

TCP でヘッダー情報を送信するには、追加の構成が必要です。

最初のステップは、mapper 属性を使用する MessageConvertingTcpMessageMapper を ConnectionFactory に提供することです。このマッパーは、任意の MessageConverter 実装に委譲して、構成された serializer および deserializer によってシリアライズおよびデシリアライズできるオブジェクトとの間でメッセージを変換します。

Spring Integration は MapMessageConverter を提供します。これにより、ペイロードとともに Map オブジェクトに追加されるヘッダーのリストを指定できます。生成されたマップには、payload と headers の 2 つのエントリがあります。headers エントリ自体が Map であり、選択されたヘッダーが含まれています。

2 番目のステップは、Map と何らかのワイヤー形式の間で変換できるシリアライザーとデシリアライザーを提供することです。これは、ピアシステムが Spring Integration アプリケーションでない場合に通常必要となるカスタム Serializer または Deserializer にすることができます。

Spring Integration は、MapJsonSerializer を提供して、Map と JSON を相互に変換します。Spring Integration JsonObjectMapper を使用します。必要に応じて、カスタム JsonObjectMapper を提供できます。デフォルトでは、シリアライザーはオブジェクト間にラインフィード（0x0a）文字を挿入します。詳細については、Javadoc を参照してください。

JsonObjectMapper は、クラスパスにある Jackson のバージョンを使用します。

DefaultSerializer および DefaultDeserializer を使用して、Map の標準 Java 直列化を使用することもできます。

次の例は、JSON を使用して correlationId、sequenceNumber、sequenceSize ヘッダーを転送する接続ファクトリの構成を示しています。

<int-ip:tcp-connection-factory id="client"
    type="client"
    host="localhost"
    port="12345"
    mapper="mapper"
    serializer="jsonSerializer"
    deserializer="jsonSerializer"/>

<bean id="mapper"
      class="o.sf.integration.ip.tcp.connection.MessageConvertingTcpMessageMapper">
    <constructor-arg name="messageConverter">
        <bean class="o.sf.integration.support.converter.MapMessageConverter">
            <property name="headerNames">
                <list>
                    <value>correlationId</value>
                    <value>sequenceNumber</value>
                    <value>sequenceSize</value>
                </list>
            </property>
        </bean>
    </constructor-arg>
</bean>

<bean id="jsonSerializer" class="o.sf.integration.ip.tcp.serializer.MapJsonSerializer" />

上記の構成で送信され、「何か」のペイロードを持つメッセージは、次のようにワイヤに表示されます。

{"headers":{"correlationId":"things","sequenceSize":5,"sequenceNumber":1},"payload":"something"}

ノンブロッキング I/O について (NIO)

NIO（IP 設定属性の using-nio を参照）を使用すると、各ソケットから読み取るためのスレッド専用になりません。少数のソケットの場合、NIO を使用せずに（QueueChannel などへの）非同期ハンドオフを実行すると、NIO を使用した場合と同等以上のパフォーマンスが得られる可能性があります。

多数の接続を処理する場合は、NIO の使用を検討する必要があります。ただし、NIO の使用には他にもいくつかの影響があります。（タスクエグゼキューター内の）スレッドのプールは、すべてのソケットで共有されます。各受信メッセージは、そのプールから選択されたスレッドの個別の作業単位として構成され、構成されたチャネルに送信されます。同じソケットに到着する 2 つの連続したメッセージは、異なるスレッドによって処理される場合があります。つまり、メッセージがチャネルに送信される順序は不定です。ソケットに到着するメッセージの厳密な順序は維持されません。

一部のアプリケーションでは、これは課題ではありません。他の人にとっては、課題です。厳密な順序付けが必要な場合は、using-nio を false に設定し、非同期ハンドオフを使用することを検討してください。

または、受信エンドポイントの下流にリシーケンサーを挿入して、メッセージを適切なシーケンスに戻すことができます。接続ファクトリで apply-sequence を true に設定すると、TCP 接続に到着するメッセージには sequenceNumber および correlationId ヘッダーが設定されます。リシーケンサはこれらのヘッダーを使用して、メッセージを適切なシーケンスに戻します。

バージョン 5.1.4 以降、既存の接続からの読み取りよりも新しい接続の受け入れが優先されます。一般に、新しい受信接続の割合が非常に高い場合を除き、これはほとんど影響を与えません。読み取り優先度を与える以前の動作に戻したい場合は、TcpNioServerConnectionFactory の multiAccept プロパティを false に設定します。

プールサイズ

プールサイズ属性は使用されなくなりました。以前は、タスクエグゼキューターが指定されていない場合、デフォルトのスレッドプールのサイズを指定していました。サーバーソケットの接続バックログを設定するためにも使用されました。最初の関数は不要になりました（次の段落を参照）。2 番目の関数は、backlog 属性に置き換えられます。

以前は、NIO で固定スレッドプールタスクエグゼキューター（デフォルト）を使用すると、デッドロックが発生して処理が停止することがありました。この問題は、バッファがいっぱいで、ソケットから読み取っているスレッドがバッファにさらにデータを追加しようとしており、バッファ内にスペースを作成するためのスレッドがなかったときに発生しました。これは、プールサイズが非常に小さい場合にのみ発生しましたが、極端な条件下では可能です。2.2 以降、2 つの変更によりこの問題が解消されました。まず、デフォルトのタスクエグゼキューターはキャッシュスレッドプールエグゼキューターです。次に、デッドロック検出ロジックが追加され、デッドロックの代わりにスレッド不足が発生した場合、例外がスローされ、デッドロックされたリソースが解放されます。

デフォルトのタスクエグゼキューターは制限されていないため、メッセージ処理に時間がかかる場合、高いレートの受信メッセージでメモリ不足状態が発生する可能性があります。アプリケーションがこの型の動作を示す場合、適切なプールサイズのプールされたタスクエグゼキューターを使用する必要がありますが、次のセクションを参照してください。

CALLER_RUNS ポリシーを使用したスレッドプールタスクエグゼキュータ

CallerRunsPolicy （<task/> 名前空間を使用する場合は CALLER_RUNS）で固定スレッドプールを使用し、キュー容量が小さい場合、いくつかの重要な考慮事項に留意する必要があります。

固定スレッドプールを使用しない場合、以下は適用されません。

NIO 接続では、3 つの異なるタスク型があります。I/O セレクター処理は、1 つの専用スレッドで実行されます（イベントの検出、新しい接続の受け入れ、タスクエグゼキューターを使用した他のスレッドへの I/O 読み取り操作のディスパッチ）。I/O リーダースレッド（読み取り操作のディスパッチ先）がデータを読み取ると、別のスレッドに渡されて受信メッセージを組み立てます。大きなメッセージは、完了するまでに数回の読み取りが必要になる場合があります。これらの「アセンブラ」スレッドは、データを待っている間ブロックできます。新しい読み取りイベントが発生すると、リーダーはこのソケットにすでにアセンブラがあるかどうかを判断し、ない場合は新しいものを実行します。アセンブリプロセスが完了すると、アセンブラスレッドがプールに返されます。

これにより、プールが使い果たされ、CALLER_RUNS 拒否ポリシーが使用され、タスクキューがいっぱいになると、デッドロックが発生する可能性があります。プールが空でキューに空きがない場合、IO セレクタースレッドは OP_READ イベントを受け取り、executor を使用して読み取りをディスパッチします。キューがいっぱいなので、セレクタースレッド自体が読み取りプロセスを開始します。現在、このソケットにアセンブラーがないことを検出し、読み取りを行う前にアセンブラーを起動します。再び、キューがいっぱいになり、セレクタースレッドがアセンブラーになります。これでアセンブラーはブロックされ、データの読み取りを待機しますが、読み取りは行われません。セレクタスレッドは新しいイベントを処理できないため、接続ファクトリはデッドロックされています。

このデッドロックを回避するには、アセンブリ（タスク）を実行するセレクター（またはリーダー）スレッドを回避する必要があります。IO 操作とアセンブリ操作に別々のプールを使用する必要があります。

フレームワークは CompositeExecutor を提供します。これにより、2 つの異なるエグゼキューターの構成が可能になります。1 つは IO 操作を実行し、もう 1 つはメッセージアセンブリ用です。この環境では、IO スレッドがアセンブラスレッドになることはなく、デッドロックは発生しません。

さらに、タスク実行プログラムは AbortPolicy （<task> を使用する場合は ABORT）を使用するように構成する必要があります。I/O タスクが完了できない場合、短時間延期され、完了してアセンブラーが割り当てられるまで継続的に再試行されます。デフォルトでは、遅延は 100 ミリ秒ですが、接続ファクトリで readDelay プロパティを設定することで変更できます（XML 名前空間で構成する場合は read-delay）。

次の 3 つの例は、複合エグゼキューターの構成方法を示しています。

@Bean
private CompositeExecutor compositeExecutor() {
    ThreadPoolTaskExecutor ioExec = new ThreadPoolTaskExecutor();
    ioExec.setCorePoolSize(4);
    ioExec.setMaxPoolSize(10);
    ioExec.setQueueCapacity(0);
    ioExec.setThreadNamePrefix("io-");
    ioExec.setRejectedExecutionHandler(new AbortPolicy());
    ioExec.initialize();
    ThreadPoolTaskExecutor assemblerExec = new ThreadPoolTaskExecutor();
    assemblerExec.setCorePoolSize(4);
    assemblerExec.setMaxPoolSize(10);
    assemblerExec.setQueueCapacity(0);
    assemblerExec.setThreadNamePrefix("assembler-");
    assemblerExec.setRejectedExecutionHandler(new AbortPolicy());
    assemblerExec.initialize();
    return new CompositeExecutor(ioExec, assemblerExec);
}
<bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
    <constructor-arg ref="io"/>
    <constructor-arg ref="assembler"/>
</bean>

<task:executor id="io" pool-size="4-10" queue-capacity="0" rejection-policy="ABORT" />
<task:executor id="assembler" pool-size="4-10" queue-capacity="0" rejection-policy="ABORT" />
<bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
    <constructor-arg>
        <bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
            <property name="threadNamePrefix" value="io-" />
            <property name="corePoolSize" value="4" />
            <property name="maxPoolSize" value="8" />
            <property name="queueCapacity" value="0" />
            <property name="rejectedExecutionHandler">
                <bean class="java.util.concurrent.ThreadPoolExecutor.AbortPolicy" />
            </property>
        </bean>
    </constructor-arg>
    <constructor-arg>
        <bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
            <property name="threadNamePrefix" value="assembler-" />
            <property name="corePoolSize" value="4" />
            <property name="maxPoolSize" value="10" />
            <property name="queueCapacity" value="0" />
            <property name="rejectedExecutionHandler">
                <bean class="java.util.concurrent.ThreadPoolExecutor.AbortPolicy" />
            </property>
        </bean>
    </constructor-arg>
</bean>

SSL/TLS サポート

Secure Sockets Layer/Transport Layer Security がサポートされています。NIO を使用する場合、JDK 5+ SSLEngine 機能を使用して、接続の確立後にハンドシェイクを処理します。NIO を使用しない場合、標準の SSLSocketFactory および SSLServerSocketFactory オブジェクトを使用して接続が作成されます。大幅なカスタマイズを可能にするために、多くの戦略インターフェースが提供されています。これらのインターフェースのデフォルト実装は、安全な通信を開始する最も簡単な方法を提供します。

入門

NIO を使用するかどうかに関係なく、接続ファクトリで ssl-context-support 属性を構成する必要があります。この属性は、必要なキーストアの場所とパスワードを記述する <Bean/> 定義を参照します。

SSL/TLS ピアには、それぞれ 2 つのキーストアが必要です。

  • ピアを識別するための秘密鍵と公開鍵のペアを含むキーストア

  • 信頼されているピアの公開鍵を含むトラストストア。JDK で提供される keytool ユーティリティのドキュメントを参照してください。不可欠なステップは

    1. 新しいキーペアを作成し、キーストアに保存します。

    2. 公開鍵をエクスポートします。

    3. 公開鍵をピアのトラストストアにインポートします。

    4. 他のピアについても繰り返します。

テストケースでは、両方のピアで同じキーストアを使用するのが一般的ですが、本番環境ではこれを避ける必要があります。

キーストアを確立した後、次のステップは、それらのロケーションを TcpSSLContextSupport Bean に示し、その Bean への参照を接続ファクトリに提供することです。

次の例では、SSL 接続を構成します。

<bean id="sslContextSupport"
    class="o.sf.integration.ip.tcp.connection.support.DefaultTcpSSLContextSupport">
    <constructor-arg value="client.ks"/>
    <constructor-arg value="client.truststore.ks"/>
    <constructor-arg value="secret"/>
    <constructor-arg value="secret"/>
</bean>

<ip:tcp-connection-factory id="clientFactory"
    type="client"
    host="localhost"
    port="1234"
    ssl-context-support="sslContextSupport" />

DefaultTcpSSLContextSupport クラスには、オプションの protocol プロパティもあり、SSL または TLS （デフォルト）にすることができます。

キーストアファイル名（最初の 2 つのコンストラクター引数）は、Spring Resource 抽象化を使用します。デフォルトでは、ファイルはクラスパスにありますが、file: プレフィックスを使用してこれをオーバーライドできます（代わりにファイルシステムでファイルを検索します）。

バージョン 4.3.6 以降、NIO を使用する場合、接続ファクトリで ssl-handshake-timeout （秒単位）を指定できます。このタイムアウト（デフォルトは 30 秒）は、データを待機するときの SSL ハンドシェーク中に使用されます。タイムアウトを超えた場合、プロセスは中止され、ソケットが閉じられます。

ホスト検証

バージョン 5.0.8 から、ホスト検証を有効にするかどうかを構成できます。バージョン 5.1 以降、デフォルトで有効になっています。無効にするメカニズムは、NIO を使用しているかどうかによって異なります。

ホスト検証は、証明書が信頼されている場合でも、接続しているサーバーが証明書の情報と一致することを確認するために使用されます。

NIO を使用する場合、たとえば DefaultTcpNioSSLConnectionSupport を構成します。

@Bean
public DefaultTcpNioSSLConnectionSupport connectionSupport() {
    DefaultTcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks",
            "test.truststore.ks", "secret", "secret");
    sslContextSupport.setProtocol("SSL");
    DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport =
            new DefaultTcpNioSSLConnectionSupport(sslContextSupport, false);
    return tcpNioConnectionSupport;
}

2 番目のコンストラクター引数は、ホスト検証を無効にします。次に、connectionSupport Bean が NIO 接続ファクトリに挿入されます。

NIO を使用しない場合、構成は TcpSocketSupport にあります。

connectionFactory.setTcpSocketSupport(new DefaultTcpSocketSupport(false));

繰り返しますが、コンストラクター引数はホスト検証を無効にします。

高度なテクニック

このセクションでは、特定の状況で役立つ高度なテクニックについて説明します。

戦略インターフェース

多くの場合、TCP/IP を介した安全な通信を可能にするために必要なのは、前述の構成だけです。ただし、Spring Integration は、ソケットファクトリとソケットのカスタマイズと変更を可能にするいくつかの戦略インターフェースを提供します。

  • TcpSSLContextSupport

  • TcpSocketFactorySupport

  • TcpSocketSupport

  • TcpNetConnectionSupport

  • TcpNioConnectionSupport

TcpSSLContextSupport 戦略インターフェース

以下のリストは、TcpSSLContextSupport 戦略インターフェースを示しています。

public interface TcpSSLContextSupport {

    SSLContext getSSLContext() throws Exception;

}

TcpSSLContextSupport インターフェースの実装は、SSLContext オブジェクトの作成を担当します。フレームワークによって提供された実装は DefaultTcpSSLContextSupport です、以前に説明されます。異なる動作が必要な場合は、このインターフェースを実装し、クラスの実装の Bean への参照を接続ファクトリに提供します。

TcpSocketFactorySupport 戦略インターフェース

次のリストは、TcpSocketFactorySupport 戦略インターフェースの定義を示しています。

public interface TcpSocketFactorySupport {

    ServerSocketFactory getServerSocketFactory();

    SocketFactory getSocketFactory();

}

このインターフェースの実装は、ServerSocketFactory と SocketFactory への参照を取得する責任があります。2 つの実装が提供されます。1 つ目は、非 SSL ソケット用の DefaultTcpNetSocketFactorySupport です（ssl-context-support 属性が定義されていない場合）。これは、JDK のデフォルトファクトリを使用します。2 番目の実装は DefaultTcpNetSSLSocketFactorySupport です。デフォルトでは、これは ssl-context-support 属性が定義されているときに使用されます。その Bean によって作成された SSLContext を使用して、ソケットファクトリを作成します。

このインターフェースは、using-nio が false の場合にのみ適用されます。NIO はソケットファクトリを使用しません。
TcpSocketSupport 戦略インターフェース

次のリストは、TcpSocketSupport 戦略インターフェースの定義を示しています。

public interface TcpSocketSupport {

    void postProcessServerSocket(ServerSocket serverSocket);

    void postProcessSocket(Socket socket);

}

このインターフェースの実装は、ソケットが作成され、構成されたすべての属性が適用された後、ソケットが使用される前にソケットを変更できます。これは、NIO を使用するかどうかに関係なく適用されます。例: このインターフェースの実装を使用して、SSL ソケットでサポートされている暗号スイートを変更したり、SSL ハンドシェイクの完了後に通知を受けるリスナーを追加したりできます。フレームワークによって提供される唯一の実装は DefaultTcpSocketSupport であり、ソケットを一切変更しません。

TcpSocketFactorySupport または TcpSocketSupport の独自の実装を提供するには、それぞれ socket-factory-support および socket-support 属性を設定して、接続ファクトリにカスタム型の Bean への参照を提供します。

TcpNetConnectionSupport 戦略インターフェース

次のリストは、TcpNetConnectionSupport 戦略インターフェースの定義を示しています。

public interface TcpNetConnectionSupport {

    TcpNetConnection createNewConnection(Socket socket,
            boolean server, boolean lookupHost,
            ApplicationEventPublisher applicationEventPublisher,
            String connectionFactoryName) throws Exception;

}

このインターフェースは、型 TcpNetConnection （またはそのサブクラス）のオブジェクトを作成するために呼び出されます。このフレームワークは、デフォルトで単純な TcpNetConnection オブジェクトを作成する単一の実装（DefaultTcpNetConnectionSupport）を提供します。pushbackCapable と pushbackBufferSize の 2 つのプロパティがあります。プッシュバックが有効な場合、実装は接続の InputStream を PushbackInputStream にラップするサブクラスを返します。PushbackInputStream のデフォルトに合わせて、バッファーサイズのデフォルトは 1 になります。これにより、デシリアライザーはストリームにバイトを「未読」（プッシュバック）できます。次の簡単な例は、最初のバイトを「覗き」、どのデシリアライザーを呼び出すかを決定する委譲デシリアライザーでどのように使用されるかを示しています。

public class CompositeDeserializer implements Deserializer<byte[]> {

    private final ByteArrayStxEtxSerializer stxEtx = new ByteArrayStxEtxSerializer();

    private final ByteArrayCrLfSerializer crlf = new ByteArrayCrLfSerializer();

    @Override
    public byte[] deserialize(InputStream inputStream) throws IOException {
        PushbackInputStream pbis = (PushbackInputStream) inputStream;
        int first = pbis.read();
        if (first < 0) {
            throw new SoftEndOfStreamException();
        }
        pbis.unread(first);
        if (first == ByteArrayStxEtxSerializer.STX) {
            this.receivedStxEtx = true;
            return this.stxEtx.deserialize(pbis);
        }
        else {
            this.receivedCrLf = true;
            return this.crlf.deserialize(pbis);
        }
    }

}
TcpNioConnectionSupport 戦略インターフェース

次のリストは、TcpNioConnectionSupport 戦略インターフェースの定義を示しています。

public interface TcpNioConnectionSupport {

    TcpNioConnection createNewConnection(SocketChannel socketChannel,
            boolean server, boolean lookupHost,
            ApplicationEventPublisher applicationEventPublisher,
            String connectionFactoryName) throws Exception;

}

このインターフェースは、TcpNioConnection オブジェクト（またはサブクラスのオブジェクト）を作成するために呼び出されます。Spring Integration は、DefaultTcpNioSSLConnectionSupport と DefaultTcpNioConnectionSupport の 2 つの実装を提供します。どちらが使用されるかは、SSL が使用されているかどうかによって異なります。一般的な使用例は、DefaultTcpNioSSLConnectionSupport をサブクラス化し、postProcessSSLEngine をオーバーライドすることです。SSL クライアント認証の例を参照してください。DefaultTcpNetConnectionSupport と同様に、これらの実装もプッシュバックをサポートしています。

サンプル: SSL クライアント認証の有効化

SSL を使用するときにクライアント証明書認証を有効にする方法は、NIO を使用するかどうかによって異なります。NIO を使用しない場合は、サーバーソケットを後処理するカスタム TcpSocketSupport 実装を提供します。

serverFactory.setTcpSocketSupport(new DefaultTcpSocketSupport() {

    @Override
    public void postProcessServerSocket(ServerSocket serverSocket) {
        ((SSLServerSocket) serverSocket).setNeedClientAuth(true);
    }

});

（XML 構成を使用する場合、socket-support 属性を設定して Bean への参照を提供します）。

NIO を使用する場合、次の例に示すように、SSLEngine を後処理するカスタム TcpNioSslConnectionSupport 実装を提供します。

@Bean
public DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport() {
    return new DefaultTcpNioSSLConnectionSupport(serverSslContextSupport) {

            @Override
            protected void postProcessSSLEngine(SSLEngine sslEngine) {
                sslEngine.setNeedClientAuth(true);
            }

    }
}

@Bean
public TcpNioServerConnectionFactory server() {
    ...
    serverFactory.setTcpNioConnectionSupport(tcpNioConnectionSupport());
    ...
}

（バージョン 4.3.7 以降、XML 構成を使用する場合は、nio-connection-support 属性を設定して Bean への参照を提供します）。

IP 設定属性

次の表に、IP 接続を構成するために設定できる属性を示します。

表 1: 接続ファクトリの属性
属性名 クライアント ? サーバー ? 許容値 属性記述

type

Y

Y

クライアントサーバー

接続ファクトリがクライアントであるかサーバーであるかを決定します。

host

Y

N

宛先のホスト名または IP アドレス。

port

Y

Y

ポート。

serializer

Y

Y

ペイロードの直列化に使用される Serializer の実装。デフォルトは ByteArrayCrLfSerializer

deserializer

Y

Y

ペイロードの逆直列化に使用される Deserializer の実装。デフォルトは ByteArrayCrLfSerializer

using-nio

Y

Y

true, false

接続で NIO を使用するかどうか。詳細については、java.nio パッケージを参照してください。ノンブロッキング I/O について (NIO) を参照してください。デフォルト: false

using-direct-buffers

Y

N

true, false

NIO を使用する場合、接続がダイレクトバッファーを使用するかどうか。詳細については、java.nio.ByteBuffer のドキュメントを参照してください。using-nio が false の場合、false でなければなりません。

apply-sequence

Y

Y

true, false

NIO を使用する場合、メッセージの順序を変更する必要がある場合があります。この属性を true、correlationId、sequenceNumber に設定すると、受信したメッセージにヘッダーが追加されます。ノンブロッキング I/O について (NIO) を参照してください。デフォルト: false。

so-timeout

Y

Y

single-use="true" のサーバー接続ファクトリを除き、デフォルトは 0 （無限大）です。その場合、デフォルトのデフォルトの応答タイムアウト（10 秒）になります。

so-send-buffer-size

Y

Y

java.net.Socket. setSendBufferSize() を参照してください。

so-receive-buffer-size

Y

Y

java.net.Socket. setReceiveBufferSize() を参照してください。

so-keep-alive

Y

Y

true, false

java.net.Socket.setKeepAlive() を参照してください。

so-linger

Y

Y

指定された値で linger を true に設定します。java.net.Socket.setSoLinger() を参照してください。

so-tcp-no-delay

Y

Y

true, false

java.net.Socket.setTcpNoDelay() を参照してください。

so-traffic-class

Y

Y

java.net.Socket. setTrafficClass() を参照してください。

local-address

N

Y

マルチホームシステムでは、ソケットがバインドされているインターフェースの IP アドレスを指定します。

task-executor

Y

Y

ソケット処理に使用される特定のエグゼキューターを指定します。指定しない場合、内部キャッシュスレッドエグゼキューターが使用されます。WorkManagerTaskExecutor など、特定のタスクエグゼキューターの使用を必要とする一部のプラットフォームで必要です。

single-use

Y

Y

true, false

接続を複数のメッセージに使用できるかどうかを指定します。true の場合、各メッセージに新しい接続が使用されます。

pool-size

N

N

この属性は使用されなくなりました。下位互換性のために、バックログを設定しますが、backlog を使用してサーバーファクトリで接続バックログを指定する必要があります。

backlog

N

Y

サーバーファクトリの接続バックログを設定します。

lookup-host

Y

Y

true, false

メッセージヘッダーで使用するホスト名に変換するために、IP アドレスで逆引きを行うかどうかを指定します。false の場合、代わりに IP アドレスが使用されます。デフォルト: true

interceptor-factory-chain

Y

Y

TCP 接続インターセプターを参照してください。

ssl-context-support

Y

Y

SSL/TLS Support を参照してください。

socket-factory-support

Y

Y

SSL/TLS Support を参照してください。

socket-support

Y

Y

SSL/TLS サポートを参照してください。

nio-connection-support

Y

Y

高度なテクニックを参照してください。

read-delay

Y

Y

長い> 0

前の試行がスレッド不足のために失敗した後、読み取りを再試行するまでの遅延（ミリ秒）。デフォルト: 100。using-nio が true の場合にのみ適用されます。

次の表は、UDP 受信チャネルアダプターを構成するために設定できる属性を示しています。

表 2: UDP 受信チャネルアダプターの属性
属性名 許容値 属性記述

port

アダプターがリッスンするポート。

multicast

true, false

UDP アダプターがマルチキャストを使用するかどうか。

multicast-address

マルチキャストが真の場合、アダプターが参加するマルチキャストアドレス。

pool-size

同時に処理できるパケットの数を指定します。task-executor が構成されていない場合にのみ適用されます。デフォルト: 5。

task-executor

ソケット処理に使用される特定のエグゼキューターを指定します。提供されない場合、内部のプールされたエグゼキューターが使用されます。WorkManagerTaskExecutor などの特定のタスクエグゼキューターの使用を必要とする一部のプラットフォームで必要です。スレッドの要件については、pool-size を参照してください。

receive-buffer-size

DatagramPackets を受信するために使用されるバッファーのサイズ。通常、最大伝送単位（MTU）サイズに設定されます。送信されたパケットのサイズよりも小さいバッファが使用される場合、切り捨てが発生する可能性があります。これは、check-length 属性を使用して検出できます。

check-length

true, false

UDP アダプターが、受信したパケットのデータ長フィールドを予期するかどうか。パケットの切り捨てを検出するために使用されます。

so-timeout

詳細については、java.net.DatagramSocket の setSoTimeout() メソッドを参照してください。

so-send-buffer-size

UDP 確認応答パケットに使用されます。詳細については、java.net.DatagramSocket の setSendBufferSize() メソッドを参照してください。

so-receive-buffer-size

詳細については、java.net.DatagramSocket.setReceiveBufferSize() を参照してください。

local-address

マルチホームシステムでは、ソケットがバインドされているインターフェースの IP アドレスを指定します。

error-channel

ダウンストリームコンポーネントが例外をスローすると、例外と失敗したメッセージを含む MessagingException メッセージがこのチャネルに送信されます。

lookup-host

true, false

メッセージヘッダーで使用するホスト名に変換するために、IP アドレスで逆引きを行うかどうかを指定します。false の場合、代わりに IP アドレスが使用されます。デフォルト: true

次の表は、UDP 送信チャネルアダプターを構成するために設定できる属性を示しています。

表 3: UDP 送信チャネルアダプターの属性
属性名 許容値 属性記述

host

宛先のホスト名または IP アドレス。マルチキャスト udp アダプターの場合、マルチキャストアドレス。

port

宛先のポート。

multicast

true, false

udp アダプターがマルチキャストを使用するかどうか。

acknowledge

true, false

UDP アダプターが宛先からの確認応答を必要とするかどうか。有効にすると、次の 4 つの属性を設定する必要があります: ack-host、ack-port、ack-timeout、min-acks-for- success。

ack-host

acknowledge が true の場合、確認応答の送信先のホストまたは IP アドレスを示します。通常は現在のホストですが、異なる場合があります。たとえば、ネットワークアドレス変換（NAT）が使用されている場合です。

ack-port

acknowledge が true の場合、確認応答の送信先ポートを示します。アダプターは、このポートで確認を待機します。

ack-timeout

acknowledge が true の場合、アダプターが確認応答を待機する時間をミリ秒で示します。確認が時間内に受信されない場合、アダプターは例外をスローします。

min-acks-for- success

デフォルトは 1 です。マルチキャストアダプターの場合、これをより大きな値に設定できます。これには、複数の宛先からの確認が必要です。

check-length

true, false

UDP アダプターが宛先に送信されるパケットにデータ長フィールドを含めるかどうか。

time-to-live

マルチキャストアダプターの場合、MulticastSocket の存続時間属性を指定します。マルチキャストの範囲を制御します。詳細については、Java API のドキュメントを参照してください。

so-timeout

詳細については、java.net.DatagramSocket setSoTimeout() メソッドを参照してください。

so-send-buffer-size

詳細については、java.net.DatagramSocket の setSendBufferSize() メソッドを参照してください。

so-receive-buffer-size

UDP 確認応答パケットに使用されます。詳細については、java.net.DatagramSocket の setReceiveBufferSize() メソッドを参照してください。

local-address

マルチホームシステムで、UDP アダプターの場合、応答メッセージ用にソケットがバインドされているインターフェースの IP アドレスを指定します。マルチキャストアダプターの場合、マルチキャストパケットが送信されるインターフェースも決定します。

task-executor

確認応答の処理に使用する特定のエグゼキューターを指定します。提供されない場合、内部シングルスレッドエグゼキューターが使用されます。WorkManagerTaskExecutor など、特定のタスクエグゼキューターの使用を必要とする一部のプラットフォームで必要です。1 つのスレッドは、確認応答の処理専用です（acknowledge オプションが true の場合）。

destination-expression

SpEL 式

発信 UDP パケットの宛先アドレスとして使用する SocketAddress を決定するために評価される SpEL 式。

socket-expression

SpEL 式

発信 UDP パケットの送信に使用するデータグラムソケットを決定するために評価される SpEL 式。

次の表に、TCP 受信チャネルアダプターを構成するために設定できる属性を示します。

表 4: TCP 受信チャネルアダプターの属性
属性名 許容値 属性記述

channel

受信メッセージが送信されるチャネル。

connection-factory

接続ファクトリの型が server である場合、ファクトリはこのアダプターによって「所有」されます。型が client の場合、送信チャネルアダプターによって「所有」され、このアダプターは送信アダプターによって作成された接続で受信メッセージを受信します。

error-channel

ダウンストリームコンポーネントによって例外がスローされると、例外と失敗したメッセージを含む MessagingException メッセージがこのチャネルに送信されます。

client-mode

true, false

true の場合、受信アダプターは、接続の確立に関してクライアントとして機能し、その接続で受信メッセージを受信します。デフォルト: false retry-interval および scheduler も参照してください。接続ファクトリは、型 client であり、single-use を false に設定する必要があります。

retry-interval

client-mode の場合、接続試行間または接続障害後に待機するミリ秒数を指定します。デフォルト: 60000 (60 秒）。

scheduler

true, false

client-mode 接続の管理に使用する TaskScheduler を指定します。指定しない場合、デフォルトはグローバル Spring Integration taskScheduler Bean になり、デフォルトのプールサイズは 10 です。タスクスケジューラの構成を参照してください。

次の表に、TCP 送信チャネルアダプターを構成するために設定できる属性を示します。

表 5: TCP 送信チャネルアダプターの属性
属性名 許容値 属性記述

channel

送信メッセージが到着するチャネル。

connection-factory

接続ファクトリの型が client である場合、ファクトリはこのアダプターによって「所有」されます。型が server の場合、受信チャネルアダプターによって「所有」され、このアダプターは、元の受信メッセージが受信された接続にメッセージを関連付けようとします。

client-mode

true, false

true の場合、送信アダプターは、開始されるとすぐに接続を確立しようとします。false の場合、最初のメッセージが送信されたときに接続が確立されます。デフォルト: false retry-interval および scheduler も参照してください。接続ファクトリは型 client であり、single-use を false に設定する必要があります。

retry-interval

client-mode の場合、接続試行間または接続障害後に待機するミリ秒数を指定します。デフォルト: 60000 (60 秒）。

scheduler

true, false

client-mode 接続の管理に使用する TaskScheduler を指定します。指定しない場合、デフォルトはグローバル Spring Integration taskScheduler Bean になり、デフォルトのプールサイズは 10 です。タスクスケジューラの構成を参照してください。

次の表に、TCP 受信ゲートウェイを構成するために設定できる属性を示します。

表 6: TCP 受信ゲートウェイの属性
属性名 許容値 属性記述

connection-factory

接続ファクトリは型サーバーでなければなりません。

request-channel

受信メッセージが送信されるチャネル。

reply-channel

応答メッセージが到着するチャネル。通常、返信は受信メッセージヘッダーに追加された一時的な返信チャネルに到着します。

reply-timeout

ゲートウェイが応答を待機する時間（ミリ秒）。デフォルト: 1000 (1 秒）。

error-channel

ダウンストリームコンポーネントによって例外がスローされると、例外と失敗したメッセージを含む MessagingException メッセージがこのチャネルに送信されます。そのフローからのレスポンスは、ゲートウェイによってレスポンスとして返されます。

client-mode

true, false

true の場合、受信ゲートウェイは、接続の確立と、その接続での受信メッセージの受信（および応答）に関してクライアントとして機能します。デフォルト: false。retry-interval および scheduler も参照してください。接続ファクトリは型 client であり、single-use を false に設定する必要があります。

retry-interval

client-mode の場合、接続試行間または接続障害後に待機するミリ秒数を指定します。デフォルト: 60000 (60 秒）。

scheduler

true, false

client-mode 接続の管理に使用する TaskScheduler を指定します。指定しない場合、デフォルトはグローバル Spring Integration taskScheduler Bean になり、デフォルトのプールサイズは 10 です。タスクスケジューラの構成を参照してください。

次の表に、TCP 送信ゲートウェイを構成するために設定できる属性を示します。

表 7: TCP 送信ゲートウェイ属性
属性名 許容値 属性記述

connection-factory

接続ファクトリは、型 client でなければなりません。

request-channel

発信メッセージが到着するチャネル。

reply-channel

オプション。応答メッセージが送信されるチャネル。

remote-timeout

ゲートウェイがリモートシステムからの応答を待機する時間（ミリ秒）。remote-timeout-expression と相互に排他的です。デフォルト: 10000 (10 秒）。注: 4.2 より前のバージョンでは、この値のデフォルトは reply-timeout （設定されている場合）でした。

remote-timeout-expression

ゲートウェイがリモートシステムからの応答を待機する時間をミリ秒で決定するためにメッセージに対して評価される SpEL 式。remote-timeout と相互に排他的。

request-timeout

使い捨ての接続ファクトリが使用されていない場合、ゲートウェイが共有接続へのアクセスを待機する時間（ミリ秒単位）。

reply-timeout

応答チャネルに応答を送信するときにゲートウェイが待機する時間（ミリ秒）。応答チャネルがブロックする可能性がある場合にのみ適用されます（現在いっぱいの境界付き QueueChannel など）。

async

送信後に送信スレッドを解放します。応答（またはエラー）は受信スレッドで送信されます。

IP メッセージヘッダー

IP メッセージヘッダー

このモジュールは、次の MessageHeader インスタンスを使用します。

ヘッダー名 IpHeaders 定数 説明

ip_hostname

HOSTNAME

TCP メッセージまたは UDP パケットを受信したホスト名。lookupHost が false の場合、これには IP アドレスが含まれます。

ip_address

IP_ADDRESS

TCP メッセージまたは UDP パケットを受信した IP アドレス。

ip_port

PORT

UDP パケットのリモートポート。

ip_localInetAddress

IP_LOCAL_ADDRESS

ソケットが接続されているローカル InetAddress （バージョン 4.2.5 以降）。

ip_ackTo

ACKADDRESS

UDP アプリケーションレベルの確認応答が送信されるリモート IP アドレス。フレームワークは、データパケットに確認情報を含めます。

ip_ackId

ACK_ID

UDP アプリケーションレベルの確認応答の相関 ID。フレームワークは、データパケットに確認情報を含めます。

ip_tcp_remotePort

REMOTE_PORT

TCP 接続のリモートポート。

ip_connectionId

CONNECTION_ID

TCP 接続の一意の識別子。受信メッセージのフレームワークによって設定されます。サーバー側の受信チャネルアダプターに送信する場合、または受信ゲートウェイに応答する場合、エンドポイントがメッセージの送信先の接続を決定できるように、このヘッダーが必要です。

ip_actualConnectionId

ACTUAL_CONNECTION_ID

情報のみ。キャッシュまたはフェールオーバークライアント接続ファクトリを使用する場合、実際の基になる接続 ID が含まれます。

contentType

MessageHeaders. CONTENT_TYPE

受信メッセージのオプションのコンテンツ型この表の後に説明します。他のヘッダー定数とは異なり、この定数は IpHeaders クラスではなく MessageHeaders クラスにあることに注意してください。

受信メッセージの場合、ip_hostname、ip_address、ip_tcp_remotePort、ip_connectionId はデフォルトの TcpHeaderMapper によってマップされます。マッパーの addContentTypeHeader プロパティを true に設定すると、マッパーは contentType ヘッダー（デフォルトでは application/octet-stream;charset="UTF-8"）を設定します。contentType プロパティを設定することにより、デフォルトを変更できます。TcpHeaderMapper をサブクラス化し、supplyCustomHeaders メソッドをオーバーライドすることで、ヘッダーを追加できます。例: SSL を使用する場合、supplyCustomHeaders メソッドの引数として提供される TcpConnection オブジェクトからセッションオブジェクトを取得することにより、SSLSession のプロパティを追加できます。

送信・メッセージの場合、String ペイロードはデフォルト（UTF-8）文字セットで byte[] に変換されます。charset プロパティを設定してデフォルトを変更します。

マッパープロパティまたはサブクラスをカスタマイズする場合、マッパーを Bean として宣言し、mapper プロパティを使用して接続ファクトリにインスタンスを提供します。

アノテーションベースの構成

サンプルリポジトリの次の例は、XML の代わりにアノテーションを使用する場合に使用できる構成オプションの一部を示しています。

@EnableIntegration (1)
@IntegrationComponentScan (2)
@Configuration
public static class Config {

    @Value(${some.port})
    private int port;

    @MessagingGateway(defaultRequestChannel="toTcp") (3)
    public interface Gateway {

        String viaTcp(String in);

    }

    @Bean
    @ServiceActivator(inputChannel="toTcp") (4)
    public MessageHandler tcpOutGate(AbstractClientConnectionFactory connectionFactory) {
        TcpOutboundGateway gate = new TcpOutboundGateway();
        gate.setConnectionFactory(connectionFactory);
        gate.setOutputChannelName("resultToString");
        return gate;
    }

    @Bean (5)
    public TcpInboundGateway tcpInGate(AbstractServerConnectionFactory connectionFactory)  {
        TcpInboundGateway inGate = new TcpInboundGateway();
        inGate.setConnectionFactory(connectionFactory);
        inGate.setRequestChannel(fromTcp());
        return inGate;
    }

    @Bean
    public MessageChannel fromTcp() {
        return new DirectChannel();
    }

    @MessageEndpoint
    public static class Echo { (6)

        @Transformer(inputChannel="fromTcp", outputChannel="toEcho")
        public String convert(byte[] bytes) {
            return new String(bytes);
        }

        @ServiceActivator(inputChannel="toEcho")
        public String upCase(String in) {
            return in.toUpperCase();
        }

        @Transformer(inputChannel="resultToString")
        public String convertResult(byte[] bytes) {
            return new String(bytes);
        }

    }

    @Bean
    public AbstractClientConnectionFactory clientCF() { (7)
        return new TcpNetClientConnectionFactory("localhost", this.port);
    }

    @Bean
    public AbstractServerConnectionFactory serverCF() { (8)
        return new TcpNetServerConnectionFactory(this.port);
    }

}
1 統合アプリケーションのインフラストラクチャーを可能にする標準 Spring Integration アノテーション。
2@MessagingGateway インターフェースを検索します。
3 フローのクライアント側へのエントリポイント。呼び出し側アプリケーションは、この Gateway Bean に @Autowired を使用し、そのメソッドを呼び出すことができます。
4 送信エンドポイントは、MessageHandler とそれをラップするコンシューマーで構成されます。このシナリオでは、@ServiceActivator はチャネル型に応じてエンドポイントを構成します。
5（TCP/UDP モジュールの）受信エンドポイントはすべてメッセージ駆動型であるため、単純な @Bean インスタンスとしてのみ宣言する必要があります。
6 このクラスは、このサンプルフローで使用する多くの POJO メソッドを提供します（サーバー側の @Transformer および @ServiceActivator、クライアント側の @Transformer）。
7 クライアント側の接続ファクトリ。
8 サーバー側の接続ファクトリ。