メッセージ

Spring Integration Message は、データの汎用コンテナーです。任意のオブジェクトをペイロードとして提供でき、各 Message インスタンスには、キーと値のペアとしてユーザー拡張可能なプロパティを含むヘッダーが含まれます。

Message インターフェース

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

public interface Message<T> {

    T getPayload();

    MessageHeaders getHeaders();

}

Message インターフェースは、API の中核部分です。データを汎用ラッパーにカプセル化することにより、メッセージングシステムはデータの種類を知らなくてもデータを渡すことができます。アプリケーションが新しい型をサポートするように進化するとき、または型自体が変更または拡張されるときに、メッセージングシステムは影響を受けません。一方、メッセージングシステムの一部のコンポーネントが Message に関する情報へのアクセスを必要とする場合、そのようなメタデータは通常、メッセージヘッダーのメタデータに格納され、そこから取得できます。

メッセージヘッダー

Spring Integration が Object を Message のペイロードとして使用できるように、ヘッダー値として Object 型もサポートします。実際、次のクラス定義が示すように、MessageHeaders クラスは java.util.Map_ interface を実装します。

public final class MessageHeaders implements Map<String, Object>, Serializable {
  ...
}
MessageHeaders クラスは Map を実装していますが、事実上読み取り専用の実装です。マップ内の値を put しようとすると、UnsupportedOperationException になります。同じことが remove と clear にも当てはまります。メッセージは複数のコンシューマーに渡される可能性があるため、Map の構造は変更できません。同様に、最初の作成後、メッセージのペイロード Object を set にすることはできません。ただし、ヘッダー値自体（またはペイロードオブジェクト）の可変性は、フレームワークユーザーの意思決定として意図的に残されています。

Map の実装として、ヘッダーの名前で get(..) を呼び出すことでヘッダーを取得できます。あるいは、予想される Class を追加パラメーターとして提供できます。さらに良いことに、定義済みの値の 1 つを取得する場合、便利な getter を使用できます。次の例は、これら 3 つのオプションのそれぞれを示しています。

Object someValue = message.getHeaders().get("someKey");

CustomerId customerId = message.getHeaders().get("customerId", CustomerId.class);

Long timestamp = message.getHeaders().getTimestamp();

次の表で、事前定義されたメッセージヘッダーについて説明します。

表 1: 定義済みのメッセージヘッダー
ヘッダー名 ヘッダー型 使用方法 
 MessageHeaders.ID
 java.util.UUID

このメッセージインスタンスの識別子。メッセージが変更されるたびに変更されます。

 MessageHeaders.
TIMESTAMP
 java.lang.Long

メッセージが作成された時刻。メッセージが変更されるたびに変更されます。

 MessageHeaders.
REPLY_CHANNEL
 java.lang.Object
(String or
MessageChannel)

明示的な出力チャネルが構成されておらず、ROUTING_SLIP がない場合、ROUTING_SLIP が使い果たされた場合に、応答（ある場合）が送信されるチャネル。値が String の場合、Bean 名を表すか、ChannelRegistry. によって生成されている必要があります 

 MessageHeaders.
ERROR_CHANNEL
 java.lang.Object
(String or
MessageChannel)

エラーの送信先のチャネル。値が String の場合、Bean 名を表すか、ChannelRegistry. によって生成されている必要があります

多くの受信および送信アダプターの実装も特定のヘッダーを提供または期待しており、追加のユーザー定義ヘッダーを構成できます。これらのヘッダーの定数は、そのようなヘッダーが存在するモジュールで見つけることができます。たとえば。AmqpHeaders、JmsHeaders など。

MessageHeaderAccessor API

Spring Framework 4.0 と Spring Integration 4.0 から、コアメッセージングの抽象化は spring-messaging モジュールに移動され、メッセージング実装の抽象化を追加するために MessageHeaderAccessor API が導入されました。すべての（コア）Spring 統合固有のメッセージヘッダー定数が IntegrationMessageHeaderAccessor クラスで宣言されるようになりました。次の表で、事前定義されたメッセージヘッダーについて説明します。

表 2: 定義済みのメッセージヘッダー
ヘッダー名 ヘッダー型 使用方法 
 IntegrationMessageHeaderAccessor.
CORRELATION_ID
 java.lang.Object

2 つ以上のメッセージを相互に関連付けるために使用されます。

 IntegrationMessageHeaderAccessor.
SEQUENCE_NUMBER
 java.lang.Integer

通常、SEQUENCE_SIZE のメッセージのグループを含むシーケンス番号ですが、<resequencer/> で使用して、メッセージの無制限のグループを再シーケンスすることもできます。

 IntegrationMessageHeaderAccessor.
SEQUENCE_SIZE
 java.lang.Integer

相関メッセージのグループ内のメッセージの数。

 IntegrationMessageHeaderAccessor.
EXPIRATION_DATE
 java.lang.Long

メッセージの有効期限が切れたことを示します。フレームワークでは直接使用されませんが、ヘッダーエンリッチャーで設定し、UnexpiredMessageSelector で構成された <filter/> で使用できます。

 IntegrationMessageHeaderAccessor.
PRIORITY
 java.lang.Integer

メッセージの優先度 — たとえば、PriorityChannel 内。

 IntegrationMessageHeaderAccessor.
DUPLICATE_MESSAGE
 java.lang.Boolean

メッセージが dem 等受信インターセプターによって重複として検出された場合は真。べき等レシーバーエンタープライズ統合パターンを参照してください。

 IntegrationMessageHeaderAccessor.
CLOSEABLE_RESOURCE
 java.io.Closeable

このヘッダーは、メッセージ処理が完了したときに閉じる必要がある Closeable にメッセージが関連付けられている場合に存在します。例として、FTP、SFTP などを使用したストリーミングファイル転送に関連付けられた Session があります。

 IntegrationMessageHeaderAccessor.
DELIVERY_ATTEMPT
 java.lang.
AtomicInteger

メッセージ駆動型チャネルアダプターが RetryTemplate の構成をサポートしている場合、このヘッダーには現在の配信試行が含まれます。

 IntegrationMessageHeaderAccessor.
ACKNOWLEDGMENT_CALLBACK
 o.s.i.support.
Acknowledgment
Callback

受信エンドポイントがサポートしている場合は、コールバックしてメッセージを受け入れる、拒否する、キューに再登録します。遅延確認応答可能なメッセージソースおよび MQTT 手動確認を参照してください。

次の例に示すように、これらのヘッダーの一部に便利な型付き getter が IntegrationMessageHeaderAccessor クラスで提供されます。

IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor(message);
int sequenceNumber = accessor.getSequenceNumber();
Object correlationId = accessor.getCorrelationId();
...

次の表では、IntegrationMessageHeaderAccessor にも表示されますが、一般にユーザーコードでは使用されないヘッダーについて説明します（つまり、一般に Spring Integration の内部部分で使用されます。ここに含まれているのは完全を期すためです）。

表 3: 定義済みのメッセージヘッダー
ヘッダー名 ヘッダー型 使用方法 
 IntegrationMessageHeaderAccessor.
SEQUENCE_DETAILS
 java.util.
List<List<Object>>

ネストされた相関が必要な場合に使用される相関データのスタック（splitter→…​→splitter→…​→aggregator→…​→aggregator など）。

 IntegrationMessageHeaderAccessor.
ROUTING_SLIP
 java.util.
Map<List<Object>, Integer>

ルーティングスリップを参照してください。

メッセージ ID 生成

アプリケーションを介してメッセージが遷移すると、メッセージが（たとえば、トランスフォーマーによって）変更されるたびに、新しいメッセージ ID が割り当てられます。メッセージ ID は UUID です。Spring Integration 3.0 以降、IS 生成に使用されるデフォルトの戦略は、以前の java.util.UUID.randomUUID() 実装よりも効率的です。毎回安全な乱数を作成する代わりに、安全な乱数シードに基づく単純な乱数を使用します。

アプリケーションコンテキストで org.springframework.util.IdGenerator を実装する Bean を宣言することにより、異なる UUID 生成戦略を選択できます。

クラスローダーで使用できる UUID 生成戦略は 1 つだけです。つまり、2 つ以上のアプリケーションコンテキストが同じクラスローダーで実行される場合、同じ戦略を共有します。コンテキストの 1 つが戦略を変更すると、すべてのコンテキストで使用されます。同じクラスローダー内の 2 つ以上のコンテキストが org.springframework.util.IdGenerator 型の Bean を宣言する場合、すべて同じクラスのインスタンスでなければなりません。そうでない場合、カスタム戦略を置換しようとするコンテキストは初期化に失敗します。戦略が同じであるが、パラメーター化されている場合、初期化される最初のコンテキストの戦略が使用されます。

デフォルトの戦略に加えて、2 つの追加の IdGenerators が提供されています。org.springframework.util.JdkIdGenerator は、以前の UUID.randomUUID() メカニズムを使用します。UUID が実際に必要でなく、単純な増分値で十分な場合は、o.s.i.support.IdGenerators.SimpleIncrementingIdGenerator を使用できます。

読み取り専用ヘッダー

MessageHeaders.ID および MessageHeaders.TIMESTAMP は読み取り専用ヘッダーであり、オーバーライドできません。

バージョン 4.3.2 以降、MessageBuilder は readOnlyHeaders(String…​ readOnlyHeaders) API を提供して、アップストリーム Message からコピーすべきではないヘッダーのリストをカスタマイズします。デフォルトでは、MessageHeaders.ID および MessageHeaders.TIMESTAMP のみが読み取り専用です。グローバル spring.integration.readOnly.headers プロパティ（グローバルプロパティを参照）は、フレームワークコンポーネント用に DefaultMessageBuilderFactory をカスタマイズするために提供されています。これは、ObjectToJsonTransformer による contentType などの一部のすぐに使用可能なヘッダーを取り込まないようにする場合に役立ちます（JSONTransformers を参照）。

MessageBuilder を使用して新しいメッセージを作成しようとすると、この種類のヘッダーは無視され、特定の INFO メッセージがログに出力されます。

バージョン 5.0 以降、メッセージングゲートウェイ、ヘッダーエンリッチャー、コンテンツエンリッチャー、ヘッダーフィルターでは、DefaultMessageBuilderFactory が使用されている場合に MessageHeaders.ID および MessageHeaders.TIMESTAMP ヘッダー名を構成できず、それらは BeanInitializationException をスローします。

ヘッダーの伝播

メッセージがメッセージ生成エンドポイント ( サービスアクティベーターなど) によって処理 (および変更) されると、通常、受信ヘッダーが送信メッセージに伝播されます。ただし、トランスフォーマーの場合は例外で、フレームワークに完全なメッセージが返されます。その場合、ユーザーコードは送信メッセージ全体を担当します。トランスフォーマーがペイロードのみを返す場合は、受信ヘッダーが伝播されます。また、ヘッダーは送信メッセージに存在しない場合にのみ伝播されるため、必要に応じてヘッダー値を変更できます。

バージョン 4.3.10 から、メッセージハンドラー（メッセージを変更して出力を生成する）を構成して、特定のヘッダーの伝搬を抑制することができます。コピーしたくないヘッダーを構成するには、MessageProducingMessageHandler 抽象クラスで setNotPropagatedHeaders() または addNotPropagatedHeaders() メソッドを呼び出します。

また、META-INF/spring.integration.properties の readOnlyHeaders プロパティをヘッダーのコンマ区切りリストに設定することにより、特定のメッセージヘッダーの伝播をグローバルに抑制することができます。

バージョン 5.0 以降、AbstractMessageProducingHandler での setNotPropagatedHeaders() 実装は、単純なパターン（xxx*、*xxx、*xxx*、または xxx*yyy）を適用して、共通の接尾部または接頭部を持つヘッダーをフィルタリングできるようにします。詳細については、PatternMatchUtils Javadoc を参照してください。パターンの 1 つが * （アスタリスク）の場合、ヘッダーは伝搬されません。他のすべてのパターンは無視されます。その場合、サービスアクティベーターはトランスフォーマーと同じように動作し、必要なヘッダーはサービスメソッドから返される Message で提供する必要があります。notPropagatedHeaders() オプションは、Java DSL の ConsumerEndpointSpec で使用可能です。not-propagated-headers 属性として <service-activator> コンポーネントの XML 構成にも使用可能です。

ヘッダー伝播の抑制は、ブリッジやルーターなど、メッセージを変更しないエンドポイントには適用されません。

メッセージの実装

Message インターフェースの基本実装は GenericMessage<T> であり、次のリストに示す 2 つのコンストラクターを提供します。

new GenericMessage<T>(T payload);

new GenericMessage<T>(T payload, Map<String, Object> headers)

Message が作成されると、ランダムな一意の ID が生成されます。Map のヘッダーを受け入れるコンストラクターは、提供されたヘッダーを新しく作成された Message にコピーします。

エラー状態を伝えるために設計された Message の便利な実装もあります。次の例に示すように、この実装は Throwable オブジェクトをペイロードとして受け取ります。

ErrorMessage message = new ErrorMessage(someThrowable);

Throwable t = message.getPayload();

この実装は、GenericMessage 基本クラスがパラメーター化されているという事実を利用していることに注意してください。両方の例に示すように、Message ペイロード Object を取得するときにキャストは必要ありません。

前述の Message クラスの実装は不変です。可変性が問題にならず、アプリケーションのロジックが同時変更を回避するように適切に設計されている場合は、MutableMessage を使用できます。

MessageBuilder ヘルパークラス

Message インターフェースはペイロードとヘッダーの取得方法を定義しているが、setter を提供していないことに気付くかもしれません。この理由は、Message は最初の作成後に変更できないためです。Message インスタンスが複数のコンシューマーに（たとえば、パブリッシュ / サブスクライブチャネルを介して）送信される場合、それらのコンシューマーの 1 つが異なるペイロード型で応答を送信する必要がある場合、新しい Message を作成する必要があります。その結果、他のコンシューマーはこれらの変更の影響を受けません。複数のコンシューマーが同じペイロードインスタンスまたはヘッダー値にアクセスする可能性があり、そのようなインスタンス自体が不変であるかどうかはユーザーに委ねられていることに注意してください。言い換えれば、Message インスタンスの契約は変更不可能な Collection の契約に似ており、MessageHeaders マップはそれをさらに例示しています。MessageHeaders クラスは java.util.Map を実装しますが、MessageHeaders インスタンスで put 操作（または「削除」または「クリア」）を呼び出そうとすると、UnsupportedOperationException になります。

GenericMessage コンストラクターに渡すために Map の作成と移植を必要とするのではなく、Spring Integration は、Messages: MessageBuilder を構築するためのはるかに便利な方法を提供します。MessageBuilder は、既存の Message から、またはペイロード Object で Message インスタンスを作成するための 2 つのファクトリメソッドを提供します。既存の Message から構築する場合、次の例に示すように、その Message のヘッダーとペイロードが新しい Message にコピーされます。

Message<String> message1 = MessageBuilder.withPayload("test")
        .setHeader("foo", "bar")
        .build();

Message<String> message2 = MessageBuilder.fromMessage(message1).build();

assertEquals("test", message2.getPayload());
assertEquals("bar", message2.getHeaders().get("foo"));

新しいペイロードを使用して Message を作成する必要があるが、既存の Message からヘッダーをコピーしたい場合は、次の例に示すように、'copy' メソッドのいずれかを使用できます。

Message<String> message3 = MessageBuilder.withPayload("test3")
        .copyHeaders(message1.getHeaders())
        .build();

Message<String> message4 = MessageBuilder.withPayload("test4")
        .setHeader("foo", 123)
        .copyHeadersIfAbsent(message1.getHeaders())
        .build();

assertEquals("bar", message3.getHeaders().get("foo"));
assertEquals(123, message4.getHeaders().get("foo"));

copyHeadersIfAbsent メソッドは既存の値を上書きしないことに注意してください。また、前の例では、setHeader を使用してユーザー定義ヘッダーを設定する方法を確認できます。最後に、事前定義されたヘッダーに使用可能な set メソッドと、ヘッダーを設定する非破棄的なメソッドがあります（MessageHeaders は、事前定義されたヘッダー名の定数も定義します）。

次の例に示すように、MessageBuilder を使用してメッセージの優先度を設定することもできます。

Message<Integer> importantMessage = MessageBuilder.withPayload(99)
        .setPriority(5)
        .build();

assertEquals(5, importantMessage.getHeaders().getPriority());

Message<Integer> lessImportantMessage = MessageBuilder.fromMessage(importantMessage)
        .setHeaderIfAbsent(IntegrationMessageHeaderAccessor.PRIORITY, 2)
        .build();

assertEquals(2, lessImportantMessage.getHeaders().getPriority());

priority ヘッダーは、PriorityChannel を使用する場合にのみ考慮されます（次の章で説明します）。java.lang.Integer として定義されています。

MutableMessageBuilder は、MutableMessage インスタンスを処理するために提供されています。このクラスのロジックは、MutableMessage を作成するか、そのままにして、ビルダーメソッドを介してそのコンテンツを変更することです。この方法では、メッセージ交換の不変性が問題にならない場合、実行中のアプリケーションのパフォーマンスがわずかに向上します。

バージョン 6.4 以降では、デフォルトのメッセージ構築ロジックの拡張を簡素化するために、MessageBuilder から BaseMessageBuilder クラスが抽出されます。例: カスタム MessageBuilderFactory と一緒に、カスタム BaseMessageBuilder 実装をアプリケーションコンテキストでグローバルに使用して、カスタム Message インスタンスを提供できます。特に、GenericMessage.toString() メソッドをオーバーライドして、そのようなメッセージがログに記録されるときに、ペイロードとヘッダーから機密情報を非表示にすることができます。

MessageBuilderFactory 抽象化

IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME を含む MessageBuilderFactory Bean は、アプリケーションコンテキストにグローバルに登録され、フレームワークのあらゆる場所で Message インスタンスを作成するために使用されます。デフォルトでは、これは DefaultMessageBuilderFactory のインスタンスです。フレームワークは、フレームワークコンポーネントに MutableMessage インスタンスを作成するために、MutableMessageBuilderFactory もすぐに使用できます。Message インスタンスの作成をカスタマイズするには、IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME を含む MessageBuilderFactory Bean をターゲットアプリケーションコンテキストで提供して、デフォルトのインスタンスをオーバーライドする必要があります。例: BaseMessageBuilder の実装にカスタム MessageBuilderFactory を登録し、そのようなメッセージがログに記録されたときにペイロードとヘッダーから機密情報を非表示にするために、オーバーライドされた toString() を含む GenericMessage 拡張機能を提供することができます。

個人を特定できる情報の軽減を示すためにこれらのクラスを簡単に実装すると、次のようになります。

class PiiMessageBuilderFactory implements MessageBuilderFactory {

	@Override
	public <T> PiiMessageBuilder<T> fromMessage(Message<T> message) {
	    return new PiiMessageBuilder<>(message.getPayload(), message);
	}

	@Override
	public <T> PiiMessageBuilder<T> withPayload(T payload) {
	    return new PiiMessageBuilder<>(payload, null);
	}

}

class PiiMessageBuilder<P> extends BaseMessageBuilder<P, PiiMessageBuilder<P>> {

    public PiiMessageBuilder(P payload, @Nullable Message<P> originalMessage) {
        super(payload, originalMessage);
    }

    @Override
    public Message<P> build() {
        return new PiiMessage<>(getPayload(), getHeaders());
    }

}

class PiiMessage<P> extends GenericMessage<P> {

    @Serial
    private static final long serialVersionUID = -354503673433669578L;

    public PiiMessage(P payload, Map<String, Object> headers) {
        super(payload, headers);
    }

    @Override
    public String toString() {
        return "PiiMessage [payload=" + getPayload() + ", headers=" + maskHeaders(getHeaders()) + ']';
    }

    private static Map<String, Object> maskHeaders(Map<String, Object> headers) {
        return headers.entrySet()
                .stream()
                .map((entry) -> entry.getKey().equals("password") ? Map.entry(entry.getKey(), "******") : entry)
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

}

この PiiMessageBuilderFactory は Bean として登録することができ、フレームワークがメッセージをログに記録するたびに (たとえば、errorChannel の場合)、password ヘッダーがマスクされます。