diff --git a/build.gradle b/build.gradle index ae334510a7..0b318f3ab9 100644 --- a/build.gradle +++ b/build.gradle @@ -105,6 +105,7 @@ ext { springWsVersion = '3.0.9.RELEASE' tomcatVersion = "9.0.36" xstreamVersion = '1.4.12' + r2dbch2Version='0.8.4.RELEASE' javaProjects = subprojects - project(':spring-integration-bom') } @@ -637,6 +638,17 @@ project('spring-integration-mongodb') { } } +project('spring-integration-r2dbc') { + description = 'Spring Integration R2DBC Support' + dependencies { + api project(':spring-integration-core') + api ('org.springframework.data:spring-data-r2dbc') { + exclude group: 'org.springframework' + } + testImplementation "io.r2dbc:r2dbc-h2:$r2dbch2Version" + } +} + project('spring-integration-mqtt') { description = 'Spring Integration MQTT Support' dependencies { 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 new file mode 100644 index 0000000000..779defbb11 --- /dev/null +++ b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java @@ -0,0 +1,251 @@ +/* + * 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.outbound; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.data.r2dbc.core.DatabaseClient; +import org.springframework.data.r2dbc.core.R2dbcEntityOperations; +import org.springframework.data.relational.core.query.Criteria; +import org.springframework.data.relational.core.query.Update; +import org.springframework.data.relational.core.sql.SqlIdentifier; +import org.springframework.expression.Expression; +import org.springframework.expression.TypeLocator; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.expression.spel.support.StandardTypeLocator; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.handler.AbstractReactiveMessageHandler; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +import reactor.core.publisher.Mono; + + +/** + * Implementation of {@link org.springframework.messaging.ReactiveMessageHandler} which writes + * Message payload into a Relational Database, using reactive r2dbc support. + * + * @author Rohan Mukesh + * @author Artem Bilan + * + * @since 5.4 + */ +public class R2dbcMessageHandler extends AbstractReactiveMessageHandler { + + private final R2dbcEntityOperations r2dbcEntityOperations; + + private StandardEvaluationContext evaluationContext; + + private Expression queryTypeExpression = new ValueExpression<>(Type.INSERT); + + @Nullable + private Expression tableNameExpression; + + @Nullable + private Expression valuesExpression; + + @Nullable + private Expression criteriaExpression; + + private volatile boolean initialized = false; + + /** + * Construct this instance using a fully created and initialized instance of provided + * {@link R2dbcEntityOperations} + * @param r2dbcEntityOperations The R2dbcEntityOperations implementation. + */ + public R2dbcMessageHandler(R2dbcEntityOperations r2dbcEntityOperations) { + Assert.notNull(r2dbcEntityOperations, "'r2dbcEntityOperations' must not be null"); + this.r2dbcEntityOperations = r2dbcEntityOperations; + } + + + public void setQueryType(R2dbcMessageHandler.Type type) { + setQueryTypeExpression(new ValueExpression<>(type)); + } + + public void setQueryTypeExpression(Expression queryTypeExpression) { + Assert.notNull(queryTypeExpression, "'queryTypeExpression' must not be null"); + this.queryTypeExpression = queryTypeExpression; + } + + public void setTableName(String tableName) { + setTableNameExpression(new LiteralExpression(tableName)); + } + + public void setTableNameExpression(Expression tableNameExpression) { + this.tableNameExpression = tableNameExpression; + } + + public void setValuesExpression(Expression valuesExpression) { + this.valuesExpression = valuesExpression; + } + + public void setCriteriaExpression(Expression criteriaExpression) { + this.criteriaExpression = criteriaExpression; + } + + + @Override + public String getComponentType() { + return "r2dbc:reactive-outbound-channel-adapter"; + } + + @Override + protected void onInit() { + super.onInit(); + + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); + TypeLocator typeLocator = this.evaluationContext.getTypeLocator(); + if (typeLocator instanceof StandardTypeLocator) { + //Register R2dbc criteria API package so FQCN can be avoided in query-expression. + ((StandardTypeLocator) typeLocator).registerImport("org.springframework.data.relational.core.query"); + } + this.initialized = true; + + } + + @Override + protected Mono handleMessageInternal(Message message) { + Assert.isTrue(this.initialized, "The instance is not yet initialized. Invoke its afterPropertiesSet() method"); + return Mono.fromSupplier(() -> this.queryTypeExpression.getValue(this.evaluationContext, message, Type.class)) + .flatMap(mode -> { + switch (mode) { + case INSERT: + return handleInsert(message); + case UPDATE: + return handleUpdate(message); + case DELETE: + return handleDelete(message); + default: + return Mono.error(new IllegalArgumentException()); + } + }).then(); + } + + + private Mono handleDelete(Message message) { + if (this.tableNameExpression != null) { + String tableName = evaluateTableNameExpression(message); + Criteria criteria = evaluateCriteriaExpression(message); + DatabaseClient.DeleteMatchingSpec deleteSpec = + this.r2dbcEntityOperations.getDatabaseClient() + .delete() + .from(tableName); + return deleteSpec.matching(criteria) + .then(); + } + else { + return this.r2dbcEntityOperations.delete(message.getPayload()) + .then(); + } + } + + private Mono handleUpdate(Message message) { + if (this.tableNameExpression != null) { + String tableName = evaluateTableNameExpression(message); + Map values = evaluateValuesExpression(message); + Map updateMap = transformIntoSqlIdentifierMap(values); + Criteria criteria = evaluateCriteriaExpression(message); + DatabaseClient.GenericUpdateSpec updateSpec = + this.r2dbcEntityOperations.getDatabaseClient().update() + .table(tableName); + return updateSpec.using(Update.from(updateMap)) + .matching(criteria) + .then(); + } + else { + return this.r2dbcEntityOperations.update(message.getPayload()) + .then(); + } + } + + private Map transformIntoSqlIdentifierMap(Map values) { + Map sqlIdentifierObjectMap = new HashMap<>(); + values.forEach((k, v) -> sqlIdentifierObjectMap.put(SqlIdentifier.unquoted(k), v)); + return sqlIdentifierObjectMap; + } + + private Mono handleInsert(Message message) { + if (this.tableNameExpression != null) { + String tableName = evaluateTableNameExpression(message); + Map values = evaluateValuesExpression(message); + DatabaseClient.GenericInsertSpec> insertSpec = + this.r2dbcEntityOperations.getDatabaseClient() + .insert() + .into(tableName); + for (Map.Entry entry : values.entrySet()) { + insertSpec = insertSpec.value(entry.getKey(), entry.getValue()); + } + return insertSpec.then(); + } + else { + return this.r2dbcEntityOperations.insert(message.getPayload()) + .then(); + } + } + + private String evaluateTableNameExpression(Message message) { + String tableName = this.tableNameExpression.getValue(this.evaluationContext, message, String.class); + Assert.notNull(tableName, "'tableNameExpression' must not evaluate to null"); + return tableName; + } + + @SuppressWarnings("unchecked") + private Map evaluateValuesExpression(Message message) { + Map fieldValues = + (Map) this.valuesExpression.getValue(this.evaluationContext, message, Map.class); + Assert.notNull(fieldValues, "'valuesExpression' must not evaluate to null"); + return fieldValues; + } + + private Criteria evaluateCriteriaExpression(Message message) { + Criteria criteria = + this.criteriaExpression.getValue(this.evaluationContext, message, Criteria.class); + Assert.notNull(criteria, "'criteriaExpression' must not evaluate to null"); + return criteria; + } + + + /** + * /** + * The mode for the {@link R2dbcMessageHandler}. + */ + public enum Type { + + /** + * Set a {@link R2dbcMessageHandler} into an {@code insert} mode. + */ + INSERT, + + /** + * Set a {@link R2dbcMessageHandler} into an {@code update} mode. + */ + UPDATE, + + /** + * Set a {@link R2dbcMessageHandler} into a {@code delete} mode. + */ + DELETE, + + } + +} diff --git a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/Person.java b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/Person.java new file mode 100644 index 0000000000..8c19789920 --- /dev/null +++ b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/Person.java @@ -0,0 +1,66 @@ +/* + * 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.outbound; + +import org.springframework.data.annotation.Id; +import org.springframework.data.relational.core.mapping.Table; + +/** + * @author Rohan Mukesh + * + * @since 5.4 + */ +@Table +class Person { + + @Id + Integer id; + + String name; + + Integer age; + + public void setId(Integer id) { + this.id = id; + } + + public void setName(String name) { + this.name = name; + } + + public void setAge(Integer age) { + this.age = age; + } + + Person(String name, Integer age) { + this.name = name; + this.age = age; + } + + public Integer getId() { + return this.id; + } + + public String getName() { + return this.name; + } + + public Integer getAge() { + return this.age; + } + +} diff --git a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/R2DbcMessageHandlerTests.java b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/R2DbcMessageHandlerTests.java new file mode 100644 index 0000000000..4fc0421a96 --- /dev/null +++ b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/R2DbcMessageHandlerTests.java @@ -0,0 +1,354 @@ +/* + * 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.outbound; + + +import static org.mockito.Mockito.mock; + +import java.time.Duration; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import org.junit.Assert; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Answers; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.FilterType; +import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration; +import org.springframework.data.r2dbc.core.DatabaseClient; +import org.springframework.data.r2dbc.core.R2dbcEntityTemplate; +import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories; +import org.springframework.data.relational.core.query.Criteria; +import org.springframework.data.repository.reactive.ReactiveCrudRepository; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import io.r2dbc.h2.H2ConnectionConfiguration; +import io.r2dbc.h2.H2ConnectionFactory; +import io.r2dbc.spi.ConnectionFactory; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Hooks; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +/** + * @author Rohan Mukesh + * + * @since 5.4 + */ +@SpringJUnitConfig +public class R2DbcMessageHandlerTests { + + @Autowired + DatabaseClient client; + + @Autowired + H2ConnectionFactory factory; + + R2dbcEntityTemplate entityTemplate; + + @Autowired + PersonRepository personRepository; + + @Configuration + @EnableR2dbcRepositories(considerNestedRepositories = true, + includeFilters = @ComponentScan.Filter(classes = PersonRepository.class, type = FilterType.ASSIGNABLE_TYPE)) + static class IntegrationTestConfiguration extends AbstractR2dbcConfiguration { + + @Bean + @Override + public ConnectionFactory connectionFactory() { + return createConnectionFactory(); + } + + } + + public static ConnectionFactory createConnectionFactory() { + return new H2ConnectionFactory(H2ConnectionConfiguration.builder() + .inMemory("r2dbc") + .username("sa") + .password("") + .option("DB_CLOSE_DELAY=-1").build()); + } + + @BeforeEach + public void setup() { + entityTemplate = new R2dbcEntityTemplate(client); + Hooks.onOperatorDebug(); + + List statements = Arrays.asList( + "DROP TABLE IF EXISTS person;", + "CREATE table person (id INT AUTO_INCREMENT NOT NULL, name VARCHAR2, age INT NOT NULL);"); + + statements.forEach(it -> client.execute(it) + .fetch() + .rowsUpdated() + .as(StepVerifier::create) + .expectNextCount(1) + .verifyComplete()); + + } + + @Test + public void validateMessageHandlingWithDefaultInsertCollection() { + R2dbcMessageHandler handler = new R2dbcMessageHandler(this.entityTemplate); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); + handler.afterPropertiesSet(); + Message message = MessageBuilder.withPayload(this.createPerson("Bob", 35)).build(); + waitFor(handler.handleMessage(message)); + + personRepository.findAll() + .as(StepVerifier::create) + .expectNextCount(1) + .verifyComplete(); + } + + @Test + public void validateMessageHandlingWithInsertQueryCollection() { + R2dbcMessageHandler handler = new R2dbcMessageHandler(this.entityTemplate); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); + handler.setValuesExpression(new FunctionExpression>(Message::getPayload)); + handler.setQueryType(R2dbcMessageHandler.Type.INSERT); + handler.setTableName("person"); + handler.afterPropertiesSet(); + Map payload = new HashMap<>(); + payload.put("name", "rohan"); + payload.put("age", 35); + Message message = MessageBuilder.withPayload(payload).build(); + waitFor(handler.handleMessage(message)); + + Flux all = client.execute("SELECT name, age FROM person") + .fetch().all(); + all.as(StepVerifier::create) + .expectNextCount(1) + .verifyComplete(); + + } + + @Test + public void validateMessageHandlingWithDefaultUpdateCollection() { + R2dbcMessageHandler handler = new R2dbcMessageHandler(this.entityTemplate); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); + handler.afterPropertiesSet(); + Message message = MessageBuilder.withPayload(this.createPerson("Bob", 35)).build(); + waitFor(handler.handleMessage(message)); + + handler.setQueryType(R2dbcMessageHandler.Type.UPDATE); + + Person person = this.client.select() + .from("person") + .as(Person.class) + .fetch() + .first() + .block(); + + person.setAge(40); + + message = MessageBuilder.withPayload(person) + .build(); + waitFor(handler.handleMessage(message)); + + personRepository.findAll() + .as(StepVerifier::create) + .consumeNextWith(p -> Assert.assertEquals(Optional.of(40), Optional.ofNullable(p.age))) + .verifyComplete(); + } + + @Test + public void validateMessageHandlingWithUpdateQueryCollection() { + R2dbcMessageHandler handler = new R2dbcMessageHandler(this.entityTemplate); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); + handler.setValuesExpression(new FunctionExpression>(Message::getPayload)); + + + handler.setQueryType(R2dbcMessageHandler.Type.INSERT); + handler.setTableName("person"); + handler.afterPropertiesSet(); + Map payload = new HashMap<>(); + payload.put("name", "Bob"); + payload.put("age", 35); + Message message = MessageBuilder.withPayload(payload).build(); + waitFor(handler.handleMessage(message)); + + payload = new HashMap<>(); + payload.put("name", "Rob"); + payload.put("age", 43); + message = MessageBuilder.withPayload(payload).build(); + waitFor(handler.handleMessage(message)); + + payload = new HashMap<>(); + handler.setQueryType(R2dbcMessageHandler.Type.UPDATE); + + Object insertedId = client.execute("SELECT id FROM person") + .fetch() + .first() + .block() + .get("id"); + + handler.setCriteriaExpression(new FunctionExpression>((m) -> Criteria.where("id").is(insertedId))); + payload.put("age", 40); + + message = MessageBuilder.withPayload(payload).build(); + waitFor(handler.handleMessage(message)); + + Flux all = client.execute("SELECT age,name FROM person where age=40") + .fetch().all(); + all.as(StepVerifier::create) + .consumeNextWith(response -> Assert.assertEquals("{AGE=40, NAME=Bob}", response.toString())) + .verifyComplete(); + + } + + @Test + public void validateMessageHandlingWithDefaultDeleteCollection() { + R2dbcMessageHandler handler = new R2dbcMessageHandler(this.entityTemplate); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); + handler.afterPropertiesSet(); + Message message = MessageBuilder.withPayload(this.createPerson("Bob", 35)).build(); + waitFor(handler.handleMessage(message)); + + Person person = this.client + .select() + .from("person") + .as(Person.class) + .fetch() + .first() + .block(); + + handler.setQueryType(R2dbcMessageHandler.Type.DELETE); + message = MessageBuilder.withPayload(person).build(); + waitFor(handler.handleMessage(message)); + + personRepository.findAll() + .as(StepVerifier::create) + .expectNextCount(0) + .verifyComplete(); + } + + @Test + public void validateMessageHandlingWithDeleteQueryCollection() { + R2dbcMessageHandler handler = new R2dbcMessageHandler(this.entityTemplate); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); + handler.setValuesExpression(new FunctionExpression>(Message::getPayload)); + + handler.setQueryType(R2dbcMessageHandler.Type.INSERT); + handler.setTableName("person"); + handler.afterPropertiesSet(); + Map payload = new HashMap<>(); + payload.put("name", "Bob"); + payload.put("age", 35); + Message message = MessageBuilder.withPayload(payload).build(); + waitFor(handler.handleMessage(message)); + + payload = new HashMap<>(); + handler.setQueryType(R2dbcMessageHandler.Type.DELETE); + + Object insertedId = client.execute("SELECT id FROM person") + .fetch() + .first() + .block() + .get("id"); + + handler.setCriteriaExpression(new FunctionExpression>((m) -> Criteria.where("id").is(insertedId))); + message = MessageBuilder.withPayload(payload).build(); + waitFor(handler.handleMessage(message)); + + Flux all = client.execute("SELECT age,name FROM person where age=35") + .fetch().all(); + all.as(StepVerifier::create) + .expectNextCount(0) + .verifyComplete(); + + } + + @Test + public void validateMessageHandlingWithDeleteQueryCollection_MultipleRows() { + R2dbcMessageHandler handler = new R2dbcMessageHandler(this.entityTemplate); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); + handler.setValuesExpression(new FunctionExpression>(Message::getPayload)); + + handler.setQueryType(R2dbcMessageHandler.Type.INSERT); + handler.setTableName("person"); + handler.afterPropertiesSet(); + Map payload = new HashMap<>(); + payload.put("name", "Bob"); + payload.put("age", 35); + Message message = MessageBuilder.withPayload(payload).build(); + waitFor(handler.handleMessage(message)); + + payload = new HashMap<>(); + payload.put("name", "Rob"); + payload.put("age", 40); + message = MessageBuilder.withPayload(payload).build(); + waitFor(handler.handleMessage(message)); + + payload = new HashMap<>(); + handler.setQueryType(R2dbcMessageHandler.Type.DELETE); + + Object insertedId = client.execute("SELECT id FROM person") + .fetch() + .first() + .block() + .get("id"); + + handler.setCriteriaExpression(new FunctionExpression>((m) -> Criteria.where("id").is(insertedId))); + message = MessageBuilder.withPayload(payload).build(); + waitFor(handler.handleMessage(message)); + + Flux all = client.execute("SELECT age,name FROM person where age=40") + .fetch().all(); + all.as(StepVerifier::create) + .expectNextCount(1) + .verifyComplete(); + + } + + + private Person createPerson(String bob, Integer age) { + return new Person(bob, age); + } + + private static T waitFor(Mono mono) { + return mono.block(Duration.ofSeconds(10)); + } + + interface PersonRepository extends ReactiveCrudRepository { + + } + +} + + diff --git a/src/reference/asciidoc/index-single.adoc b/src/reference/asciidoc/index-single.adoc index 47ee42c766..34d8634a9a 100644 --- a/src/reference/asciidoc/index-single.adoc +++ b/src/reference/asciidoc/index-single.adoc @@ -57,6 +57,8 @@ include::./mongodb.adoc[] include::./mqtt.adoc[] +include::./r2dbc.adoc[] + include::./redis.adoc[] include::./resource.adoc[] diff --git a/src/reference/asciidoc/index.adoc b/src/reference/asciidoc/index.adoc index dbd17d66ce..cb2f4643da 100644 --- a/src/reference/asciidoc/index.adoc +++ b/src/reference/asciidoc/index.adoc @@ -38,6 +38,7 @@ This documentation is also available as single searchable link:index-single.html <<./mail.adoc#mail,Mail Support>> :: <<./mongodb.adoc#mongodb,MongoDb Support>> :: <<./mqtt.adoc#mqtt,MQTT Support>> :: +<<./r2dbc.adoc#r2dbc,R2DBC Support>> :: <<./redis.adoc#redis,Redis Support>> :: <<./resource.adoc#resource,Resource Support>> :: <<./rmi.adoc#rmi,RMI Support>> :: diff --git a/src/reference/asciidoc/r2dbc.adoc b/src/reference/asciidoc/r2dbc.adoc new file mode 100644 index 0000000000..10666c22b3 --- /dev/null +++ b/src/reference/asciidoc/r2dbc.adoc @@ -0,0 +1,34 @@ +[[r2dbc]] +== R2DBC Support + +Spring Integration provides channel adapters for receiving and sending messages by using reactive access to databases via https://r2dbc.io/[R2DBC] drivers. + +You need to include this dependency into your project: + +==== +.Maven +[source, xml, subs="normal"] +---- + + org.springframework.integration + spring-integration-r2dbc + {project-version} + +---- + +.Gradle +[source, groovy, subs="normal"] +---- +compile "org.springframework.integration:spring-integration-r2dbc:{project-version}" +---- +==== + +[[r2dbc-inbound-channel-adapter]] +=== Inbound Channel Adapter + +TBD + +[[r2dbc-outbound-channel-adapter]] +=== Outbound Channel Adapter + +TBD diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e523d9ba9b..ae3de2f627 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -20,6 +20,11 @@ If you are interested in more details, see the Issue Tracker tickets that were r The standalone https://projects.spring.io/spring-integration-kafka/[Spring Integration Kafka] project has been merged as a `spring-integration-kafka` module to this project. See <<./kafka.adoc#kafka,Spring for Apache Kafka Support>> for more information. +==== R2DBC Channel Adapters + +The Channel Adapters for R2DBC database interaction have been introduced. +See <<./r2dbc.adoc#r2dbc,R2DBC Support>> for more information. + [[x5.4-general]] === General Changes