Add Java DSL for R2DBC module (#3415)

* Add Java DSL for R2DBC module

* Add JavaDocs for R2DBC component setters
* Introduce a `R2dbcMessageSource.SelectCreator` to narrow
a `StatementMapper` API access from SpEL
* Add missed `package-info.java` into R2DBC packages
* Add `hamcrest-core` as test dependency into SI-R2DBC
since Awaitility doesn't work without it

* * Add `@NonNullApi` to R2DBC packages

build.gradle

@@ -647,6 +647,7 @@ project('spring-integration-r2dbc') {
 		}
 		api 'org.springframework:spring-r2dbc'
 		testImplementation "io.r2dbc:r2dbc-h2:$r2dbch2Version"
+		testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion"
 	}
 }
 

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbc.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.r2dbc.dsl;
+
+import java.util.function.Function;
+
+import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
+import org.springframework.data.r2dbc.core.StatementMapper;
+import org.springframework.expression.Expression;
+import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.integration.r2dbc.inbound.R2dbcMessageSource;
+
+/**
+ * Java DSL Factory class for R2DBC components.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.4
+ */
+public final class R2dbc {
+
+	/**
+	 * Create an instance of {@link R2dbcMessageSourceSpec} for the provided {@link R2dbcEntityOperations}
+	 * and query string.
+	 * @param r2dbcEntityOperations the {@link R2dbcEntityOperations} to use.
+	 * @param query the query to execute.
+	 * @return the spec.
+	 */
+	public static R2dbcMessageSourceSpec inboundChannelAdapter(R2dbcEntityOperations r2dbcEntityOperations,
+			String query) {
+
+		return new R2dbcMessageSourceSpec(r2dbcEntityOperations, query);
+	}
+
+	/**
+	 * Create an instance of {@link R2dbcMessageSourceSpec} for the provided {@link R2dbcEntityOperations}
+	 * and function to create a {@link StatementMapper.SelectSpec} instance.
+	 * @param r2dbcEntityOperations the {@link R2dbcEntityOperations} to use.
+	 * @param selectFunction the expression to evaluate a query for execution.
+	 * @return the spec.
+	 */
+	public static R2dbcMessageSourceSpec inboundChannelAdapter(R2dbcEntityOperations r2dbcEntityOperations,
+			Function<R2dbcMessageSource.SelectCreator, StatementMapper.SelectSpec> selectFunction) {
+
+		return inboundChannelAdapter(r2dbcEntityOperations, new FunctionExpression<>(selectFunction));
+	}
+
+	/**
+	 * Create an instance of {@link R2dbcMessageSourceSpec} for the provided {@link R2dbcEntityOperations}
+	 * and SpEL expression for query.
+	 * @param r2dbcEntityOperations the {@link R2dbcEntityOperations} to use.
+	 * @param queryExpression the expression to evaluate a query for execution.
+	 * @return the spec.
+	 */
+	public static R2dbcMessageSourceSpec inboundChannelAdapter(R2dbcEntityOperations r2dbcEntityOperations,
+			Expression queryExpression) {
+
+		return new R2dbcMessageSourceSpec(r2dbcEntityOperations, queryExpression);
+	}
+
+	/**
+	 * Create an instance of {@link R2dbcMessageHandlerSpec} for the provided {@link R2dbcEntityOperations}.
+	 * @param r2dbcEntityOperations the {@link R2dbcEntityOperations} to use.
+	 * @return the spec.
+	 */
+	public static R2dbcMessageHandlerSpec outboundChannelAdapter(R2dbcEntityOperations r2dbcEntityOperations) {
+		return new R2dbcMessageHandlerSpec(r2dbcEntityOperations);
+	}
+
+	private R2dbc() {
+	}
+
+}

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbcMessageHandlerSpec.java

@@ -0,0 +1,185 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.r2dbc.dsl;
+
+import java.util.Map;
+import java.util.function.Function;
+
+import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
+import org.springframework.data.relational.core.query.Criteria;
+import org.springframework.expression.Expression;
+import org.springframework.integration.dsl.ReactiveMessageHandlerSpec;
+import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.integration.r2dbc.outbound.R2dbcMessageHandler;
+import org.springframework.messaging.Message;
+
+/**
+ * The {@link ReactiveMessageHandlerSpec} for the {@link R2dbcMessageHandler}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.4
+ */
+public class R2dbcMessageHandlerSpec extends ReactiveMessageHandlerSpec<R2dbcMessageHandlerSpec, R2dbcMessageHandler> {
+
+	protected R2dbcMessageHandlerSpec(R2dbcEntityOperations r2dbcEntityOperations) {
+		super(new R2dbcMessageHandler(r2dbcEntityOperations));
+	}
+
+	/**
+	 * Set a {@link R2dbcMessageHandler.Type} for query to execute.
+	 * @param type the {@link R2dbcMessageHandler.Type} to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageHandlerSpec queryType(R2dbcMessageHandler.Type type) {
+		this.reactiveMessageHandler.setQueryType(type);
+		return this;
+	}
+
+	/**
+	 * Set a {@link Function} to evaluate a {@link R2dbcMessageHandler.Type} for query to execute against a request
+	 * message.
+	 * @param queryTypeFunction the function to use.
+	 * @param <P> the payload type.
+	 * @return the spec
+	 */
+	public <P> R2dbcMessageHandlerSpec queryTypeFunction(
+			Function<Message<P>, R2dbcMessageHandler.Type> queryTypeFunction) {
+
+		return queryTypeExpression(new FunctionExpression<>(queryTypeFunction));
+	}
+
+	/**
+	 * Set a SpEL expression to evaluate a {@link R2dbcMessageHandler.Type} for query to execute.
+	 * @param queryTypeExpression the expression to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageHandlerSpec queryTypeExpression(String queryTypeExpression) {
+		return queryTypeExpression(PARSER.parseExpression(queryTypeExpression));
+	}
+
+	/**
+	 * Set a SpEL expression to evaluate a {@link R2dbcMessageHandler.Type} for query to execute.
+	 * @param queryTypeExpression the expression to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageHandlerSpec queryTypeExpression(Expression queryTypeExpression) {
+		this.reactiveMessageHandler.setQueryTypeExpression(queryTypeExpression);
+		return this;
+	}
+
+	/**
+	 * Specify a table in the target database to execute the query.
+	 * @param tableName the name of the table to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageHandlerSpec tableName(String tableName) {
+		this.reactiveMessageHandler.setTableName(tableName);
+		return this;
+	}
+
+	/**
+	 * Set a {@link Function} to evaluate a table name at runtime against request message.
+	 * @param tableNameFunction the function to use.
+	 * @param <P> the payload type.
+	 * @return the spec
+	 */
+	public <P> R2dbcMessageHandlerSpec tableNameFunction(Function<Message<P>, String> tableNameFunction) {
+		return tableNameExpression(new FunctionExpression<>(tableNameFunction));
+	}
+
+	/**
+	 * Set a SpEL expression to evaluate a table name at runtime against request message.
+	 * @param tableNameExpression the expression to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageHandlerSpec tableNameExpression(String tableNameExpression) {
+		return tableNameExpression(PARSER.parseExpression(tableNameExpression));
+	}
+
+	/**
+	 * Set a SpEL expression to evaluate a table name at runtime against request message.
+	 * @param tableNameExpression the expression to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageHandlerSpec tableNameExpression(Expression tableNameExpression) {
+		this.reactiveMessageHandler.setTableNameExpression(tableNameExpression);
+		return this;
+	}
+
+	/**
+	 * Set a {@link Function} to evaluate a {@link Map} for name-value pairs to bind as parameters
+	 * into a query.
+	 * @param valuesFunction the function to use.
+	 * @param <P> the payload type.
+	 * @return the spec
+	 */
+	public <P> R2dbcMessageHandlerSpec values(Function<Message<P>, Map<String, ?>> valuesFunction) {
+		return values(new FunctionExpression<>(valuesFunction));
+	}
+
+	/**
+	 * Set a SpEL expression to evaluate a {@link Map} for name-value pairs to bind as parameters
+	 * into a query.
+	 * @param valuesExpression the expression to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageHandlerSpec values(String valuesExpression) {
+		return values(PARSER.parseExpression(valuesExpression));
+	}
+
+	/**
+	 * Set a SpEL expression to evaluate a {@link Map} for name-value pairs to bind as parameters
+	 * into a query.
+	 * @param valuesExpression the expression to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageHandlerSpec values(Expression valuesExpression) {
+		this.reactiveMessageHandler.setValuesExpression(valuesExpression);
+		return this;
+	}
+
+	/**
+	 * Set a {@link Function} to evaluate a {@link Criteria} for query to execute.
+	 * @param criteriaFunction the function to use.
+	 * @param <P> the payload type.
+	 * @return the spec
+	 */
+	public <P> R2dbcMessageHandlerSpec criteria(Function<Message<P>, Criteria> criteriaFunction) {
+		return criteria(new FunctionExpression<>(criteriaFunction));
+	}
+
+	/**
+	 * Set a SpEL expression to evaluate a {@link Criteria} for query to execute.
+	 * @param criteriaExpression the expression to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageHandlerSpec criteria(String criteriaExpression) {
+		return criteria(PARSER.parseExpression(criteriaExpression));
+	}
+
+	/**
+	 * Set a SpEL expression to evaluate a {@link Criteria} for query to execute.
+	 * @param criteriaExpression the expression to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageHandlerSpec criteria(Expression criteriaExpression) {
+		this.reactiveMessageHandler.setCriteriaExpression(criteriaExpression);
+		return this;
+	}
+
+}

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbcMessageSourceSpec.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.r2dbc.dsl;
+
+import java.util.function.BiFunction;
+
+import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
+import org.springframework.expression.Expression;
+import org.springframework.integration.dsl.MessageSourceSpec;
+import org.springframework.integration.r2dbc.inbound.R2dbcMessageSource;
+import org.springframework.r2dbc.core.DatabaseClient;
+
+/**
+ * The {@link MessageSourceSpec} for the {@link R2dbcMessageSource}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.4
+ */
+public class R2dbcMessageSourceSpec extends MessageSourceSpec<R2dbcMessageSourceSpec, R2dbcMessageSource> {
+
+	protected R2dbcMessageSourceSpec(R2dbcEntityOperations r2dbcEntityOperations, String query) {
+		this.target = new R2dbcMessageSource(r2dbcEntityOperations, query);
+	}
+
+	protected R2dbcMessageSourceSpec(R2dbcEntityOperations r2dbcEntityOperations, Expression queryExpression) {
+		this.target = new R2dbcMessageSource(r2dbcEntityOperations, queryExpression);
+	}
+
+	/**
+	 * Set the expected payload type.
+	 * @param payloadType the class to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageSourceSpec payloadType(Class<?> payloadType) {
+		this.target.setPayloadType(payloadType);
+		return this;
+	}
+
+	/**
+	 * Configure an update query.
+	 * @param updateSql the update query string.
+	 * @return the spec
+	 */
+	public R2dbcMessageSourceSpec updateSql(String updateSql) {
+		this.target.setUpdateSql(updateSql);
+		return this;
+	}
+
+	/**
+	 * Set a {@link BiFunction} which is used to bind parameters into the update query.
+	 * @param bindFunction the {@link BiFunction} to use.
+	 * @return the spec
+	 */
+	public R2dbcMessageSourceSpec bindFunction(
+			BiFunction<DatabaseClient.GenericExecuteSpec, ?, DatabaseClient.GenericExecuteSpec> bindFunction) {
+
+		this.target.setBindFunction(bindFunction);
+		return this;
+	}
+
+	/**
+	 * The flag to manage which find* method to invoke on {@link R2dbcEntityOperations}.
+	 * @param expectSingleResult true if a single result is expected.
+	 * @return the spec
+	 */
+	public R2dbcMessageSourceSpec expectSingleResult(boolean expectSingleResult) {
+		this.target.setExpectSingleResult(expectSingleResult);
+		return this;
+	}
+
+}

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Provides classes for supporting Java DSL for R2DBC components.
+ */
+@org.springframework.lang.NonNullApi
+package org.springframework.integration.r2dbc.dsl;

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSource.java

@@ -25,6 +25,7 @@ import org.reactivestreams.Publisher;
 import org.springframework.data.r2dbc.convert.EntityRowMapper;
 import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
 import org.springframework.data.r2dbc.core.StatementMapper;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
@@ -65,6 +66,8 @@ public class R2dbcMessageSource extends AbstractMessageSource<Publisher<?>> {
 
 	private final StatementMapper statementMapper;
 
+	private final SelectCreator selectCreator = new SelectCreator();
+
 	private final Expression queryExpression;
 
 	private Class<?> payloadType = Map.class;
@@ -95,12 +98,12 @@ public class R2dbcMessageSource extends AbstractMessageSource<Publisher<?>> {
 
 	/**
 	 * Create an instance with the provided {@link R2dbcEntityOperations} and SpEL expression
-	 * which should resolve to a Relational 'query' string.
+	 * which should resolve to a query string or {@link StatementMapper.SelectSpec} instance.
 	 * It assumes that the {@link R2dbcEntityOperations} is fully initialized and ready to be used.
 	 * The 'queryExpression' will be evaluated on every call to the {@link #receive()} method.
 	 * @param r2dbcEntityOperations  The reactive for performing database calls.
-	 * @param queryExpression The query expression. The root object for evaluation context is a {@link StatementMapper}
-	 * for its {@link StatementMapper#createSelect} fluent API.
+	 * @param queryExpression The query expression. The root object for evaluation context is a {@link SelectCreator}
+	 * for delegation int the {@link StatementMapper#createSelect} fluent API.
 	 */
 	@SuppressWarnings("deprecation")
 	public R2dbcMessageSource(R2dbcEntityOperations r2dbcEntityOperations, Expression queryExpression) {
@@ -113,9 +116,8 @@ public class R2dbcMessageSource extends AbstractMessageSource<Publisher<?>> {
 	}
 
 	/**
-	 * Provide a way to set the type of the entityClass that will be passed to the
-	 * {@link DatabaseClient#sql(String)} method.
-	 * @param payloadType The t class.
+	 * Set the type of the entityClass which is used for the {@link EntityRowMapper}.
+	 * @param payloadType The class to use.
 	 */
 	public void setPayloadType(Class<?> payloadType) {
 		Assert.notNull(payloadType, "'payloadType' must not be null");
@@ -123,18 +125,16 @@ public class R2dbcMessageSource extends AbstractMessageSource<Publisher<?>> {
 	}
 
 	/**
-	 * Provide a way to set update query that will be passed to the
-	 * {@link DatabaseClient#sql(String)} method.
-	 * @param updateSql Update query string.
+	 * Set an update query that will be passed to the {@link DatabaseClient#sql(String)} method.
+	 * @param updateSql the update query string.
 	 */
 	public void setUpdateSql(String updateSql) {
 		this.updateSql = updateSql;
 	}
 
 	/**
-	 * Provide a way to set BindFunction which will be used to bind parameters
-	 * in the update query.
-	 * @param bindFunction The bindFunction.
+	 * Set a {@link BiFunction} which is used to bind parameters into the update query.
+	 * @param bindFunction the {@link BiFunction} to use.
 	 */
 	@SuppressWarnings("unchecked")
 	public void setBindFunction(
@@ -145,7 +145,7 @@ public class R2dbcMessageSource extends AbstractMessageSource<Publisher<?>> {
 	}
 
 	/**
-	 * Provide a way to manage which find* method to invoke on {@link R2dbcEntityOperations}.
+	 * The flag to manage which find* method to invoke on {@link R2dbcEntityOperations}.
 	 * Default is 'false', which means the {@link #receive()} method will use
 	 * the {@link DatabaseClient#sql(String)} method and will fetch all. If set
 	 * to 'true'{@link #receive()} will use {@link DatabaseClient#sql(String)}
@@ -179,14 +179,14 @@ public class R2dbcMessageSource extends AbstractMessageSource<Publisher<?>> {
 	 * or a single element of type identified by {@link #payloadType}
 	 * based on the value of {@link #expectSingleResult} attribute which defaults to 'false' resulting
 	 * {@link org.springframework.messaging.Message} with payload of type
-	 * {@link reactor.core.publisher.Flux}. The collection name used in the
+	 * {@link reactor.core.publisher.Flux}.
 	 */
 	@Override
 	protected Object doReceive() {
 		Assert.isTrue(this.initialized, "This class is not yet initialized. Invoke its afterPropertiesSet() method");
 		Mono<RowsFetchSpec<?>> queryMono =
 				Mono.fromSupplier(() ->
-						this.queryExpression.getValue(this.evaluationContext, this.statementMapper))
+						this.queryExpression.getValue(this.evaluationContext, this.selectCreator))
 						.map(this::prepareFetch);
 		if (this.expectSingleResult) {
 			return queryMono.flatMap(RowsFetchSpec::one)
@@ -226,4 +226,23 @@ public class R2dbcMessageSource extends AbstractMessageSource<Publisher<?>> {
 				"or org.springframework.data.r2dbc.core.StatementMapper.SelectSpec, but not: " + queryObject);
 	}
 
+	/**
+	 * An instance of this class is used as a root object for query expression
+	 * to give a limited access only to the {@link StatementMapper#createSelect} fluent API.
+	 */
+	public class SelectCreator {
+
+		SelectCreator() {
+		}
+
+		public StatementMapper.SelectSpec createSelect(String table) {
+			return R2dbcMessageSource.this.statementMapper.createSelect(table);
+		}
+
+		public StatementMapper.SelectSpec createSelect(SqlIdentifier table) {
+			return R2dbcMessageSource.this.statementMapper.createSelect(table);
+		}
+
+	}
+
 }

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Provides classes for supporting R2DBC inbound components.
+ */
+@org.springframework.lang.NonNullApi
+package org.springframework.integration.r2dbc.inbound;

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java

@@ -84,27 +84,52 @@ public class R2dbcMessageHandler extends AbstractReactiveMessageHandler {
 	}
 
 
+	/**
+	 * Set a {@link R2dbcMessageHandler.Type} for query to execute.
+	 * @param type the {@link R2dbcMessageHandler.Type} to use.
+	 */
 	public void setQueryType(R2dbcMessageHandler.Type type) {
 		setQueryTypeExpression(new ValueExpression<>(type));
 	}
 
+	/**
+	 * Set a SpEL expression to evaluate a {@link R2dbcMessageHandler.Type} for query to execute.
+	 * @param queryTypeExpression the expression to use.
+	 */
 	public void setQueryTypeExpression(Expression queryTypeExpression) {
 		Assert.notNull(queryTypeExpression, "'queryTypeExpression' must not be null");
 		this.queryTypeExpression = queryTypeExpression;
 	}
 
+	/**
+	 * Specify a table in the target database to execute the query.
+	 * @param tableName the name of the table to use.
+	 */
 	public void setTableName(String tableName) {
 		setTableNameExpression(new LiteralExpression(tableName));
 	}
 
+	/**
+	 * Set a SpEL expression to evaluate a table name at runtime against request message.
+	 * @param tableNameExpression the expression to use.
+	 */
 	public void setTableNameExpression(Expression tableNameExpression) {
 		this.tableNameExpression = tableNameExpression;
 	}
 
+	/**
+	 * Set a SpEL expression to evaluate a {@link Map} for name-value pairs to bind as parameters
+	 * into a query.
+	 * @param valuesExpression the expression to use.
+	 */
 	public void setValuesExpression(Expression valuesExpression) {
 		this.valuesExpression = valuesExpression;
 	}
 
+	/**
+	 * Set a SpEL expression to evaluate a {@link Criteria} for query to execute.
+	 * @param criteriaExpression the expression to use.
+	 */
 	public void setCriteriaExpression(Expression criteriaExpression) {
 		this.criteriaExpression = criteriaExpression;
 	}

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Provides classes for supporting R2DBC outbound components.
+ */
+@org.springframework.lang.NonNullApi
+package org.springframework.integration.r2dbc.outbound;

spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/dsl/R2dbcDslTests.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.r2dbc.dsl;
+
+import static org.awaitility.Awaitility.await;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.Lifecycle;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
+import org.springframework.data.relational.core.query.Criteria;
+import org.springframework.data.relational.core.query.Query;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.MessageChannels;
+import org.springframework.integration.r2dbc.config.R2dbcDatabaseConfiguration;
+import org.springframework.integration.r2dbc.entity.Person;
+import org.springframework.integration.r2dbc.outbound.R2dbcMessageHandler;
+import org.springframework.r2dbc.core.DatabaseClient;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.4
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class R2dbcDslTests {
+
+	@Autowired
+	R2dbcEntityTemplate r2dbcEntityTemplate;
+
+	@Autowired
+	Lifecycle r2dbcInboundChannelAdapter;
+
+	@BeforeEach
+	public void setup() {
+		List<String> statements =
+				Arrays.asList(
+						"DROP TABLE IF EXISTS person;",
+						"CREATE table person (id INT AUTO_INCREMENT NOT NULL, name VARCHAR2, age INT NOT NULL);");
+
+		DatabaseClient databaseClient = this.r2dbcEntityTemplate.getDatabaseClient();
+		statements.forEach(it ->
+				databaseClient.sql(it)
+						.fetch()
+						.rowsUpdated()
+						.as(StepVerifier::create)
+						.expectNextCount(1)
+						.verifyComplete());
+	}
+
+	@Test
+	void testR2DbcDsl() {
+		this.r2dbcInboundChannelAdapter.start();
+
+		this.r2dbcEntityTemplate.insert(new Person("Bob", 35))
+				.then()
+				.as(StepVerifier::create)
+				.verifyComplete();
+
+		await().until(() ->
+				this.r2dbcEntityTemplate.select(Person.class)
+						.matching(Query.query(Criteria.where("age").is(36)))
+						.one()
+						.block(Duration.ofMillis(100)) != null);
+	}
+
+	@Configuration
+	@EnableIntegration
+	@Import(R2dbcDatabaseConfiguration.class)
+	static class R2dbcMessageSourceConfiguration {
+
+		@Bean
+		IntegrationFlow r2dbcDslFlow(R2dbcEntityTemplate r2dbcEntityTemplate) {
+			return IntegrationFlows
+					.from(R2dbc.inboundChannelAdapter(r2dbcEntityTemplate,
+							(selectCreator) ->
+									selectCreator.createSelect("person")
+											.withProjection("*")
+											.withCriteria(Criteria.where("id").is(1)))
+									.expectSingleResult(true)
+									.payloadType(Person.class)
+									.updateSql("UPDATE Person SET id='2' where id = :id")
+									.bindFunction((DatabaseClient.GenericExecuteSpec bindSpec, Person o) ->
+											bindSpec.bind("id", o.getId())),
+							e -> e.poller(p -> p.fixedDelay(100)).autoStartup(false).id("r2dbcInboundChannelAdapter"))
+					.<Mono<?>>handle((p, h) -> p, e -> e.async(true))
+					.channel(MessageChannels.flux())
+					.handle(R2dbc.outboundChannelAdapter(r2dbcEntityTemplate)
+							.queryType(R2dbcMessageHandler.Type.UPDATE)
+							.tableNameExpression("payload.class.simpleName")
+							.criteria((message) -> Criteria.where("id").is(2))
+							.values("{age:36}"))
+					.get();
+		}
+
+	}
+
+}

spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/entity/Person.java

@@ -16,11 +16,14 @@
 
 package org.springframework.integration.r2dbc.entity;
 
+import java.util.Objects;
+
 import org.springframework.data.annotation.Id;
 import org.springframework.data.relational.core.mapping.Table;
 
 /**
  *  @author Rohan Mukesh
+ *  @author Artem Bilan
  *
  *  @since 5.4
  */
@@ -63,4 +66,29 @@ public class Person {
 		return this.age;
 	}
 
+	@Override public boolean equals(Object o) {
+		if (this == o) {
+			return true;
+		}
+		if (o == null || getClass() != o.getClass()) {
+			return false;
+		}
+		Person person = (Person) o;
+		return Objects.equals(this.id, person.id) &&
+				Objects.equals(this.name, person.name) &&
+				Objects.equals(this.age, person.age);
+	}
+
+	@Override public int hashCode() {
+		return Objects.hash(this.id, this.name, this.age);
+	}
+
+	@Override public String toString() {
+		return "Person{" +
+				"id=" + this.id +
+				", name='" + this.name + '\'' +
+				", age=" + this.age +
+				'}';
+	}
+
 }