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 new file mode 100644 index 0000000000..d003e26d5d --- /dev/null +++ b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSource.java @@ -0,0 +1,179 @@ +/* + * 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.inbound; + + +import java.util.Map; + +import org.reactivestreams.Publisher; + +import org.springframework.data.r2dbc.core.DatabaseClient; +import org.springframework.data.r2dbc.core.FetchSpec; +import org.springframework.data.r2dbc.core.R2dbcEntityOperations; +import org.springframework.data.relational.core.query.Query; +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.endpoint.AbstractMessageSource; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.util.Assert; + +import reactor.core.publisher.Mono; + +/** + * An instance of {@link org.springframework.integration.core.MessageSource} which returns + * a {@link org.springframework.messaging.Message} with a payload which is the result of + * execution of a {@link Query}. When {@code expectSingleResult} is false (default), the R2dbc + * {@link Query} is executed using {@link R2dbcEntityOperations#select(Query, Class)} method which + * returns a {@link reactor.core.publisher.Flux}. + * The returned {@link reactor.core.publisher.Flux} will be used as the payload of the + * {@link org.springframework.messaging.Message} returned by the {@link #receive()} + * method. + *

+ * When {@code expectSingleResult} is true, the {@link R2dbcEntityOperations#selectOne(Query, Class)} is + * used instead, and the message payload will be a {@link reactor.core.publisher.Mono} + * for the single object returned from the query. + * + * @author Rohan Mukesh + * @author Artem Bilan + * + * @since 5.4 + */ +public class R2dbcMessageSource extends AbstractMessageSource> { + + private final DatabaseClient databaseClient; + + private final Expression queryExpression; + + private Class payloadType = Map.class; + + private boolean expectSingleResult = false; + + private StandardEvaluationContext evaluationContext; + + private volatile boolean initialized = false; + + /** + * Create an instance with the provided {@link DatabaseClient} and SpEL expression + * which should resolve to a Relational 'query' string. + * It assumes that the {@link DatabaseClient} is fully initialized and ready to be used. + * The 'query' will be evaluated on every call to the {@link #receive()} method. + * @param databaseClient The reactive database client for performing database calls. + * @param query The query String. + */ + public R2dbcMessageSource(DatabaseClient databaseClient, String query) { + this(databaseClient, new LiteralExpression(query)); + } + + /** + * Create an instance with the provided {@link DatabaseClient} and SpEL expression + * which should resolve to a Relational 'query' string. + * It assumes that the {@link DatabaseClient} is fully initialized and ready to be used. + * The 'queryExpression' will be evaluated on every call to the {@link #receive()} method. + * @param databaseClient The reactive for performing database calls. + * @param queryExpression The query expression. + */ + public R2dbcMessageSource(DatabaseClient databaseClient, Expression queryExpression) { + Assert.notNull(databaseClient, "'databaseClient' must not be null"); + Assert.notNull(queryExpression, "'queryExpression' must not be null"); + this.databaseClient = databaseClient; + this.queryExpression = queryExpression; + } + + /** + * Provide a way to set the type of the entityClass that will be passed to the + * {@link org.springframework.data.r2dbc.core.DatabaseClient#execute(String)} + * method. + * @param payloadType The t class. + */ + public void setPayloadType(Class payloadType) { + Assert.notNull(payloadType, "'payloadType' must not be null"); + this.payloadType = payloadType; + } + + /** + * Provide a way to manage which find* method to invoke on {@link R2dbcEntityOperations}. + * Default is 'false', which means the {@link #receive()} method will use + * the {@link org.springframework.data.r2dbc.core.DatabaseClient#execute(String)} method and will fetch all. If set + * to 'true'{@link #receive()} will use {@link org.springframework.data.r2dbc.core.DatabaseClient#execute(String)} + * and will fetch one and the payload of the returned {@link org.springframework.messaging.Message} + * will be the returned target Object of type + * identified by {@link #payloadType} instead of a List. + * @param expectSingleResult true if a single result is expected. + */ + public void setExpectSingleResult(boolean expectSingleResult) { + this.expectSingleResult = expectSingleResult; + } + + @Override + public String getComponentType() { + return "r2dbc:inbound-channel-adapter"; + } + + @Override + protected void onInit() { + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); + TypeLocator typeLocator = this.evaluationContext.getTypeLocator(); + if (typeLocator instanceof StandardTypeLocator) { + /* + * Register the R2dbc Query DSL package so they don't need a FQCN for QueryBuilder, for example. + */ + ((StandardTypeLocator) typeLocator).registerImport("org.springframework.data.relational.core.query"); + } + this.initialized = true; + } + + /** + * Execute a {@link Query} returning its results as the Message payload. + * The payload can be either {@link reactor.core.publisher.Flux} or + * {@link reactor.core.publisher.Mono} of objects of type identified by {@link #payloadType}, + * 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 + */ + @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)) + .map(this::prepareFetch); + if (this.expectSingleResult) { + return queryMono.flatMap(FetchSpec::one); + } + return queryMono.flatMapMany(FetchSpec::all); + } + + private FetchSpec prepareFetch(Object queryObject) { + String queryString = evaluateQueryObject(queryObject); + return this.databaseClient + .execute(queryString) + .as(this.payloadType) + .fetch(); + } + + private String evaluateQueryObject(Object queryObject) { + if (queryObject instanceof String) { + return (String) queryObject; + } + throw new IllegalStateException("'queryExpression' must evaluate to String " + + "or org.springframework.data.relational.core.query.Query, but not: " + queryObject); + } + +} diff --git a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/config/R2dbcDatabaseConfiguration.java b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/config/R2dbcDatabaseConfiguration.java new file mode 100644 index 0000000000..8e3f97d34f --- /dev/null +++ b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/config/R2dbcDatabaseConfiguration.java @@ -0,0 +1,59 @@ +/* + * 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.config; + + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration; +import org.springframework.data.r2dbc.core.DatabaseClient; +import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories; + +import io.r2dbc.h2.H2ConnectionConfiguration; +import io.r2dbc.h2.H2ConnectionFactory; +import io.r2dbc.spi.ConnectionFactory; + +/** + * @author Rohan Mukesh + * + * @since 5.4 + */ +@Configuration +@EnableR2dbcRepositories(basePackages = "org.springframework.integration.r2dbc.repository") +public class R2dbcDatabaseConfiguration 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()); + } + + @Bean + public DatabaseClient databaseClient(ConnectionFactory connectionFactory) { + return DatabaseClient.create(connectionFactory); + } + +} 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/entity/Person.java similarity index 86% rename from spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/Person.java rename to spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/entity/Person.java index 8c19789920..83b273d21e 100644 --- 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/entity/Person.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.r2dbc.outbound; +package org.springframework.integration.r2dbc.entity; import org.springframework.data.annotation.Id; import org.springframework.data.relational.core.mapping.Table; @@ -25,14 +25,14 @@ import org.springframework.data.relational.core.mapping.Table; * @since 5.4 */ @Table -class Person { +public class Person { @Id - Integer id; + private Integer id; - String name; + private String name; - Integer age; + private Integer age; public void setId(Integer id) { this.id = id; @@ -46,7 +46,7 @@ class Person { this.age = age; } - Person(String name, Integer age) { + public Person(String name, Integer age) { this.name = name; this.age = age; } diff --git a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSourceTests.java b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSourceTests.java new file mode 100644 index 0000000000..67e1199282 --- /dev/null +++ b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSourceTests.java @@ -0,0 +1,173 @@ +/* + * 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.inbound; + + +import static org.assertj.core.api.Assertions.assertThat; + +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.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.data.r2dbc.core.DatabaseClient; +import org.springframework.data.r2dbc.core.R2dbcEntityTemplate; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.r2dbc.config.R2dbcDatabaseConfiguration; +import org.springframework.integration.r2dbc.entity.Person; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Hooks; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +/** + * @author Rohan Mukesh + * @author Artem Bilan + * + * @since 5.4 + */ +@SpringJUnitConfig +@DirtiesContext +public class R2dbcMessageSourceTests { + + @Autowired + DatabaseClient client; + + R2dbcEntityTemplate entityTemplate; + + @Autowired + R2dbcMessageSource r2dbcMessageSourceSelectOne; + + @Autowired + R2dbcMessageSource r2dbcMessageSourceSelectMany; + + @Autowired + R2dbcMessageSource r2dbcMessageSourceError; + + @BeforeEach + public void setup() { + Hooks.onOperatorDebug(); + entityTemplate = new R2dbcEntityTemplate(this.client); + 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 -> this.client.execute(it) + .fetch() + .rowsUpdated() + .as(StepVerifier::create) + .expectNextCount(1) + .verifyComplete()); + } + + @Test + public void validateSuccessfulQueryWithoutSettingExpectedElement() { + this.entityTemplate.insert(new Person("Bob", 35)) + .then() + .as(StepVerifier::create) + .verifyComplete(); + + StepVerifier.create((Flux) r2dbcMessageSourceSelectMany.receive().getPayload()) + .assertNext(person -> assertThat(((Person) person).getName()).isEqualTo("Bob")) + .verifyComplete(); + } + + + @Test + public void validateSuccessfulQueryWithSingleElementOfMonoDBObject() { + this.entityTemplate.insert(new Person("Bob", 35)) + .then() + .as(StepVerifier::create) + .verifyComplete(); + + r2dbcMessageSourceSelectOne.setExpectSingleResult(true); + + StepVerifier.create((Mono) r2dbcMessageSourceSelectOne.receive().getPayload()) + .assertNext(person -> assertThat(((Person) person).getName()).isEqualTo("Bob")) + .verifyComplete(); + + } + + @Test + public void validateSuccessfulQueryWithMultipleElementOfFluxDBObject() { + this.entityTemplate.insert(new Person("Bob", 35)) + .then() + .as(StepVerifier::create) + .verifyComplete(); + + this.entityTemplate.insert(new Person("Tom", 40)) + .then() + .as(StepVerifier::create) + .verifyComplete(); + + StepVerifier.create((Flux) r2dbcMessageSourceSelectMany.receive().getPayload()) + .assertNext(person -> assertThat(((Person) person).getName()).isEqualTo("Bob")) + .assertNext(person -> assertThat(((Person) person).getName()).isEqualTo("Tom")) + .verifyComplete(); + + } + + @Test + public void testAnyOtherObjectQueryExpression() { + + StepVerifier.create((Flux) r2dbcMessageSourceError.receive().getPayload()) + .expectErrorMatches(throwable -> throwable instanceof IllegalStateException + && throwable.getMessage().contains("'queryExpression' must evaluate to String or")) + .verify(); + } + + @Configuration + @Import(R2dbcDatabaseConfiguration.class) + static class R2dbcMessageSourceConfiguration { + + @Autowired + DatabaseClient databaseClient; + + @Bean + public R2dbcMessageSource r2dbcMessageSourceSelectOne() { + R2dbcMessageSource r2dbcMessageSource = new R2dbcMessageSource(databaseClient, + "select * from person Where id = 1"); + r2dbcMessageSource.setPayloadType(Person.class); + return r2dbcMessageSource; + } + + @Bean + public R2dbcMessageSource r2dbcMessageSourceSelectMany() { + R2dbcMessageSource r2dbcMessageSource = new R2dbcMessageSource(databaseClient, "select * from person"); + r2dbcMessageSource.setPayloadType(Person.class); + return r2dbcMessageSource; + } + + @Bean + public R2dbcMessageSource r2dbcMessageSourceError() { + R2dbcMessageSource r2dbcMessageSource = new R2dbcMessageSource(databaseClient, + new ValueExpression<>(new Object())); + r2dbcMessageSource.setPayloadType(Person.class); + return r2dbcMessageSource; + } + + } + +} 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 similarity index 52% rename from spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/R2DbcMessageHandlerTests.java rename to spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandlerTests.java index 4fc0421a96..9c754aefe8 100644 --- 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 @@ -17,8 +17,6 @@ package org.springframework.integration.r2dbc.outbound; -import static org.mockito.Mockito.mock; - import java.time.Duration; import java.util.Arrays; import java.util.HashMap; @@ -29,79 +27,51 @@ 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.context.annotation.Import; 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.r2dbc.config.R2dbcDatabaseConfiguration; +import org.springframework.integration.r2dbc.entity.Person; +import org.springframework.integration.r2dbc.repository.PersonRepository; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; +import org.springframework.test.annotation.DirtiesContext; 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 + * @author Rohan Mukesh * - * @since 5.4 + * @since 5.4 */ @SpringJUnitConfig -public class R2DbcMessageHandlerTests { +@DirtiesContext +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()); - } + @Autowired + R2dbcMessageHandler r2dbcMessageHandler; @BeforeEach public void setup() { - entityTemplate = new R2dbcEntityTemplate(client); Hooks.onOperatorDebug(); - + r2dbcMessageHandler.setQueryType(R2dbcMessageHandler.Type.INSERT); + r2dbcMessageHandler.setTableNameExpression(null); List statements = Arrays.asList( "DROP TABLE IF EXISTS person;", "CREATE table person (id INT AUTO_INCREMENT NOT NULL, name VARCHAR2, age INT NOT NULL);"); @@ -112,17 +82,12 @@ public class R2DbcMessageHandlerTests { .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)); + Message message = MessageBuilder.withPayload(createPerson("Bob", 35)).build(); + waitFor(r2dbcMessageHandler.handleMessage(message)); personRepository.findAll() .as(StepVerifier::create) @@ -132,18 +97,14 @@ public class R2DbcMessageHandlerTests { @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(); + r2dbcMessageHandler.setValuesExpression(new FunctionExpression>(Message::getPayload)); + r2dbcMessageHandler.setQueryType(R2dbcMessageHandler.Type.INSERT); + r2dbcMessageHandler.setTableName("person"); Map payload = new HashMap<>(); payload.put("name", "rohan"); payload.put("age", 35); Message message = MessageBuilder.withPayload(payload).build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); Flux all = client.execute("SELECT name, age FROM person") .fetch().all(); @@ -155,14 +116,10 @@ public class R2DbcMessageHandlerTests { @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)); + Message message = MessageBuilder.withPayload(createPerson("Bob", 35)).build(); + waitFor(r2dbcMessageHandler.handleMessage(message)); - handler.setQueryType(R2dbcMessageHandler.Type.UPDATE); + r2dbcMessageHandler.setQueryType(R2dbcMessageHandler.Type.UPDATE); Person person = this.client.select() .from("person") @@ -175,39 +132,33 @@ public class R2DbcMessageHandlerTests { message = MessageBuilder.withPayload(person) .build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); personRepository.findAll() .as(StepVerifier::create) - .consumeNextWith(p -> Assert.assertEquals(Optional.of(40), Optional.ofNullable(p.age))) + .consumeNextWith(p -> Assert.assertEquals(Optional.of(40), Optional.ofNullable(p.getAge()))) .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(); + r2dbcMessageHandler.setValuesExpression(new FunctionExpression>(Message::getPayload)); + r2dbcMessageHandler.setQueryType(R2dbcMessageHandler.Type.INSERT); + r2dbcMessageHandler.setTableName("person"); Map payload = new HashMap<>(); payload.put("name", "Bob"); payload.put("age", 35); Message message = MessageBuilder.withPayload(payload).build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); payload = new HashMap<>(); payload.put("name", "Rob"); payload.put("age", 43); message = MessageBuilder.withPayload(payload).build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); payload = new HashMap<>(); - handler.setQueryType(R2dbcMessageHandler.Type.UPDATE); + r2dbcMessageHandler.setQueryType(R2dbcMessageHandler.Type.UPDATE); Object insertedId = client.execute("SELECT id FROM person") .fetch() @@ -215,11 +166,12 @@ public class R2DbcMessageHandlerTests { .block() .get("id"); - handler.setCriteriaExpression(new FunctionExpression>((m) -> Criteria.where("id").is(insertedId))); + r2dbcMessageHandler.setCriteriaExpression( + new FunctionExpression>((m) -> Criteria.where("id").is(insertedId))); payload.put("age", 40); message = MessageBuilder.withPayload(payload).build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); Flux all = client.execute("SELECT age,name FROM person where age=40") .fetch().all(); @@ -231,12 +183,8 @@ public class R2DbcMessageHandlerTests { @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)); + Message message = MessageBuilder.withPayload(createPerson("Bob", 35)).build(); + waitFor(r2dbcMessageHandler.handleMessage(message)); Person person = this.client .select() @@ -246,9 +194,9 @@ public class R2DbcMessageHandlerTests { .first() .block(); - handler.setQueryType(R2dbcMessageHandler.Type.DELETE); + r2dbcMessageHandler.setQueryType(R2dbcMessageHandler.Type.DELETE); message = MessageBuilder.withPayload(person).build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); personRepository.findAll() .as(StepVerifier::create) @@ -258,22 +206,17 @@ public class R2DbcMessageHandlerTests { @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(); + r2dbcMessageHandler.setValuesExpression(new FunctionExpression>(Message::getPayload)); + r2dbcMessageHandler.setQueryType(R2dbcMessageHandler.Type.INSERT); + r2dbcMessageHandler.setTableName("person"); Map payload = new HashMap<>(); payload.put("name", "Bob"); payload.put("age", 35); Message message = MessageBuilder.withPayload(payload).build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); payload = new HashMap<>(); - handler.setQueryType(R2dbcMessageHandler.Type.DELETE); + r2dbcMessageHandler.setQueryType(R2dbcMessageHandler.Type.DELETE); Object insertedId = client.execute("SELECT id FROM person") .fetch() @@ -281,9 +224,10 @@ public class R2DbcMessageHandlerTests { .block() .get("id"); - handler.setCriteriaExpression(new FunctionExpression>((m) -> Criteria.where("id").is(insertedId))); + r2dbcMessageHandler.setCriteriaExpression( + new FunctionExpression>((m) -> Criteria.where("id").is(insertedId))); message = MessageBuilder.withPayload(payload).build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); Flux all = client.execute("SELECT age,name FROM person where age=35") .fetch().all(); @@ -295,28 +239,23 @@ public class R2DbcMessageHandlerTests { @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(); + r2dbcMessageHandler.setValuesExpression(new FunctionExpression>(Message::getPayload)); + r2dbcMessageHandler.setQueryType(R2dbcMessageHandler.Type.INSERT); + r2dbcMessageHandler.setTableName("person"); Map payload = new HashMap<>(); payload.put("name", "Bob"); payload.put("age", 35); Message message = MessageBuilder.withPayload(payload).build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); payload = new HashMap<>(); payload.put("name", "Rob"); payload.put("age", 40); message = MessageBuilder.withPayload(payload).build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); payload = new HashMap<>(); - handler.setQueryType(R2dbcMessageHandler.Type.DELETE); + r2dbcMessageHandler.setQueryType(R2dbcMessageHandler.Type.DELETE); Object insertedId = client.execute("SELECT id FROM person") .fetch() @@ -324,9 +263,10 @@ public class R2DbcMessageHandlerTests { .block() .get("id"); - handler.setCriteriaExpression(new FunctionExpression>((m) -> Criteria.where("id").is(insertedId))); + r2dbcMessageHandler.setCriteriaExpression( + new FunctionExpression>((m) -> Criteria.where("id").is(insertedId))); message = MessageBuilder.withPayload(payload).build(); - waitFor(handler.handleMessage(message)); + waitFor(r2dbcMessageHandler.handleMessage(message)); Flux all = client.execute("SELECT age,name FROM person where age=40") .fetch().all(); @@ -345,7 +285,18 @@ public class R2DbcMessageHandlerTests { return mono.block(Duration.ofSeconds(10)); } - interface PersonRepository extends ReactiveCrudRepository { + + @Configuration + @Import(R2dbcDatabaseConfiguration.class) + static class R2dbcMessageHandlerConfiguration { + + @Autowired + DatabaseClient databaseClient; + + @Bean + public R2dbcMessageHandler r2dbcMessageHandler() { + return new R2dbcMessageHandler(new R2dbcEntityTemplate(databaseClient)); + } } diff --git a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/repository/PersonRepository.java b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/repository/PersonRepository.java new file mode 100644 index 0000000000..53fa60ea13 --- /dev/null +++ b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/repository/PersonRepository.java @@ -0,0 +1,29 @@ +/* + * 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.repository; + +import org.springframework.data.repository.reactive.ReactiveCrudRepository; +import org.springframework.integration.r2dbc.entity.Person; + +/** + * @author Rohan Mukesh + * + * @since 5.4 + */ +public interface PersonRepository extends ReactiveCrudRepository { + +}