INT-4566: R2DBC Outbound Channel Adapter
JIRA: https://jira.spring.io/browse/INT-4566 * Fixed review comments * added `DELETE` and `Criteria` implementation for outbound channel adapter * Clean up code style * Add initial docs
This commit is contained in:
fnu, rohan
12
build.gradle
12
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 {
|
||||
|
||||
@@ -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<Void> 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<Void> 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<Void> handleUpdate(Message<?> message) {
|
||||
if (this.tableNameExpression != null) {
|
||||
String tableName = evaluateTableNameExpression(message);
|
||||
Map<String, Object> values = evaluateValuesExpression(message);
|
||||
Map<SqlIdentifier, Object> 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<SqlIdentifier, Object> transformIntoSqlIdentifierMap(Map<String, Object> values) {
|
||||
Map<SqlIdentifier, Object> sqlIdentifierObjectMap = new HashMap<>();
|
||||
values.forEach((k, v) -> sqlIdentifierObjectMap.put(SqlIdentifier.unquoted(k), v));
|
||||
return sqlIdentifierObjectMap;
|
||||
}
|
||||
|
||||
private Mono<Void> handleInsert(Message<?> message) {
|
||||
if (this.tableNameExpression != null) {
|
||||
String tableName = evaluateTableNameExpression(message);
|
||||
Map<String, Object> values = evaluateValuesExpression(message);
|
||||
DatabaseClient.GenericInsertSpec<Map<String, Object>> insertSpec =
|
||||
this.r2dbcEntityOperations.getDatabaseClient()
|
||||
.insert()
|
||||
.into(tableName);
|
||||
for (Map.Entry<String, Object> 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<String, Object> evaluateValuesExpression(Message<?> message) {
|
||||
Map<String, Object> fieldValues =
|
||||
(Map<String, Object>) 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,
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String> 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<Person> 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<?>>(Message::getPayload));
|
||||
handler.setQueryType(R2dbcMessageHandler.Type.INSERT);
|
||||
handler.setTableName("person");
|
||||
handler.afterPropertiesSet();
|
||||
Map<String, Object> 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<Person> 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<?>>(Message::getPayload));
|
||||
|
||||
|
||||
handler.setQueryType(R2dbcMessageHandler.Type.INSERT);
|
||||
handler.setTableName("person");
|
||||
handler.afterPropertiesSet();
|
||||
Map<String, Object> 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<Message<?>>((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<Person> 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<?>>(Message::getPayload));
|
||||
|
||||
handler.setQueryType(R2dbcMessageHandler.Type.INSERT);
|
||||
handler.setTableName("person");
|
||||
handler.afterPropertiesSet();
|
||||
Map<String, Object> 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<Message<?>>((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<?>>(Message::getPayload));
|
||||
|
||||
handler.setQueryType(R2dbcMessageHandler.Type.INSERT);
|
||||
handler.setTableName("person");
|
||||
handler.afterPropertiesSet();
|
||||
Map<String, Object> 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<Message<?>>((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> T waitFor(Mono<T> mono) {
|
||||
return mono.block(Duration.ofSeconds(10));
|
||||
}
|
||||
|
||||
interface PersonRepository extends ReactiveCrudRepository<Person, Integer> {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
@@ -57,6 +57,8 @@ include::./mongodb.adoc[]
|
||||
|
||||
include::./mqtt.adoc[]
|
||||
|
||||
include::./r2dbc.adoc[]
|
||||
|
||||
include::./redis.adoc[]
|
||||
|
||||
include::./resource.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>> ::
|
||||
|
||||
34
src/reference/asciidoc/r2dbc.adoc
Normal file
34
src/reference/asciidoc/r2dbc.adoc
Normal file
@@ -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"]
|
||||
----
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-r2dbc</artifactId>
|
||||
<version>{project-version}</version>
|
||||
</dependency>
|
||||
----
|
||||
|
||||
.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
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user