MongoDb サポート
バージョン 2.1 は MongoDB (英語) のサポートを導入しました: 「高性能、オープンソース、ドキュメント指向データベース」。
この依存関係はプロジェクトに必要です:
MongoDB をダウンロード、インストール、実行するには、MongoDB ドキュメント (英語) を参照してください。
MongoDb への接続
ブロッキングまたは反応？
5.3 バージョン以降、Spring Integration は、MongoDB へのアクセス時にノンブロッキング I/O を有効にするために、リアクティブ MongoDB ドライバーをサポートしています。リアクティブサポートを有効にするには、MongoDB リアクティブストリームドライバーを依存関係に追加します。
Maven
Gradle
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongodb-driver-reactivestreams</artifactId>
</dependency>implementation "org.mongodb:mongodb-driver-reactivestreams"通常の同期クライアントの場合は、それぞれのドライバーを依存関係に追加する必要があります。
Maven
Gradle
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongodb-driver-sync</artifactId>
</dependency>implementation "org.mongodb:mongodb-driver-sync" どちらも optional であり、エンドユーザーの選択サポートを強化するためのフレームワークです。
MongoDB との対話を開始するには、まずそれに接続する必要があります。Spring Integration は、別の Spring プロジェクト Spring Data MongoDB によって提供されるサポートに基づいて構築されています。これは、MongoDatabaseFactory および ReactiveMongoDatabaseFactory と呼ばれるファクトリクラスを提供し、MongoDB クライアント API との統合を簡素化します。
| Spring Data はデフォルトでブロッキング MongoDB ドライバーを提供しますが、上記の依存関係を含めることでリアクティブな使用を選択できます。 |
MongoDatabaseFactory を使用する
MongoDB に接続するには、MongoDatabaseFactory インターフェースの実装を使用できます。
次の例は、SimpleMongoClientDatabaseFactory の使用方法を示しています。
Java
XML
MongoDatabaseFactory mongoDbFactory =
new SimpleMongoClientDatabaseFactory(com.mongodb.client.MongoClients.create(), "test");<bean id="mongoDbFactory" class="o.s.data.mongodb.core.SimpleMongoClientDatabaseFactory">
<constructor-arg>
<bean class="com.mongodb.client.MongoClients" factory-method="create"/>
</constructor-arg>
<constructor-arg value="test"/>
</bean>SimpleMongoClientDatabaseFactory は 2 つの引数を取ります: MongoClient インスタンスとデータベースの名前を指定する String。host、port などのプロパティを構成する必要がある場合は、基になる MongoClients クラスによって提供されるコンストラクターの 1 つを使用して渡すことができます。MongoDB の構成方法の詳細については、Spring-Data-MongoDB リファレンスを参照してください。
ReactiveMongoDatabaseFactory を使用する
リアクティブドライバーを使用して MongoDB に接続するには、ReactiveMongoDatabaseFactory インターフェースの実装を使用できます。
次の例は、SimpleReactiveMongoDatabaseFactory の使用方法を示しています。
Java
XML
ReactiveMongoDatabaseFactory mongoDbFactory =
new SimpleReactiveMongoDatabaseFactory(com.mongodb.reactivestreams.client.MongoClients.create(), "test");<bean id="mongoDbFactory" class="o.s.data.mongodb.core.SimpleReactiveMongoDatabaseFactory">
<constructor-arg>
<bean class="com.mongodb.reactivestreams.client.MongoClients" factory-method="create"/>
</constructor-arg>
<constructor-arg value="test"/>
</bean>MongoDB メッセージストア
エンタープライズ統合パターン（EIP）ブックに従って、メッセージストア (英語) を使用するとメッセージを保持できます。信頼性が懸念される場合、メッセージをバッファリングする機能（QueueChannel、aggregator、resequencer など）を備えたコンポーネントを扱う場合、これは便利です。Spring Integration では、MessageStore 戦略はクレームチェック (英語) パターンの基盤も提供します。これは EIP でも説明されています。
Spring Integration の MongoDB モジュールは、MongoDbMessageStore を提供します。これは、MessageStore 戦略（主にクレームチェックパターンで使用）と MessageGroupStore 戦略（主にアグリゲーターパターンおよびリシーケンサパターンで使用）の両方の実装です。
次の例では、MongoDbMessageStore が QueueChannel と aggregator を使用するように構成しています。
<bean id="mongoDbMessageStore" class="o.s.i.mongodb.store.MongoDbMessageStore">
<constructor-arg ref="mongoDbFactory"/>
</bean>
<int:channel id="somePersistentQueueChannel">
<int:queue message-store="mongoDbMessageStore"/>
<int:channel>
<int:aggregator input-channel="inputChannel" output-channel="outputChannel"
message-store="mongoDbMessageStore"/> 上記の例は単純な Bean 構成であり、コンストラクター引数として MongoDbFactory を想定しています。
MongoDbMessageStore は、Spring Data Mongo マッピングメカニズムを使用して、ネストされたすべてのプロパティを含む Mongo ドキュメントとして Message を拡張します。監査または分析のために payload または headers にアクセスする必要がある場合（たとえば、保存されているメッセージに対して）に役立ちます。
MongoDbMessageStore は、カスタム MappingMongoConverter 実装を使用して Message インスタンスを MongoDB ドキュメントとして格納します。Message のプロパティ（payload および header 値）にはいくつかの制限があります。 |
バージョン 5.1.6 以降、MongoDbMessageStore は、内部 MappingMongoConverter 実装に伝搬されるカスタムコンバーターで構成できます。詳細については、MongoDbMessageStore.setCustomConverters(Object… customConverters) JavaDocs を参照してください。
Spring Integration 3.0 は ConfigurableMongoDbMessageStore を導入しました。MessageStore と MessageGroupStore の両方のインターフェースを実装しています。このクラスは、コンストラクター引数として MongoTemplate を受け取ることができます。これを使用して、たとえば、カスタム WriteConcern を構成できます。別のコンストラクターには MappingMongoConverter と MongoDbFactory が必要です。これにより、Message インスタンスとそのプロパティにカスタム変換を提供できます。デフォルトでは、ConfigurableMongoDbMessageStore は標準の Java シリアライゼーションを使用して、Message インスタンスを MongoDB に読み書きし（MongoDbMessageBytesConverter を参照）、MongoTemplate の他のプロパティのデフォルト値に依存することに注意してください。提供されている MongoDbFactory および MappingMongoConverter から MongoTemplate を構築します。ConfigurableMongoDbMessageStore によって格納されるコレクションのデフォルト名は configurableStoreMessages です。メッセージに複雑なデータ型が含まれている場合、この実装を使用して堅牢で柔軟なソリューションを作成することをお勧めします。
バージョン 6.0.8 以降、AbstractConfigurableMongoDbMessageStore には、自動インデックス作成を無効にするために使用できる setCreateIndexes(boolean) (デフォルトは true) オプションが用意されています。次の例は、Bean を宣言し、自動インデックス作成を無効にする方法を示しています。
@Bean
public MongoDbChannelMessageStore mongoDbChannelMessageStore(MongoDatabaseFactory databaseFactory) {
MongoDbChannelMessageStore mongoDbChannelMessageStore = new MongoDbChannelMessageStore(databaseFactory);
mongoDbChannelMessageStore.setCreateIndexes(false);
return mongoDbChannelMessageStore;
}MongoDB チャネルメッセージストア
バージョン 4.0 は、新しい MongoDbChannelMessageStore を導入しました。QueueChannel インスタンスで使用するために最適化された MessageGroupStore です。priorityEnabled = true を使用すると、<int:priority-queue> インスタンスでそれを使用して、持続メッセージの優先順位ポーリングを実現できます。優先度 MongoDB ドキュメントフィールドは、IntegrationMessageHeaderAccessor.PRIORITY （priority）メッセージヘッダーから入力されます。
さらに、すべての MongoDB MessageStore インスタンスには、MessageGroup ドキュメント用の sequence フィールドがあります。sequence 値は、同じコレクションからの単純な sequence ドキュメントに対する $inc 操作の結果であり、要求に応じて作成されます。sequence フィールドは poll 操作で使用され、メッセージが同じミリ秒以内に保存される場合、先入れ先出し（FIFO）メッセージの順序（設定されている場合は優先順位内）を提供します。
priorityEnabled オプションはストア全体に適用されるため、優先度と非優先度に同じ MongoDbChannelMessageStore Bean を使用することはお勧めしません。ただし、ストアからのメッセージポーリングはソートされ、インデックスを使用するため、同じ collection を両方の MongoDbChannelMessageStore 型に使用できます。そのシナリオを構成するには、次の例に示すように、一方のメッセージストア Bean を他方から拡張できます。 |
<bean id="channelStore" class="o.s.i.mongodb.store.MongoDbChannelMessageStore">
<constructor-arg name="mongoDbFactory" ref="mongoDbFactory"/>
</bean>
<int:channel id="queueChannel">
<int:queue message-store="store"/>
</int:channel>
<bean id="priorityStore" parent="channelStore">
<property name="priorityEnabled" value="true"/>
</bean>
<int:channel id="priorityChannel">
<int:priority-queue message-store="priorityStore"/>
</int:channel>自動インデックス作成を無効にして AbstractConfigurableMongoDbMessageStore を使用する
バージョン 6.0.8 以降、AbstractConfigurableMongoDbMessageStore は setCreateIndex(boolean) を実装しており、これを使用して自動インデックス作成を無効化または有効化（デフォルト）できます。次の例は、Bean を宣言し、自動インデックス作成を無効にする方法を示しています。
@Bean
public AbstractConfigurableMongoDbMessageStore mongoDbChannelMessageStore(MongoDatabaseFactory databaseFactory)
{
AbstractConfigurableMongoDbMessageStore mongoDbChannelMessageStore = new MongoDbChannelMessageStore(databaseFactory);
mongoDbChannelMessageStore.setCreateIndex(false);
return mongoDbChannelMessageStore;
}MongoDB メタデータストア
Spring Integration 4.2 は、MongoDB ベースの新しい MetadataStore （メタデータストアを参照）実装を導入しました。MongoDbMetadataStore を使用して、アプリケーションの再起動後もメタデータの状態を維持できます。この新しい MetadataStore 実装は、次のようなアダプターで使用できます。
これらのアダプターに新しい MongoDbMetadataStore を使用するように指示するには、Bean 名が metadataStore の Spring Bean を宣言します。フィード受信チャネルアダプターは、宣言された MongoDbMetadataStore を自動的に取得して使用します。次の例は、metadataStore という名前の Bean を宣言する方法を示しています。
@Bean
public MetadataStore metadataStore(MongoDbFactory factory) {
return new MongoDbMetadataStore(factory, "integrationMetadataStore");
}MongoDbMetadataStore は ConcurrentMetadataStore も実装しており、キーの値を保存または変更できるインスタンスは 1 つだけであるため、複数のアプリケーションインスタンス間で確実に共有できます。MongoDB の保証により、これらの操作はすべてアトミックです。
MongoDB 受信チャネルアダプター
MongoDB 受信チャネルアダプターは、MongoDB からデータを読み取り、Message ペイロードとして送信するポーリングコンシューマーです。次の例は、MongoDB 受信チャネルアダプターを構成する方法を示しています。
<int-mongodb:inbound-channel-adapter id="mongoInboundAdapter"
channel="replyChannel"
query="{'name' : 'Bob'}"
entity-class="java.lang.Object"
auto-startup="false">
<int:poller fixed-rate="100"/>
</int-mongodb:inbound-channel-adapter> 上記の構成が示すように、inbound-channel-adapter 要素を使用し、次のようなさまざまな属性の値を提供することにより、MongoDb 受信チャネルアダプターを構成します。
query: JSON クエリ (MongoDB クエリ (英語) を参照してください)query-expression: JSON クエリ文字列（上記のquery属性として）またはo.s.data.mongodb.core.query.Queryのインスタンスに評価される SpEL 式。query属性と相互に排他的。entity-class: ペイロードオブジェクトの型。指定しない場合、com.mongodb.DBObjectが返されます。collection-nameまたはcollection-name-expression: 使用する MongoDB コレクションの名前を識別します。mongodb-factory:o.s.data.mongodb.MongoDbFactoryのインスタンスへの参照mongo-template:o.s.data.mongodb.core.MongoTemplateのインスタンスへの参照他のすべての受信アダプターに共通するその他の属性 (「チャネル」など)。
mongo-template と mongodb-factory の両方を設定することはできません。 |
上記の例は、query のリテラル値を持ち、collection のデフォルト名を使用するため、比較的単純で静的です。条件によっては、実行時にこれらの値を変更する必要がある場合があります。そのためには、-expression の同等物（query-expression および collection-name-expression）を使用します。ここで、提供された式は有効な SpEL 式です。
また、MongoDB から読み込んだ処理済みのデータに対して、何らかの後処理を行いたい場合もあるでしょう。たとえば、処理済みのドキュメントを移動または削除したい場合などです。これは、Spring Integration 2.2 に追加されたトランザクション同期機能を使用することで実現できます。以下の例を参照してください。
<int-mongodb:inbound-channel-adapter id="mongoInboundAdapter"
channel="replyChannel"
query-expression="new BasicQuery('{''name'' : ''Bob''}').limit(100)"
entity-class="java.lang.Object"
auto-startup="false">
<int:poller fixed-rate="200" max-messages-per-poll="1">
<int:transactional synchronization-factory="syncFactory"/>
</int:poller>
</int-mongodb:inbound-channel-adapter>
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit
expression="@documentCleaner.remove(#mongoTemplate, payload, headers.mongo_collectionName)"
channel="someChannel"/>
</int:transaction-synchronization-factory>
<bean id="documentCleaner" class="thing1.thing2.DocumentCleaner"/>
<bean id="transactionManager" class="o.s.i.transaction.PseudoTransactionManager"/> 次の例は、前述の例で参照されている DocumentCleaner を示しています。
public class DocumentCleaner {
public void remove(MongoOperations mongoOperations, Object target, String collectionName) {
if (target instanceof List<?> documents){
for (Object document : documents) {
mongoOperations.remove(new BasicQuery(JSON.serialize(document)), collectionName);
}
}
}
}transactional 要素を使用することで、ポーラーをトランザクション対応として宣言できます。この要素は、たとえばフロー内の他の部分で JDBC を呼び出す場合など、実際のトランザクションマネージャーを参照できます。「実際の」トランザクションがない場合は、Spring の PlatformTransactionManager の実装である o.s.i.transaction.PseudoTransactionManager のインスタンスを使用できます。これにより、実際のトランザクションがない場合でも、Mongo アダプターのトランザクション同期機能を使用できます。
| これを行っても、MongoDB 自体はトランザクションになりません。成功（コミット）または失敗（ロールバック）の前後にアクションの同期を取ることができます。 |
ポーラーがトランザクション対応になったら、transactional 要素に o.s.i.transaction.TransactionSynchronizationFactory のインスタンスを設定できます。TransactionSynchronizationFactory は、TransactionSynchronization のインスタンスを作成します。便宜上、デフォルトの SpEL ベースの TransactionSynchronizationFactory を公開しています。これにより SpEL 式を設定でき、その実行はトランザクションと調整 (同期) されます。コミット前、コミット後、ロールバック後のイベントの式がサポートされており、評価結果 (ある場合) が送信される各イベントのチャネルもサポートされています。子要素ごとに、expression 属性と channel 属性を指定できます。channel 属性のみが存在する場合、受信メッセージは特定の同期シナリオの一部としてそこに送信されます。expression 属性のみが存在し、式の結果が null 以外の値である場合は、結果をペイロードとするメッセージが生成され、デフォルトチャネル (NullChannel) に送信されて、ログ ( DEBUG レベル) に表示されます。評価結果を特定のチャネルに送信したい場合は、channel 属性を追加します。式の結果が null または void の場合、メッセージは生成されません。
トランザクション同期の詳細については、トランザクションの同期を参照してください。
バージョン 5.5 以降、MongoDbMessageSource は updateExpression で構成できます。updateExpression は、MongoDb update 構文を持つ String、または org.springframework.data.mongodb.core.query.Update インスタンスに評価される必要があります。これは、上記の後処理手順を記述するための代替手段として使用でき、コレクションから取得されたエンティティを変更します。これにより、次回のポーリングサイクルでコレクションから再度取得されることはありません（更新によってクエリで使用される値が変更されると仮定）。同じコレクションに対してクラスター内で複数の MongoDbMessageSource インスタンスを使用する場合は、実行の分離とデータの一貫性を確保するために、トランザクションを使用することをお勧めします。
MongoDB 変更ストリーム受信チャネルアダプター
バージョン 5.3 以降、spring-integration-mongodb モジュールに MongoDbChangeStreamMessageProducer が導入されました。これは、Spring Data ReactiveMongoOperations.changeStream(String, ChangeStreamOptions, Class) API のリアクティブ MessageProducerSupport 実装です。このコンポーネントは、デフォルトで ChangeStreamEvent の body をペイロードとし、変更ストリーム関連ヘッダー（MongoHeaders を参照）を含む Flux メッセージを生成します。この MongoDbChangeStreamMessageProducer は、オンデマンドサブスクリプションと下流のイベント消費のために、FluxMessageChannel と組み合わせて outputChannel として利用することをお勧めします。
このチャネルアダプターの Java DSL 構成は次のようになります。
@Bean
IntegrationFlow changeStreamFlow(ReactiveMongoOperations mongoTemplate) {
return IntegrationFlow.from(
MongoDb.changeStreamInboundChannelAdapter(mongoTemplate)
.domainType(Person.class)
.collection("person")
.extractBody(false))
.channel(MessageChannels.flux())
.get();
}MongoDbChangeStreamMessageProducer が停止するか、下流でサブスクリプションがキャンセルされるか、MongoDb 変更ストリームが OperationType.INVALIDATE を生成すると、Publisher が完了します。チャネルアダプターは再起動でき、ソースデータの新しい Publisher が作成され、MessageProducerSupport.subscribeToPublisher(Publisher<? extends Message<?>>) に自動的にサブスクライブされます。他の場所から変更ストリームイベントを消費する必要がある場合、このチャネルアダプターは起動と起動の間に新しいオプションで再構成できます。
変更ストリームのサポートの詳細については、Spring Data MongoDB ドキュメントを参照してください。
MongoDB 送信チャネルアダプター
MongoDB 送信チャネルアダプターを使用すると、次の例に示すように、メッセージペイロードを MongoDB ドキュメントストアに書き込むことができます。
<int-mongodb:outbound-channel-adapter id="fullConfigWithCollectionExpression"
collection-name="myCollection"
mongo-converter="mongoConverter"
mongodb-factory="mongoDbFactory" /> 上記の構成が示すように、outbound-channel-adapter 要素を使用して MongoDB 送信チャネルアダプターを構成し、次のようなさまざまな属性の値を提供できます。
collection-nameまたはcollection-name-expression: 使用する MongoDb コレクションの名前を識別します。mongo-converter: 生の Java オブジェクトを JSON ドキュメント表現に変換するのを支援するo.s.data.mongodb.core.convert.MongoConverterのインスタンスへの参照。mongodb-factory:o.s.data.mongodb.MongoDbFactoryのインスタンスへの参照。mongo-template:o.s.data.mongodb.core.MongoTemplateのインスタンスへの参照。注意:mongo-templateとmongodb-factoryの両方を設定することはできません。すべての受信 アダプターに共通するその他の属性 (「チャネル」など)。
前の例は、collection-name のリテラル値を持っているため、比較的単純で静的です。状況によっては、実行時にこの値を変更する必要がある場合があります。これを行うには、collection-name-expression を使用します。指定された式は有効な SpEL 式です。
MongoDB 送信ゲートウェイ
バージョン 5.0 では、MongoDB 送信ゲートウェイが導入されました。これにより、リクエストチャネルにメッセージを送信することでデータベースをクエリできます。ゲートウェイはレスポンスをリプライチャネルに送信します。次の例に示すように、メッセージのペイロードとヘッダーを使用して、クエリとコレクション名を指定できます。
Java DSL
Kotlin DSL
Java
XML
@SpringBootApplication
public class MongoDbJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(MongoDbJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private MongoDbFactory;
@Autowired
private MongoConverter;
@Bean
public IntegrationFlow gatewaySingleQueryFlow() {
return f -> f
.handle(queryOutboundGateway())
.channel(c -> c.queue("retrieveResults"));
}
private MongoDbOutboundGatewaySpec queryOutboundGateway() {
return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
.query("{name : 'Bob'}")
.collectionNameFunction(m -> m.getHeaders().get("collection"))
.expectSingleResult(true)
.entityClass(Person.class);
}
}class MongoDbKotlinApplication {
fun main(args: Array<String>) = runApplication<MongoDbKotlinApplication>(*args)
@Autowired
lateinit var mongoDbFactory: MongoDatabaseFactory
@Autowired
lateinit var mongoConverter: MongoConverter
@Bean
fun gatewaySingleQueryFlow() =
integrationFlow {
handle(queryOutboundGateway())
channel { queue("retrieveResults") }
}
private fun queryOutboundGateway(): MongoDbOutboundGatewaySpec {
return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
.query("{name : 'Bob'}")
.collectionNameFunction<Any> { m -> m.headers["collection"] as String }
.expectSingleResult(true)
.entityClass(Person::class.java)
}
}@SpringBootApplication
public class MongoDbJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(MongoDbJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private MongoDbFactory mongoDbFactory;
@Bean
@ServiceActivator(inputChannel = "requestChannel")
public MessageHandler mongoDbOutboundGateway() {
MongoDbOutboundGateway gateway = new MongoDbOutboundGateway(this.mongoDbFactory);
gateway.setCollectionNameExpressionString("'myCollection'");
gateway.setQueryExpressionString("'{''name'' : ''Bob''}'");
gateway.setExpectSingleResult(true);
gateway.setEntityClass(Person.class);
gateway.setOutputChannelName("replyChannel");
return gateway;
}
@Bean
@ServiceActivator(inputChannel = "replyChannel")
public MessageHandler handler() {
return message -> System.out.println(message.getPayload());
}
}<int-mongodb:outbound-gateway id="gatewayQuery"
mongodb-factory="mongoDbFactory"
mongo-converter="mongoConverter"
query="{firstName: 'Bob'}"
collection-name="myCollection"
request-channel="in"
reply-channel="out"
entity-class="org.springframework.integration.mongodb.test.entity$Person"/>MongoDB 送信ゲートウェイで次の属性を使用できます。
collection-nameまたはcollection-name-expression: 使用する MongoDB コレクションの名前を識別します。mongo-converter: 生の Java オブジェクトを JSON ドキュメント表現に変換するのを支援するo.s.data.mongodb.core.convert.MongoConverterのインスタンスへの参照。mongodb-factory:o.s.data.mongodb.MongoDbFactoryのインスタンスへの参照。mongo-template:o.s.data.mongodb.core.MongoTemplateのインスタンスへの参照。注意:mongo-templateとmongodb-factoryの両方を設定することはできません。entity-class: MongoTemplate のfind(..)およびfindOne(..)メソッドに渡されるエンティティクラスの完全修飾名。この属性が提供されない場合、デフォルト値はorg.bson.Documentです。queryまたはquery-expression: MongoDB クエリを指定します。その他のクエリサンプルについては、MongoDB ドキュメント (英語) を参照してください。collection-callback:org.springframework.data.mongodb.core.CollectionCallbackのインスタンスへの参照。5.0.11 がリクエストメッセージコンテキストであるため、o.s.i.mongodb.outbound.MessageCollectionCallbackのインスタンスが好ましい。詳細については、Javadocs を参照してください。注意:collection-callbackといずれかのクエリ属性の両方を持つことはできません。
query および query-expression プロパティの代替として、MessageCollectionCallback 関数インターフェース実装への参照として collectionCallback プロパティを使用することにより、他のデータベース操作を指定できます。次の例では、カウント操作を指定しています。
private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway() {
return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
.collectionCallback((collection, requestMessage) -> collection.count())
.collectionName("myCollection");
}MongoDB Reactive Channel Adapters
バージョン 5.3 以降、ReactiveMongoDbStoringMessageHandler および ReactiveMongoDbMessageSource 実装が提供されています。それらは Spring Data の ReactiveMongoOperations に基づいており、org.mongodb:mongodb-driver-reactivestreams 依存関係が必要です。
ReactiveMongoDbStoringMessageHandler は ReactiveMessageHandler の実装であり、反応フローの構成が統合フローの定義に含まれている場合に、フレームワークでネイティブにサポートされます。詳細については、ReactiveMessageHandler を参照してください。
構成の観点からは、他の多くの標準チャネルアダプターと違いはありません。例: Java DSL では、このようなチャネルアダプターは次のように使用できます。
@Bean
public IntegrationFlow reactiveMongoDbFlow(ReactiveMongoDatabaseFactory mongoDbFactory) {
return f -> f
.channel(MessageChannels.flux())
.handle(MongoDb.reactiveOutboundChannelAdapter(mongoDbFactory));
} このサンプルでは、提供された ReactiveMongoDatabaseFactory を介して MongoDb に接続し、リクエストメッセージからのデータを data 名でデフォルトのコレクションに保存します。実際の操作は、内部で作成された ReactiveStreamsConsumer の反応ストリーム構成からオンデマンドで実行されます。
ReactiveMongoDbMessageSource は、提供された ReactiveMongoDatabaseFactory または ReactiveMongoOperations と MongoDb クエリ（または式）に基づく AbstractMessageSource 実装であり、expectSingleResult オプションに従って find() または findOne() 操作を呼び出し、想定される entityClass 型でクエリ結果を変換します。生成されたメッセージのペイロードに含まれる Publisher （expectSingleResult オプションに従って Flux または Mono）がサブスクライブされると、クエリの実行と結果の評価がオンデマンドで実行されます。フレームワークは、スプリッターと FluxMessageChannel が下流で使用される場合、そのようなペイロード（基本的に flatMap）を自動的にサブスクライブできます。それ以外の場合、下流のエンドポイントでポーリングされたパブリッシャーにサブスクライブするのは、ターゲットアプリケーションの責任です。
Java DSL では、このようなチャネルアダプターは次のように構成できます。
@Bean
public IntegrationFlow reactiveMongoDbFlow(ReactiveMongoDatabaseFactory mongoDbFactory) {
return IntegrationFlow
.from(MongoDb.reactiveInboundChannelAdapter(mongoDbFactory, "{'name' : 'Name'}")
.entityClass(Person.class),
c -> c.poller(Pollers.fixedDelay(1000)))
.split()
.channel(c -> c.flux("output"))
.get();
} バージョン 5.5 以降、ReactiveMongoDbMessageSource は updateExpression で構成できます。ブロッキング MongoDbMessageSource と同じ機能があります。詳細については、MongoDB 受信チャネルアダプターおよび AbstractMongoDbMessageSourceSpec JavaDocs を参照してください。
