From 5221ac1740b4a1857253060eeb4a1f34fadec5d9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 26 Oct 2020 16:10:34 -0400 Subject: [PATCH] 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 | 1 + .../integration/r2dbc/dsl/R2dbc.java | 87 ++++++++ .../r2dbc/dsl/R2dbcMessageHandlerSpec.java | 185 ++++++++++++++++++ .../r2dbc/dsl/R2dbcMessageSourceSpec.java | 86 ++++++++ .../integration/r2dbc/dsl/package-info.java | 5 + .../r2dbc/inbound/R2dbcMessageSource.java | 49 +++-- .../r2dbc/inbound/package-info.java | 5 + .../r2dbc/outbound/R2dbcMessageHandler.java | 25 +++ .../r2dbc/outbound/package-info.java | 5 + .../integration/r2dbc/dsl/R2dbcDslTests.java | 129 ++++++++++++ .../integration/r2dbc/entity/Person.java | 28 +++ 11 files changed, 590 insertions(+), 15 deletions(-) create mode 100644 spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbc.java create mode 100644 spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbcMessageHandlerSpec.java create mode 100644 spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbcMessageSourceSpec.java create mode 100644 spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/package-info.java create mode 100644 spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/package-info.java create mode 100644 spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/package-info.java create mode 100644 spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/dsl/R2dbcDslTests.java diff --git a/build.gradle b/build.gradle index b70e7e72a7..ac4227f5f6 100644 --- a/build.gradle +++ b/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" } } diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbc.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbc.java new file mode 100644 index 0000000000..153555f7a9 --- /dev/null +++ b/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 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() { + } + +} diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbcMessageHandlerSpec.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbcMessageHandlerSpec.java new file mode 100644 index 0000000000..20e789d62d --- /dev/null +++ b/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 { + + 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

the payload type. + * @return the spec + */ + public

R2dbcMessageHandlerSpec queryTypeFunction( + Function, 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

the payload type. + * @return the spec + */ + public

R2dbcMessageHandlerSpec tableNameFunction(Function, 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

the payload type. + * @return the spec + */ + public

R2dbcMessageHandlerSpec values(Function, Map> 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

the payload type. + * @return the spec + */ + public

R2dbcMessageHandlerSpec criteria(Function, 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; + } + +} diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbcMessageSourceSpec.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/R2dbcMessageSourceSpec.java new file mode 100644 index 0000000000..e0b7aa2733 --- /dev/null +++ b/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 { + + 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 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; + } + +} diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/package-info.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/dsl/package-info.java new file mode 100644 index 0000000000..043505dda2 --- /dev/null +++ b/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; diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSource.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSource.java index 1d5960049e..cea52b5d33 100644 --- a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSource.java +++ b/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> { 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> { /** * 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> { } /** - * 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> { } /** - * 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> { } /** - * 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> { * 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> 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> { "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); + } + + } + } diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/package-info.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/package-info.java new file mode 100644 index 0000000000..501e1ccd15 --- /dev/null +++ b/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; diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java index d60f76c5b3..810898c1e0 100644 --- a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java +++ b/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; } diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/package-info.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/package-info.java new file mode 100644 index 0000000000..20c7e8a1b0 --- /dev/null +++ b/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; diff --git a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/dsl/R2dbcDslTests.java b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/dsl/R2dbcDslTests.java new file mode 100644 index 0000000000..df734bf567 --- /dev/null +++ b/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 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")) + .>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(); + } + + } + +} diff --git a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/entity/Person.java b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/entity/Person.java index 83b273d21e..3a1dcaec7d 100644 --- a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/entity/Person.java +++ b/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 + + '}'; + } + }