[Spring Integration] @EventListenerを利用してTCPコネクションのイベントログを出力する

Spring Integrationの機能を利用してTCPサーバやTCPクライアントを構築する際に、サーバ起動の成功・失敗、コネクションの接続・切断・エラーなどをイベントログを出力方法を紹介したいと思います。

# デフォルトのログ

もちろん最低限のログ（起動、停止、エラーログなど）は出ますが、商用サービスにおけるログ監視などのことを考慮すると・・・デフォルトのログだけで運用するのは少し厳しいかもしれません。例えば、イベントを識別するためのメッセージID(イベントID)や障害時のエラー解析を円滑に行えるようにするための付加情報などを埋め込む必要があったりします。

# Spring Integration提供のEventクラス

Spring Integrationは、TCP通信関連の各種処理を行った結果をアプリケーションレイヤに通知するためのイベントクラスをいくつか用意しています。

| カテゴリ | イベントクラス | 契機 | 実装例の有無 |
| --- | --- | ---- | --- |
| サーバ向け | TcpConnectionServerListeningEvent | TCPサーバのポートを開いたタイミングでイベントが発生する | ○ |
| | TcpConnectionServerExceptionEvent | TCPサーバの起動や起動後にエラーを検知しタイミングでイベントが発生する | ○ |
| | UdpServerListeningEvent | TCPサーバのポートを開いたタイミングでイベントが発生する | × |
| クライアント向け | TcpConnectionFailedEvent | TCPサーバへの接続に失敗したタイミングでイベントが発生する | ○ |
| 共通 | TcpConnectionOpenEvent | ソケットを接続した際にイベントが発生する | ○ |
| | TcpConnectionCloseEvent | ソケットを切断した際にイベントが発生する | ○ |
| | TcpConnectionExceptionEvent | 接続後に何らかのエラーを検知した際にイベントが発生する | ○ |
| | TcpConnectionFailedCorrelationEvent | メッセージの送信が行えないまたは不正なタイミングでメッセージを受信した際にイベントが発生する | ○ |
| | TcpDeserializationExceptionEvent | メッセージのデシリアライズ時(＝読み込み中)に何かしらのエラーを検知した際に発生する | × |

# ログ出力用のコンポーネント

以下は、Spring Integrationが発行するイベントを受け取ってログを出力するコンポーネントの作成例です。

> **IMPORTANT**
>
> ここで紹介している実装例は利用する環境によっては正しく動作しない可能性があります。特に例外メッセージの中身を見て分岐している部分については、利用するJVMやJVMの設定に依存する可能性がありますので、必ずご利用の環境下で動作検証を行い、開発するシステムのログ要件にあった実装にするようにしてください！！

```java
package com.example.demo;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.event.EventListener;
import org.springframework.integration.ip.tcp.TcpInboundGateway;
import org.springframework.integration.ip.tcp.TcpOutboundGateway;
import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
import org.springframework.integration.ip.tcp.connection.*;
import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.Optional;

@Component
public class TcpConnectionLoggingEventListeners {

    private final Logger logger = LoggerFactory.getLogger(TcpConnectionLoggingEventListeners.class);

    @EventListener
    public void handleTcpConnectionServerListeningEvent(TcpConnectionServerListeningEvent event) {
        logger.info("TCPサーバを起動しました。ポート[{}]", event.getPort());
    }

    @EventListener
    public void handleTcpConnectionServerExceptionEvent(TcpConnectionServerExceptionEvent event) {
        AbstractServerConnectionFactory factory = (AbstractServerConnectionFactory) event.getSource();
        logger.error("TCPサーバの起動に失敗しました。ポート[{}]", factory.getPort());
    }

    @EventListener
    public void handleTcpConnectionOpenEvent(TcpConnectionOpenEvent event) {
        SocketInfo socketInfo = ((TcpConnection) event.getSource()).getSocketInfo();
        logger.info("接続しました。ローカル[{}] リモート[{}]", socketInfo.getLocalSocketAddress(), socketInfo.getRemoteSocketAddress());
    }

    @EventListener
    public void handleTcpConnectionFailedEvent(TcpConnectionFailedEvent event) {
        AbstractClientConnectionFactory factory = (AbstractClientConnectionFactory) event.getSource();
        logger.error("接続に失敗しました。ホスト[{}] ポート[{}]", factory.getHost(), factory.getPort());
    }

    @EventListener
    public void handleTcpConnectionCloseEvent(TcpConnectionCloseEvent event) {
        SocketInfo socketInfo = ((TcpConnection) event.getSource()).getSocketInfo();
        logger.info("切断しました。ローカル[{}] リモート[{}]", socketInfo.getLocalSocketAddress(),
                Optional.ofNullable(socketInfo.getRemoteSocketAddress()).map(Objects::toString).orElse("-"));
    }

    @EventListener
    public void handleTcpConnectionFailedEvent(TcpConnectionExceptionEvent event) {
        if (!((TcpConnection) event.getSource()).isOpen() && event.getCause().getMessage().toLowerCase().contains("close")) {
            // ローカルソケット切断後に発生する切断通知のエラーは無視(受信スレッド側で検知するケースあり)
            return;
        }

        SocketInfo socketInfo = ((TcpConnection) event.getSource()).getSocketInfo();
        if (event.getCause() instanceof SoftEndOfStreamException) {
            logger.info("正常切断(FIN)を検知しました。ローカル[{}] リモート:[{}]", socketInfo.getLocalSocketAddress(),
                    Optional.ofNullable(socketInfo.getRemoteSocketAddress()).map(Objects::toString).orElse("-"));
            return;
        }
        if (event.getCause().getMessage().toLowerCase().contains("reset")) {
            logger.info("強制切断(RST)を検知しました。ローカル[{}] リモート[{}]", socketInfo.getLocalSocketAddress(),
                    Optional.ofNullable(socketInfo.getRemoteSocketAddress()).map(Objects::toString).orElse("-"));
            return;
        }
        logger.error("異常を検知しました。ローカル[{}] リモート[{}]", socketInfo.getLocalSocketAddress(),
                Optional.ofNullable(socketInfo.getRemoteSocketAddress()).map(Objects::toString).orElse("-"), event.getCause());
    }

    @EventListener
    public void handleTcpConnectionFailedCorrelationEvent(TcpConnectionFailedCorrelationEvent event) {
        String connectionId = Optional.ofNullable(event.getConnectionId()).orElse("-");
        if (event.getSource() instanceof TcpInboundGateway) {
            logger.error("有効な接続が存在しないためメッセージを応答できません。コネクションID[{}]", connectionId);
            return;
        }
        if (event.getSource() instanceof TcpSendingMessageHandler) {
            logger.error("有効な接続が存在しないためメッセージを送信できません。コネクションID[{}]", connectionId);
            return;
        }
        if (event.getSource() instanceof TcpOutboundGateway) {
            logger.warn("不正なタイミングでメッセージを受信したためメッセージを破棄します。コネクションID[{}]", connectionId);
            return;
        }
        logger.warn("処理を継続できないためメッセージを破棄します。コネクションID[{}]", connectionId);
    }

}
```


# まとめ

こんな感じでイベントをハンドリングすることで任意の処理を加えることができます。今回はログ出力を題材にしましたが、例えば・・・エラー検出時の再接続処理などをこの仕組みを利用して実装するのもアリだと思います。