Skip to content

Commit 0606a6f

Browse files
artembilangaryrussell
artembilan
authored and
garyrussell
committed
Add JavaDSL for ReactiveMongoDbMessageSource
* Implement `ReactiveMongoDbMessageSourceSpec` and factories for it * Rework `ReactiveMongoDbMessageSourceTests` to use new Java DSL for `ReactiveMongoDbMessageSource` * Document changes
1 parent c7c7422 commit 0606a6f

File tree

6 files changed

+221
-27
lines changed

6 files changed

+221
-27
lines changed

‎spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java

Lines changed: 64 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,11 +21,15 @@
2121
import org.springframework.data.mongodb.core.MongoOperations;
2222
import org.springframework.data.mongodb.core.ReactiveMongoOperations;
2323
import org.springframework.data.mongodb.core.convert.MongoConverter;
24+
import org.springframework.data.mongodb.core.query.Query;
25+
import org.springframework.expression.common.LiteralExpression;
26+
import org.springframework.integration.expression.ValueExpression;
2427

2528
/**
2629
* Factory class for building MongoDb components
2730
*
2831
* @author Xavier Padro
32+
* @author Artem Bilan
2933
*
3034
* @since 5.0
3135
*/
@@ -74,10 +78,69 @@ public static ReactiveMongoDbMessageHandlerSpec reactiveOutboundChannelAdapter(
7478
* @return the {@link ReactiveMongoDbMessageHandlerSpec} instance
7579
* @since 5.3
7680
*/
77-
public static ReactiveMongoDbMessageHandlerSpec reactiveOutboundChannelAdapter(ReactiveMongoOperations mongoTemplate) {
81+
public static ReactiveMongoDbMessageHandlerSpec reactiveOutboundChannelAdapter(
82+
ReactiveMongoOperations mongoTemplate) {
83+
7884
return new ReactiveMongoDbMessageHandlerSpec(mongoTemplate);
7985
}
8086

87+
/**
88+
* Create a {@link ReactiveMongoDbMessageSourceSpec} builder instance
89+
* based on the provided {@link ReactiveMongoDatabaseFactory}.
90+
* @param mongoDbFactory the {@link ReactiveMongoDatabaseFactory} to use.
91+
* @param query the MongoDb query
92+
* @return the {@link ReactiveMongoDbMessageSourceSpec} instance
93+
* @since 5.3
94+
*/
95+
public static ReactiveMongoDbMessageSourceSpec reactiveInboundChannelAdapter(
96+
ReactiveMongoDatabaseFactory mongoDbFactory, String query) {
97+
98+
return new ReactiveMongoDbMessageSourceSpec(mongoDbFactory, new LiteralExpression(query));
99+
}
100+
101+
/**
102+
* Create a {@link ReactiveMongoDbMessageSourceSpec} builder instance
103+
* based on the provided {@link ReactiveMongoDatabaseFactory}.
104+
* @param mongoDbFactory the {@link ReactiveMongoDatabaseFactory} to use.
105+
* @param query the MongoDb query DSL object
106+
* @return the {@link ReactiveMongoDbMessageSourceSpec} instance
107+
* @since 5.3
108+
*/
109+
public static ReactiveMongoDbMessageSourceSpec reactiveInboundChannelAdapter(
110+
ReactiveMongoDatabaseFactory mongoDbFactory, Query query) {
111+
112+
return new ReactiveMongoDbMessageSourceSpec(mongoDbFactory, new ValueExpression<>(query));
113+
}
114+
115+
/**
116+
* Create a {@link ReactiveMongoDbMessageSourceSpec} builder instance
117+
* based on the provided {@link ReactiveMongoOperations}.
118+
* @param mongoTemplate the {@link ReactiveMongoOperations} to use.
119+
* @param query the MongoDb query
120+
* @return the {@link ReactiveMongoDbMessageSourceSpec} instance
121+
* @since 5.3
122+
*/
123+
public static ReactiveMongoDbMessageSourceSpec reactiveInboundChannelAdapter(ReactiveMongoOperations mongoTemplate,
124+
String query) {
125+
126+
return new ReactiveMongoDbMessageSourceSpec(mongoTemplate, new LiteralExpression(query));
127+
}
128+
129+
/**
130+
* Create a {@link ReactiveMongoDbMessageSourceSpec} builder instance
131+
* based on the provided {@link ReactiveMongoOperations}.
132+
* @param mongoTemplate the {@link ReactiveMongoOperations} to use.
133+
* @param query the MongoDb query DSL object
134+
* @return the {@link ReactiveMongoDbMessageSourceSpec} instance
135+
* @since 5.3
136+
*/
137+
public static ReactiveMongoDbMessageSourceSpec reactiveInboundChannelAdapter(ReactiveMongoOperations mongoTemplate,
138+
Query query) {
139+
140+
return new ReactiveMongoDbMessageSourceSpec(mongoTemplate, new ValueExpression<>(query));
141+
}
142+
143+
81144
private MongoDb() {
82145
}
83146

‎spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageSourceSpec.java

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
1+
/*
2+
* Copyright 2020 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.integration.mongodb.dsl;
18+
19+
import java.util.function.Supplier;
20+
21+
import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
22+
import org.springframework.data.mongodb.core.ReactiveMongoOperations;
23+
import org.springframework.data.mongodb.core.convert.MongoConverter;
24+
import org.springframework.data.mongodb.core.query.Query;
25+
import org.springframework.expression.Expression;
26+
import org.springframework.expression.common.LiteralExpression;
27+
import org.springframework.integration.dsl.MessageSourceSpec;
28+
import org.springframework.integration.expression.SupplierExpression;
29+
import org.springframework.integration.mongodb.inbound.ReactiveMongoDbMessageSource;
30+
31+
/**
32+
* A {@link MessageSourceSpec} implementation for a {@link ReactiveMongoDbMessageSource}.
33+
*
34+
* @author Artem Bilan
35+
*
36+
* @since 5.3
37+
*/
38+
public class ReactiveMongoDbMessageSourceSpec
39+
extends MessageSourceSpec<ReactiveMongoDbMessageSourceSpec, ReactiveMongoDbMessageSource> {
40+
41+
ReactiveMongoDbMessageSourceSpec(ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory,
42+
Expression queryExpression) {
43+
this.target = new ReactiveMongoDbMessageSource(reactiveMongoDatabaseFactory, queryExpression);
44+
}
45+
46+
ReactiveMongoDbMessageSourceSpec(ReactiveMongoOperations reactiveMongoTemplate, Expression queryExpression) {
47+
this.target = new ReactiveMongoDbMessageSource(reactiveMongoTemplate, queryExpression);
48+
}
49+
50+
/**
51+
* Allow you to set the type of the entityClass that will be passed to the
52+
* {@link ReactiveMongoOperations#find(Query, Class)} or {@link ReactiveMongoOperations#findOne(Query, Class)}
53+
* method.
54+
* Default is {@link com.mongodb.DBObject}.
55+
* @param entityClass The entity class.
56+
* @return the spec
57+
* @see ReactiveMongoDbMessageSource#setEntityClass(Class)
58+
*/
59+
public ReactiveMongoDbMessageSourceSpec entityClass(Class<?> entityClass) {
60+
this.target.setEntityClass(entityClass);
61+
return this;
62+
}
63+
64+
/**
65+
* Allow you to manage which find* method to invoke on {@link ReactiveMongoOperations}.
66+
* @param expectSingleResult true if a single result is expected.
67+
* @return the spec
68+
* @see ReactiveMongoDbMessageSource#setExpectSingleResult(boolean)
69+
*/
70+
public ReactiveMongoDbMessageSourceSpec expectSingleResult(boolean expectSingleResult) {
71+
this.target.setExpectSingleResult(expectSingleResult);
72+
return this;
73+
}
74+
75+
/**
76+
* Configure a collection name to query against.
77+
* @param collectionName the name of the MongoDb collection
78+
* @return the spec
79+
*/
80+
public ReactiveMongoDbMessageSourceSpec collectionName(String collectionName) {
81+
return collectionNameExpression(new LiteralExpression(collectionName));
82+
}
83+
84+
/**
85+
* Configure a SpEL expression to evaluation a collection name on each {@code receive()} call.
86+
* @param collectionNameExpression the SpEL expression for name of the MongoDb collection
87+
* @return the spec
88+
*/
89+
public ReactiveMongoDbMessageSourceSpec collectionNameExpression(String collectionNameExpression) {
90+
return collectionNameExpression(PARSER.parseExpression(collectionNameExpression));
91+
}
92+
93+
/**
94+
* Configure a {@link Supplier} to obtain a collection name on each {@code receive()} call.
95+
* @param collectionNameSupplier the {@link Supplier} for name of the MongoDb collection
96+
* @return the spec
97+
*/
98+
public ReactiveMongoDbMessageSourceSpec collectionNameSupplier(Supplier<String> collectionNameSupplier) {
99+
return collectionNameExpression(new SupplierExpression<>(collectionNameSupplier));
100+
}
101+
102+
/**
103+
* Configure a SpEL expression to evaluation a collection name on each {@code receive()} call.
104+
* @param collectionNameExpression the SpEL expression for name of the MongoDb collection
105+
* @return the spec
106+
* @see ReactiveMongoDbMessageSource#setCollectionNameExpression(Expression)
107+
*/
108+
public ReactiveMongoDbMessageSourceSpec collectionNameExpression(Expression collectionNameExpression) {
109+
this.target.setCollectionNameExpression(collectionNameExpression);
110+
return this;
111+
}
112+
113+
/**
114+
* Configure a custom {@link MongoConverter} used to assist in deserialization
115+
* data read from MongoDb. Only allowed if this instance was constructed with a
116+
* {@link ReactiveMongoDatabaseFactory}.
117+
* @param mongoConverter The mongo converter.
118+
* @return the spec
119+
* @see ReactiveMongoDbMessageSource#setMongoConverter(MongoConverter)
120+
*/
121+
public ReactiveMongoDbMessageSourceSpec mongoConverter(MongoConverter mongoConverter) {
122+
this.target.setMongoConverter(mongoConverter);
123+
return this;
124+
}
125+
126+
}

‎spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSourceTests.java

Lines changed: 7 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@
3232
import org.bson.conversions.Bson;
3333
import org.junit.Test;
3434
import org.mockito.Mockito;
35-
import org.reactivestreams.Publisher;
3635

3736
import org.springframework.beans.factory.BeanFactory;
3837
import org.springframework.context.ConfigurableApplicationContext;
@@ -47,10 +46,10 @@
4746
import org.springframework.expression.common.LiteralExpression;
4847
import org.springframework.integration.channel.FluxMessageChannel;
4948
import org.springframework.integration.config.EnableIntegration;
50-
import org.springframework.integration.core.MessageSource;
5149
import org.springframework.integration.dsl.IntegrationFlow;
5250
import org.springframework.integration.dsl.IntegrationFlows;
5351
import org.springframework.integration.dsl.Pollers;
52+
import org.springframework.integration.mongodb.dsl.MongoDb;
5453
import org.springframework.integration.mongodb.rules.MongoDbAvailable;
5554
import org.springframework.integration.mongodb.rules.MongoDbAvailableTests;
5655

@@ -245,30 +244,14 @@ private static <T> T waitFor(Mono<T> mono) {
245244
static class TestContext {
246245

247246
@Bean
248-
FluxMessageChannel output() {
249-
return new FluxMessageChannel();
250-
}
251-
252-
@Bean
253-
public MessageSource<Publisher<?>> mongodbMessageSource(ReactiveMongoDatabaseFactory mongoDatabaseFactory) {
254-
Expression queryExpression = new LiteralExpression("{'name' : 'Oleg'}");
255-
ReactiveMongoDbMessageSource reactiveMongoDbMessageSource =
256-
new ReactiveMongoDbMessageSource(mongoDatabaseFactory, queryExpression);
257-
reactiveMongoDbMessageSource.setEntityClass(Person.class);
258-
return reactiveMongoDbMessageSource;
259-
}
260-
261-
@Bean
262-
ReactiveMongoDatabaseFactory mongoDatabaseFactory() {
263-
return MongoDbAvailableTests.REACTIVE_MONGO_DATABASE_FACTORY;
264-
}
265-
266-
@Bean
267-
public IntegrationFlow pollingFlow(MessageSource<Publisher<?>> mongodbMessageSource) {
247+
public IntegrationFlow pollingFlow() {
268248
return IntegrationFlows
269-
.from(mongodbMessageSource, c -> c.poller(Pollers.fixedDelay(100).maxMessagesPerPoll(1)))
249+
.from(MongoDb.reactiveInboundChannelAdapter(
250+
MongoDbAvailableTests.REACTIVE_MONGO_DATABASE_FACTORY, "{'name' : 'Oleg'}")
251+
.entityClass(Person.class),
252+
c -> c.poller(Pollers.fixedDelay(100).maxMessagesPerPoll(1)))
270253
.split()
271-
.channel(output())
254+
.channel(c -> c.flux("output"))
272255
.get();
273256
}
274257

‎src/reference/asciidoc/mongodb.adoc

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -544,3 +544,26 @@ public IntegrationFlow reactiveMongoDbFlow(ReactiveMongoDatabaseFactory mongoDbF
544544

545545
In this sample we are going to connect to the MongoDb via provided `ReactiveMongoDatabaseFactory` and store a data from request message into a default collection with the `data` name.
546546
The real operation is going to be performed on-demand from the reactive stream composition in the internally created `ReactiveStreamsConsumer`.
547+
548+
The `ReactiveMongoDbMessageSource` is an `AbstractMessageSource` implementation based on the provided `ReactiveMongoDatabaseFactory` or `ReactiveMongoOperations` and MongoDb query (or expression), calls `find()` or `findOne()` operation according an `expectSingleResult` option with an expected `entityClass` type to convert a query result.
549+
A query execution and result evaluation is performed on demand when `Publisher` (`Flux` or `Mono` according `expectSingleResult` option) in the payload of produced message is subscribed.
550+
The framework can subscribe to such a payload automatically (essentially `flatMap`) when splitter and `FluxMessageChannel` are used downstream.
551+
Otherwise it is target application responsibility to subscribe into a polled publishers in downstream endpoints.
552+
553+
With Java DSL such a channel adapter could be configured like:
554+
555+
====
556+
[source, java]
557+
----
558+
@Bean
559+
public IntegrationFlow reactiveMongoDbFlow(ReactiveMongoDatabaseFactory mongoDbFactory) {
560+
return IntegrationFlows
561+
.from(MongoDb.reactiveInboundChannelAdapter(mongoDbFactory, "{'name' : 'Name'}")
562+
.entityClass(Person.class),
563+
c -> c.poller(Pollers.fixedDelay(1000)))
564+
.split()
565+
.channel(c -> c.flux("output"))
566+
.get();
567+
}
568+
----
569+
====

‎src/reference/asciidoc/reactive-streams.adoc

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -134,5 +134,5 @@ A returned reactive type can be subscribed immediately if we are in one-way, fir
134134

135135
Currently Spring Integration provides channel adapter (or gateway) implementations for <<./webflux.adoc#webflux,WebFlux>>, <<./rsocket.adoc#rsocket,RSocket>> and <<./mongodb.adoc#mongodb,MongoDb>>.
136136
Also an https://github.com/spring-projects/spring-integration-extensions/tree/master/spring-integration-cassandra[Apache Cassandra Extension] provides a `MessageHandler` implementation for the Cassandra reactive driver.
137-
More reactive channel adapters are coming, for example for https://r2dbc.io/[R2DBC], https://mongodb.github.io/mongo-java-driver-reactivestreams/[MongoDB], for Apache Kafka in https://github.com/spring-projects/spring-integration-kafka[Spring Integration Kafka] based on the `ReactiveKafkaProducerTemplate` and `ReactiveKafkaConsumerTemplate` from https://spring.io/projects/spring-kafka[Spring for Apache Kafka] etc.
137+
More reactive channel adapters are coming, for example for https://r2dbc.io/[R2DBC], for Apache Kafka in https://github.com/spring-projects/spring-integration-kafka[Spring Integration Kafka] based on the `ReactiveKafkaProducerTemplate` and `ReactiveKafkaConsumerTemplate` from https://spring.io/projects/spring-kafka[Spring for Apache Kafka] etc.
138138
For many other non-reactive channel adapters thread pools are recommended to avoid blocking during reactive stream processing.

‎src/reference/asciidoc/whats-new.adoc

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,5 @@ See <<./mongodb.adoc#mongodb-reactive-channel-adapters,MongoDB Reactive Channel
3939
The gateway proxy now doesn't proxy `default` methods by default.
4040
See <<./gateway.adoc/gateway-calling-default-methods,Invoking `default` Methods>> for more information.
4141

42-
4342
Internal components (such as `_org.springframework.integration.errorLogger`) now have a shortened name when they are represented in the integration graph.
4443
See <<./graph.adoc#integration-graph,Integration Graph>> for more information.

0 commit comments

Comments
 (0)