From af5bcdf4d7dc398e685ca3ea848a351768eea1b2 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 21 Aug 2020 14:57:56 -0400 Subject: [PATCH] Honor Emission result in FluxMessageChannel * Implement `Emission` handling in the `FluxMessageChannel` and `IntegrationReactiveUtils` * Upgrade to Spring Kafka `2.6.0` * Fix R2DBC components for deprecation in the Spring Data R2DBC * Implement `StatementMapper.SelectSpec` for query expression * Clean up for some sporadic test failures --- build.gradle | 3 +- .../channel/FluxMessageChannel.java | 37 +++++++++- .../util/IntegrationReactiveUtils.java | 19 +++-- .../ReactiveStreamsConsumerTests.java | 2 - ...ChannelAdapterIntegrationTests-context.xml | 70 +++++++++---------- ...utboundChannelAdapterIntegrationTests.java | 9 +-- ...dGatewayWithNamespaceIntegrationTests.java | 39 +++++------ .../r2dbc/inbound/R2dbcMessageSource.java | 51 +++++++------- .../r2dbc/outbound/R2dbcMessageHandler.java | 1 + 9 files changed, 134 insertions(+), 97 deletions(-) diff --git a/build.gradle b/build.gradle index ce98511dbd..df816be948 100644 --- a/build.gradle +++ b/build.gradle @@ -100,7 +100,7 @@ ext { soapVersion = '1.4.0' springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.3.0-SNAPSHOT' springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2020.0.0-SNAPSHOT' - springKafkaVersion = '2.6.0-SNAPSHOT' + springKafkaVersion = '2.6.0' springRetryVersion = '1.3.0' springSecurityVersion = project.hasProperty('springSecurityVersion') ? project.springSecurityVersion : '5.4.0-RC1' springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.3.0-SNAPSHOT' @@ -199,7 +199,6 @@ configure(javaProjects) { subproject -> compileKotlin { kotlinOptions { jvmTarget = '1.8' - allWarningsAsErrors = true } } compileTestKotlin { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 7f6eaa3e1b..26b754eecd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2015-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. @@ -16,6 +16,9 @@ package org.springframework.integration.channel; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; + import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; @@ -50,6 +53,8 @@ public class FluxMessageChannel extends AbstractMessageChannel private final Disposable.Composite upstreamSubscriptions = Disposables.composite(); + private volatile boolean active = true; + public FluxMessageChannel() { this.sink = Sinks.many().multicast().onBackpressureBuffer(1, false); this.processor = FluxProcessor.fromSink(this.sink); @@ -57,9 +62,34 @@ public class FluxMessageChannel extends AbstractMessageChannel @Override protected boolean doSend(Message message, long timeout) { - Assert.state(this.processor.hasDownstreams(), + Assert.state(this.active && this.processor.hasDownstreams(), () -> "The [" + this + "] doesn't have subscribers to accept messages"); - return this.sink.tryEmitNext(message).hasSucceeded(); + long remainingTime = 0; + if (timeout > 0) { + remainingTime = timeout; + } + long parkTimeout = 10; // NOSONAR + long parkTimeoutNs = TimeUnit.MILLISECONDS.toNanos(parkTimeout); + while (this.active && !tryEmitMessage(message)) { + if (timeout >= 0 && (remainingTime -= parkTimeout) <= 0) { + return false; + } + LockSupport.parkNanos(parkTimeoutNs); + } + return true; + } + + private boolean tryEmitMessage(Message message) { + switch (this.sink.tryEmitNext(message)) { + case FAIL_OVERFLOW: + return false; + case FAIL_TERMINATED: + case FAIL_CANCELLED: + throw new IllegalStateException("Cannot emit messages into the cancelled or terminated sink: " + + this.sink); + default: + return true; + } } @Override @@ -89,6 +119,7 @@ public class FluxMessageChannel extends AbstractMessageChannel @Override public void destroy() { + this.active = false; this.subscribedSignal.emitNext(false); this.upstreamSubscriptions.dispose(); this.processor.onComplete(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java index 61940a13bf..8bee8807bc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java @@ -126,11 +126,22 @@ public final class IntegrationReactiveUtils { private static Flux> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) { return Flux.defer(() -> { - Sinks.Many> sink = Sinks.many().multicast().onBackpressureBuffer(1); - @SuppressWarnings("unchecked") + Sinks.Many> sink = Sinks.many().unicast().onBackpressureError(); MessageHandler messageHandler = (message) -> { - while (!sink.tryEmitNext((Message) message).hasSucceeded()) { - LockSupport.parkNanos(100); // NOSONAR + while (true) { + @SuppressWarnings("unchecked") + Sinks.Emission emission = sink.tryEmitNext((Message) message); + switch (emission) { + case FAIL_OVERFLOW: + LockSupport.parkNanos(1000); // NOSONAR + break; + case FAIL_TERMINATED: + case FAIL_CANCELLED: + throw new IllegalStateException("Cannot emit messages into the cancelled " + + "or terminated sink for message channel: " + inputChannel); + default: + return; + } } }; inputChannel.subscribe(messageHandler); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java index 7173b74767..d6311bd6e0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java @@ -51,7 +51,6 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.endpoint.ReactiveStreamsConsumer; import org.springframework.integration.handler.MethodInvokingMessageHandler; -import org.springframework.integration.test.condition.LogLevels; import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; @@ -175,7 +174,6 @@ public class ReactiveStreamsConsumerTests { reactiveConsumer.stop(); } - @LogLevels(level = "trace", categories = "org.springframework.integration") @Test @SuppressWarnings("unchecked") public void testReactiveStreamsConsumerPollableChannel() throws InterruptedException { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileOutboundChannelAdapterIntegrationTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/FileOutboundChannelAdapterIntegrationTests-context.xml index 94aa0f70e1..54ad6fcd25 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileOutboundChannelAdapterIntegrationTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileOutboundChannelAdapterIntegrationTests-context.xml @@ -1,62 +1,60 @@ - - - - - - - - + + + + + + + + + channel="inputChannelSaveToBaseDir" auto-create-directory="true" + filename-generator-expression="'foo.txt'" directory="target/base-directory"/> + channel="inputChannelSaveToBaseDirDeleteSource" auto-create-directory="true" + delete-source-files="true" filename-generator-expression="'foo.txt'" + directory="target/base-directory"/> + channel="inputChannelSaveToSubDirWrongExpression" auto-create-directory="true" + directory-expression="'target/base-directory/sub-directory/foo.txt'"/> + channel="inputChannelSaveToSubDirEmptyStringExpression" + auto-create-directory="true" + directory-expression="' '"/> + channel="inputChannelSaveToSubDir" auto-create-directory="true" + directory-expression="'target/base-directory/sub-directory'" + filename-generator-expression="'foo.txt'"/> + channel="inputChannelSaveToSubDirWithHeader" auto-create-directory="true" + directory-expression="headers['myFileLocation']" + filename-generator-expression="'foo.txt'"/> + channel="inputChannelSaveToSubDirAutoCreateOff" auto-create-directory="false" + directory-expression="'target/base-directory2/sub-directory2'" + filename-generator-expression="'foo.txt'"/> + channel="inputChannelSaveToSubDirWithFile" auto-create-directory="true" + directory-expression="headers['subDirectory']" + filename-generator-expression="'foo.txt'"/> diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileOutboundChannelAdapterIntegrationTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileOutboundChannelAdapterIntegrationTests.java index de4140fa44..41f754d41e 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileOutboundChannelAdapterIntegrationTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileOutboundChannelAdapterIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-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. @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import java.io.File; import java.io.FileOutputStream; +import java.nio.charset.StandardCharsets; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -32,6 +33,7 @@ import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; +import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.util.FileCopyUtils; @@ -41,10 +43,9 @@ import org.springframework.util.FileCopyUtils; * @author Gary Russell */ @SpringJUnitConfig +@DirtiesContext class FileOutboundChannelAdapterIntegrationTests { - static final String DEFAULT_ENCODING = "UTF-8"; - static final String SAMPLE_CONTENT = "HelloWorld"; @Autowired @@ -78,7 +79,7 @@ class FileOutboundChannelAdapterIntegrationTests { @BeforeEach void setUp(@TempDir File tmpDir) throws Exception { sourceFile = new File(tmpDir, "anyFile.txt"); - FileCopyUtils.copy(SAMPLE_CONTENT.getBytes(DEFAULT_ENCODING), + FileCopyUtils.copy(SAMPLE_CONTENT.getBytes(StandardCharsets.UTF_8), new FileOutputStream(sourceFile, false)); message = MessageBuilder.withPayload(sourceFile).build(); } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithNamespaceIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithNamespaceIntegrationTests.java index 2544dc098d..cec5b9a9d9 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithNamespaceIntegrationTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithNamespaceIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-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. @@ -26,30 +26,27 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.jdbc.storedproc.CreateUser; import org.springframework.integration.jdbc.storedproc.User; -import org.springframework.integration.support.MessageBuilder; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Gunnar Hillert * @author Artem Bilan */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) -@DirtiesContext // close at the end after class +@SpringJUnitConfig +@DirtiesContext public class StoredProcOutboundGatewayWithNamespaceIntegrationTests { @Autowired @@ -67,7 +64,7 @@ public class StoredProcOutboundGatewayWithNamespaceIntegrationTests { @Autowired PollableChannel replyChannel; - @Before + @BeforeEach public void setUp() { this.jdbcTemplate.execute("delete from USERS"); } @@ -77,7 +74,7 @@ public class StoredProcOutboundGatewayWithNamespaceIntegrationTests { createUser.createUser(new User("myUsername", "myPassword", "myEmail")); - List>> received = new ArrayList>>(); + List>> received = new ArrayList<>(); received.add(consumer.poll(2000)); @@ -98,10 +95,9 @@ public class StoredProcOutboundGatewayWithNamespaceIntegrationTests { } - @Test //INT-1029 - public void testStoredProcOutboundGatewayInsideChain() throws Exception { - - Message requestMessage = MessageBuilder.withPayload(new User("myUsername", "myPassword", "myEmail")).build(); + @Test + public void testStoredProcOutboundGatewayInsideChain() { + Message requestMessage = new GenericMessage<>(new User("myUsername", "myPassword", "myEmail")); storedProcOutboundGatewayInsideChain.send(requestMessage); @@ -127,28 +123,30 @@ public class StoredProcOutboundGatewayWithNamespaceIntegrationTests { private final AtomicInteger count = new AtomicInteger(); - public Integer next() throws InterruptedException { + public Integer next() { if (count.get() > 2) { //prevent message overload return null; } return count.incrementAndGet(); } + } static class Consumer { - private final BlockingQueue>> messages = new LinkedBlockingQueue>>(); + private final BlockingQueue>> messages = new LinkedBlockingQueue<>(); @ServiceActivator public void receive(Message> message) { - messages.add(message); + this.messages.add(message); } Message> poll(long timeoutInMillis) throws InterruptedException { - return messages.poll(timeoutInMillis, TimeUnit.MILLISECONDS); + return this.messages.poll(timeoutInMillis, TimeUnit.MILLISECONDS); } + } static class TestService { @@ -156,6 +154,7 @@ public class StoredProcOutboundGatewayWithNamespaceIntegrationTests { public String quote(String s) { return "'" + s + "'"; } + } } 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 bad9ef1c04..1d5960049e 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 @@ -16,19 +16,18 @@ package org.springframework.integration.r2dbc.inbound; - import java.util.Map; import java.util.function.BiFunction; +import java.util.function.Supplier; import org.reactivestreams.Publisher; +import org.springframework.data.r2dbc.convert.EntityRowMapper; import org.springframework.data.r2dbc.core.R2dbcEntityOperations; -import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy; +import org.springframework.data.r2dbc.core.StatementMapper; 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.r2dbc.core.ColumnMapRowMapper; @@ -40,6 +39,7 @@ import io.r2dbc.spi.Row; import io.r2dbc.spi.RowMetadata; 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 @@ -59,9 +59,11 @@ import reactor.core.publisher.Mono; */ public class R2dbcMessageSource extends AbstractMessageSource> { + private final R2dbcEntityOperations r2dbcEntityOperations; + private final DatabaseClient databaseClient; - private final ReactiveDataAccessStrategy dataAccessStrategy; + private final StatementMapper statementMapper; private final Expression queryExpression; @@ -97,20 +99,22 @@ public class R2dbcMessageSource extends AbstractMessageSource> { * 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. + * @param queryExpression The query expression. The root object for evaluation context is a {@link StatementMapper} + * for its {@link StatementMapper#createSelect} fluent API. */ + @SuppressWarnings("deprecation") public R2dbcMessageSource(R2dbcEntityOperations r2dbcEntityOperations, Expression queryExpression) { Assert.notNull(r2dbcEntityOperations, "'r2dbcEntityOperations' must not be null"); Assert.notNull(queryExpression, "'queryExpression' must not be null"); + this.r2dbcEntityOperations = r2dbcEntityOperations; this.databaseClient = r2dbcEntityOperations.getDatabaseClient(); - this.dataAccessStrategy = r2dbcEntityOperations.getDataAccessStrategy(); + this.statementMapper = r2dbcEntityOperations.getDataAccessStrategy().getStatementMapper(); 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. + * {@link DatabaseClient#sql(String)} method. * @param payloadType The t class. */ public void setPayloadType(Class payloadType) { @@ -120,8 +124,7 @@ public class R2dbcMessageSource extends AbstractMessageSource> { /** * Provide a way to set update query that will be passed to the - * {@link org.springframework.data.r2dbc.core.DatabaseClient#execute(String)} - * method. + * {@link DatabaseClient#sql(String)} method. * @param updateSql Update query string. */ public void setUpdateSql(String updateSql) { @@ -163,15 +166,8 @@ public class R2dbcMessageSource extends AbstractMessageSource> { @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"); - } if (!Map.class.isAssignableFrom(this.payloadType)) { - this.rowMapper = this.dataAccessStrategy.getRowMapper(this.payloadType); + this.rowMapper = new EntityRowMapper<>(this.payloadType, this.r2dbcEntityOperations.getConverter()); } this.initialized = true; } @@ -189,7 +185,8 @@ public class R2dbcMessageSource extends AbstractMessageSource> { 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)) + Mono.fromSupplier(() -> + this.queryExpression.getValue(this.evaluationContext, this.statementMapper)) .map(this::prepareFetch); if (this.expectSingleResult) { return queryMono.flatMap(RowsFetchSpec::one) @@ -213,18 +210,20 @@ public class R2dbcMessageSource extends AbstractMessageSource> { } private RowsFetchSpec prepareFetch(Object queryObject) { - String queryString = evaluateQueryObject(queryObject); - return this.databaseClient - .sql(queryString) + Supplier query = evaluateQueryObject(queryObject); + return this.databaseClient.sql(query) .map(this.rowMapper); } - private String evaluateQueryObject(Object queryObject) { + private Supplier evaluateQueryObject(Object queryObject) { if (queryObject instanceof String) { - return (String) queryObject; + return () -> (String) queryObject; + } + else if (queryObject instanceof StatementMapper.SelectSpec) { + return this.statementMapper.getMappedObject((StatementMapper.SelectSpec) queryObject); } throw new IllegalStateException("'queryExpression' must evaluate to String " + - "or org.springframework.data.relational.core.query.Query, but not: " + queryObject); + "or org.springframework.data.r2dbc.core.StatementMapper.SelectSpec, but not: " + queryObject); } } 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 d326d5970c..b613996ec6 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 @@ -76,6 +76,7 @@ public class R2dbcMessageHandler extends AbstractReactiveMessageHandler { * {@link R2dbcEntityOperations} * @param r2dbcEntityOperations The R2dbcEntityOperations implementation. */ + @SuppressWarnings("deprecation") public R2dbcMessageHandler(R2dbcEntityOperations r2dbcEntityOperations) { Assert.notNull(r2dbcEntityOperations, "'r2dbcEntityOperations' must not be null"); this.r2dbcEntityOperations = r2dbcEntityOperations;