diff --git a/build.gradle b/build.gradle index f6a2ab7783..f2f8e16bd7 100644 --- a/build.gradle +++ b/build.gradle @@ -4,7 +4,7 @@ buildscript { } dependencies { classpath 'io.spring.gradle:dependency-management-plugin:1.0.2.RELEASE' - classpath 'io.spring.gradle:spring-io-plugin:0.0.6.RELEASE' + classpath 'io.spring.gradle:spring-io-plugin:0.0.7.RELEASE' classpath 'io.spring.gradle:docbook-reference-plugin:0.3.1' classpath 'org.asciidoctor:asciidoctor-gradle-plugin:1.5.0' } 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 99dd819a25..6b42b60a19 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 @@ -18,22 +18,21 @@ package org.springframework.integration.channel; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.springframework.messaging.Message; -import org.springframework.util.Assert; -import reactor.core.publisher.DirectProcessor; +import reactor.core.publisher.ConnectableFlux; import reactor.core.publisher.Flux; -import reactor.core.publisher.FluxProcessor; import reactor.core.publisher.FluxSink; /** * The {@link AbstractMessageChannel} implementation for the - * Reactive Streams {@link Publisher} based on the Project Reactor {@link FluxProcessor}. + * Reactive Streams {@link Publisher} based on the Project Reactor {@link Flux}. * * @author Artem Bilan * @author Gary Russell @@ -41,26 +40,21 @@ import reactor.core.publisher.FluxSink; * @since 5.0 */ public class FluxMessageChannel extends AbstractMessageChannel - implements Publisher>, FluxSubscribableChannel { + implements Publisher>, ReactiveStreamsSubscribableChannel { private final List>> subscribers = new ArrayList<>(); - private final List>> publishers = new CopyOnWriteArrayList<>(); + private final Map>, ConnectableFlux>> publishers = new ConcurrentHashMap<>(); - private final FluxProcessor, Message> processor; + private final Flux> flux; - private final FluxSink> sink; - - private volatile boolean upstreamSubscribed; + private FluxSink> sink; public FluxMessageChannel() { - this(DirectProcessor.create()); - } - - public FluxMessageChannel(FluxProcessor, Message> processor) { - Assert.notNull(processor, "'processor' must not be null"); - this.processor = processor; - this.sink = processor.sink(); + this.flux = + Flux.>create(emitter -> this.sink = emitter, FluxSink.OverflowStrategy.IGNORE) + .publish() + .autoConnect(); } @Override @@ -73,32 +67,26 @@ public class FluxMessageChannel extends AbstractMessageChannel public void subscribe(Subscriber> subscriber) { this.subscribers.add(subscriber); - this.processor.doOnCancel(() -> FluxMessageChannel.this.subscribers.remove(subscriber)) + this.flux.doOnCancel(() -> this.subscribers.remove(subscriber)) + .retry() .subscribe(subscriber); - if (!this.upstreamSubscribed) { - this.publishers.forEach(this::doSubscribeTo); - } + this.publishers.values().forEach(ConnectableFlux::connect); } @Override - public void subscribeTo(Flux> publisher) { - this.publishers.add(publisher); + public void subscribeTo(Publisher> publisher) { + ConnectableFlux> connectableFlux = + Flux.from(publisher) + .doOnComplete(() -> this.publishers.remove(publisher)) + .doOnNext(this::send) + .publish(); + + this.publishers.put(publisher, connectableFlux); + if (!this.subscribers.isEmpty()) { - doSubscribeTo(publisher); + connectableFlux.connect(); } } - private void doSubscribeTo(Publisher> publisher) { - Flux.from(publisher) - .doOnSubscribe(s -> FluxMessageChannel.this.upstreamSubscribed = true) - .doOnComplete(() -> { - FluxMessageChannel.this.publishers.remove(publisher); - if (FluxMessageChannel.this.publishers.isEmpty()) { - FluxMessageChannel.this.upstreamSubscribed = false; - } - }) - .subscribe(this.processor); - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java index d84ce87591..6d1a77db42 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java @@ -16,9 +16,6 @@ package org.springframework.integration.channel; -import java.time.Duration; -import java.util.Iterator; - import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; @@ -30,7 +27,7 @@ import org.springframework.messaging.SubscribableChannel; import reactor.core.publisher.Flux; import reactor.core.publisher.FluxSink; -import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; /** * Utilities for adaptation {@link MessageChannel}s to the {@link Publisher}s. @@ -105,31 +102,17 @@ public final class MessageChannelReactiveUtils { @Override @SuppressWarnings("unchecked") public void subscribe(Subscriber> subscriber) { - Iterator> messageIterator = new Iterator>() { - - private Message next = null; - - @Override - public Message next() { - Message message = this.next; - this.next = null; - return message; - } - - @Override - public boolean hasNext() { - if (this.next == null) { - this.next = PollableChannelPublisherAdapter.this.channel.receive(0); - } - return this.next != null; - } - - }; - - Mono.>delay(Duration.ofMillis(100)) - .repeat() - .concatMap(value -> Flux.fromIterable(() -> messageIterator)) - .subscribe((Subscriber>) subscriber); + Flux + .>create(sink -> + sink.onRequest(n -> { + Message m; + while (n-- > 0 && (m = this.channel.receive()) != null) { + sink.next((Message) m); + } + }), + FluxSink.OverflowStrategy.IGNORE) + .subscribeOn(Schedulers.elastic()) + .subscribe(subscriber); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxSubscribableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveStreamsSubscribableChannel.java similarity index 84% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/FluxSubscribableChannel.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveStreamsSubscribableChannel.java index 756666ea91..01510c5b20 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxSubscribableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveStreamsSubscribableChannel.java @@ -16,9 +16,9 @@ package org.springframework.integration.channel; -import org.springframework.messaging.Message; +import org.reactivestreams.Publisher; -import reactor.core.publisher.Flux; +import org.springframework.messaging.Message; /** * @author Artem Bilan @@ -26,8 +26,8 @@ import reactor.core.publisher.Flux; * * @since 5.0 */ -public interface FluxSubscribableChannel { +public interface ReactiveStreamsSubscribableChannel { - void subscribeTo(Flux> publisher); + void subscribeTo(Publisher> publisher); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 6807985b0d..2a31913ad6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -39,7 +39,7 @@ import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; -import org.springframework.integration.endpoint.ReactiveConsumer; +import org.springframework.integration.endpoint.ReactiveStreamsConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.integration.scheduling.PollerMetadata; @@ -286,7 +286,7 @@ public class ConsumerEndpointFactoryBean this.endpoint = pollingConsumer; } else { - this.endpoint = new ReactiveConsumer(channel, this.handler); + this.endpoint = new ReactiveStreamsConsumer(channel, this.handler); } this.endpoint.setBeanName(this.beanName); this.endpoint.setBeanFactory(this.beanFactory); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index b4bb80ac21..3767423f10 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -55,7 +55,7 @@ import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.AbstractPollingEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; -import org.springframework.integration.endpoint.ReactiveConsumer; +import org.springframework.integration.endpoint.ReactiveStreamsConsumer; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; @@ -321,7 +321,7 @@ public abstract class AbstractMethodAnnotationPostProcessor, Message> processor) { - return MessageChannels.flux(processor); - } - - public FluxMessageChannelSpec flux(String id, FluxProcessor, Message> processor) { - return MessageChannels.flux(id, processor); - } - Channels() { super(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index de2eefc411..7921273449 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -35,8 +35,6 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; -import reactor.core.publisher.Flux; - /** * The central factory for fluent {@link IntegrationFlowBuilder} API. * @@ -307,10 +305,10 @@ public final class IntegrationFlows { * @param publisher the {@link Publisher} to subscribe to. * @return new {@link IntegrationFlowBuilder}. */ - public static IntegrationFlowBuilder from(Flux> publisher) { + public static IntegrationFlowBuilder from(Publisher> publisher) { FluxMessageChannel reactiveChannel = new FluxMessageChannel(); reactiveChannel.subscribeTo(publisher); - return from(reactiveChannel); + return from((MessageChannel) reactiveChannel); } private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway, diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/FluxMessageChannelSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/FluxMessageChannelSpec.java index 5f8c261525..b03f3841aa 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/FluxMessageChannelSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/FluxMessageChannelSpec.java @@ -17,9 +17,6 @@ package org.springframework.integration.dsl.channel; import org.springframework.integration.channel.FluxMessageChannel; -import org.springframework.messaging.Message; - -import reactor.core.publisher.FluxProcessor; /** * @author Artem Bilan @@ -33,8 +30,4 @@ public class FluxMessageChannelSpec extends MessageChannelSpec, Message> processor) { - this.channel = new FluxMessageChannel(processor); - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java index 5465d55837..0c4900b3f0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java @@ -23,8 +23,6 @@ import org.springframework.integration.store.ChannelMessageStore; import org.springframework.integration.store.PriorityCapableChannelMessageStore; import org.springframework.messaging.Message; -import reactor.core.publisher.FluxProcessor; - /** * @author Artem Bilan * @author Gary Russell @@ -135,15 +133,6 @@ public final class MessageChannels { .id(id); } - public static FluxMessageChannelSpec flux(String id, FluxProcessor, Message> processor) { - return flux(processor) - .id(id); - } - - public static FluxMessageChannelSpec flux(FluxProcessor, Message> processor) { - return new FluxMessageChannelSpec(processor); - } - private MessageChannels() { super(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java similarity index 76% rename from spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java rename to spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java index 1f00ebe4df..1ae2f01007 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java @@ -25,6 +25,8 @@ import org.reactivestreams.Subscription; import org.springframework.context.Lifecycle; import org.springframework.integration.channel.MessageChannelReactiveUtils; import org.springframework.integration.channel.MessagePublishingErrorHandler; +import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.router.MessageRouter; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -33,16 +35,18 @@ import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; import reactor.core.Disposable; -import reactor.core.Exceptions; import reactor.core.publisher.BaseSubscriber; -import reactor.core.publisher.Operators; /** * @author Artem Bilan * @since 5.0 */ -public class ReactiveConsumer extends AbstractEndpoint { +public class ReactiveStreamsConsumer extends AbstractEndpoint implements IntegrationConsumer { + + private final MessageChannel inputChannel; + + private final MessageHandler messageHandler; private final Publisher> publisher; @@ -55,28 +59,56 @@ public class ReactiveConsumer extends AbstractEndpoint { private volatile Subscription subscription; @SuppressWarnings("unchecked") - public ReactiveConsumer(MessageChannel inputChannel, MessageHandler messageHandler) { + public ReactiveStreamsConsumer(MessageChannel inputChannel, MessageHandler messageHandler) { this(inputChannel, messageHandler instanceof Subscriber ? (Subscriber>) messageHandler : new MessageHandlerSubscriber(messageHandler)); } - public ReactiveConsumer(MessageChannel inputChannel, final Subscriber> subscriber) { + public ReactiveStreamsConsumer(MessageChannel inputChannel, final Subscriber> subscriber) { + this.inputChannel = inputChannel; Assert.notNull(inputChannel, "'inputChannel' must not be null"); Assert.notNull(subscriber, "'subscriber' must not be null"); this.publisher = MessageChannelReactiveUtils.toPublisher(inputChannel); - this.subscriber = subscriber; - this.lifecycleDelegate = subscriber instanceof Lifecycle ? (Lifecycle) subscriber : null; + if (subscriber instanceof MessageHandlerSubscriber) { + this.messageHandler = ((MessageHandlerSubscriber) subscriber).messageHandler; + } + else { + this.messageHandler = this.subscriber::onNext; + } } public void setErrorHandler(ErrorHandler errorHandler) { this.errorHandler = errorHandler; } + @Override + public MessageChannel getInputChannel() { + return this.inputChannel; + } + + @Override + public MessageChannel getOutputChannel() { + if (this.messageHandler instanceof MessageProducer) { + return ((MessageProducer) this.messageHandler).getOutputChannel(); + } + else if (this.messageHandler instanceof MessageRouter) { + return ((MessageRouter) this.messageHandler).getDefaultOutputChannel(); + } + else { + return null; + } + } + + @Override + public MessageHandler getHandler() { + return this.messageHandler; + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -93,11 +125,11 @@ public class ReactiveConsumer extends AbstractEndpoint { } this.publisher.subscribe(new BaseSubscriber>() { - private final Subscriber> delegate = ReactiveConsumer.this.subscriber; + private final Subscriber> delegate = ReactiveStreamsConsumer.this.subscriber; public void hookOnSubscribe(Subscription s) { this.delegate.onSubscribe(s); - ReactiveConsumer.this.subscription = s; + ReactiveStreamsConsumer.this.subscription = s; } public void hookOnNext(Message message) { @@ -105,7 +137,7 @@ public class ReactiveConsumer extends AbstractEndpoint { this.delegate.onNext(message); } catch (Exception e) { - ReactiveConsumer.this.errorHandler.handleError(e); + ReactiveStreamsConsumer.this.errorHandler.handleError(e); hookOnError(e); } } @@ -160,18 +192,11 @@ public class ReactiveConsumer extends AbstractEndpoint { @Override public void onError(Throwable t) { - if (t == null) { - throw Exceptions.argumentIsNullException(); - } - onComplete(); - Operators.onErrorDropped(t); } @Override public void onComplete() { - if (this.subscription != null) { - this.subscription = null; - } + dispose(); } @Override @@ -185,7 +210,7 @@ public class ReactiveConsumer extends AbstractEndpoint { @Override public boolean isDisposed() { - return false; + return this.subscription == null; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index 929cfbdb45..ada1c4ef78 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -20,7 +20,6 @@ import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; import org.springframework.core.Ordered; -import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.context.Orderable; import org.springframework.integration.history.MessageHistory; @@ -37,9 +36,6 @@ import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; -import org.springframework.util.ErrorHandler; - -import reactor.core.publisher.Operators; /** * Base class for MessageHandler implementations that provides basic validation @@ -72,8 +68,6 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im private volatile boolean loggingEnabled = true; - private ErrorHandler reactiveErrorHandler; - @Override public boolean isLoggingEnabled() { return this.loggingEnabled; @@ -104,16 +98,6 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im this.shouldTrack = shouldTrack; } - /** - * Set the error handler to use when an exception occurs when this handler - * is invoked as a reactive {@link Subscriber}. - * @param reactiveErrorHandler the error handler. - * @since 5.0 - */ - public void setReactiveErrorHandler(ErrorHandler reactiveErrorHandler) { - this.reactiveErrorHandler = reactiveErrorHandler; - } - @Override public void configureMetrics(AbstractMessageHandlerMetrics metrics) { Assert.notNull(metrics, "'metrics' must not be null"); @@ -125,14 +109,6 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im if (this.statsEnabled) { this.handlerMetrics.setFullStatsEnabled(true); } - if (this.reactiveErrorHandler == null) { - if (getBeanFactory() != null) { - this.reactiveErrorHandler = new MessagePublishingErrorHandler(getChannelResolver()); - } - else { - this.reactiveErrorHandler = new MessagePublishingErrorHandler(); - } - } } @Override @@ -176,17 +152,12 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im @Override public void onNext(Message message) { - try { - handleMessage(message); - } - catch (MessagingException e) { - this.reactiveErrorHandler.handleError(e); - } + handleMessage(message); } @Override public void onError(Throwable throwable) { - Operators.onErrorDropped(throwable); + } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index 401e63ae9f..3a3ef26100 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -24,7 +24,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.reactivestreams.Publisher; import org.springframework.integration.IntegrationMessageHeaderAccessor; -import org.springframework.integration.channel.FluxSubscribableChannel; +import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.routingslip.RoutingSlipRouteStrategy; @@ -147,6 +147,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan return false; } + @SuppressWarnings("unchecked") protected void produceOutput(Object reply, final Message requestMessage) { final MessageHeaders requestHeaders = requestMessage.getHeaders(); @@ -190,7 +191,8 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } if (this.async && (reply instanceof ListenableFuture || reply instanceof Publisher)) { - if (reply instanceof ListenableFuture || !(getOutputChannel() instanceof FluxSubscribableChannel)) { + if (reply instanceof ListenableFuture || + !(getOutputChannel() instanceof ReactiveStreamsSubscribableChannel)) { ListenableFuture future; if (reply instanceof ListenableFuture) { future = (ListenableFuture) reply; @@ -235,9 +237,10 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan }); } else { - ((FluxSubscribableChannel) getOutputChannel()) - .subscribeTo(Flux.from((Publisher) reply) - .map(result -> createOutputMessage(result, requestHeaders))); + ((ReactiveStreamsSubscribableChannel) getOutputChannel()) + .subscribeTo( + Flux.from((Publisher) reply) + .map(result -> createOutputMessage(result, requestHeaders))); } } else { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java similarity index 90% rename from spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java index 8a73265c49..c616338ee9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java @@ -22,8 +22,8 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; -import static org.mockito.BDDMockito.willAnswer; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.willAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; @@ -46,28 +46,26 @@ import org.reactivestreams.Subscription; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.ConsumerEndpointFactoryBean; -import org.springframework.integration.endpoint.ReactiveConsumer; +import org.springframework.integration.endpoint.ReactiveStreamsConsumer; import org.springframework.integration.handler.MethodInvokingMessageHandler; import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.support.GenericMessage; -import reactor.core.publisher.EmitterProcessor; - /** * @author Artem Bilan * * @since 5.0 */ -public class ReactiveConsumerTests { +public class ReactiveStreamsConsumerTests { @Test - public void testReactiveConsumerReactiveChannel() throws InterruptedException { - FluxMessageChannel testChannel = new FluxMessageChannel(EmitterProcessor.create(false)); + public void testReactiveStreamsConsumerFluxMessageChannel() throws InterruptedException { + FluxMessageChannel testChannel = new FluxMessageChannel(); List> result = new LinkedList<>(); CountDownLatch stopLatch = new CountDownLatch(2); @@ -79,7 +77,7 @@ public class ReactiveConsumerTests { MessageHandler testSubscriber = new MethodInvokingMessageHandler(messageHandler, (String) null); - ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber); + ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber); reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); reactiveConsumer.afterPropertiesSet(); reactiveConsumer.start(); @@ -103,7 +101,7 @@ public class ReactiveConsumerTests { @Test @SuppressWarnings("unchecked") - public void testReactiveConsumerDirectChannel() throws InterruptedException { + public void testReactiveStreamsConsumerDirectChannel() throws InterruptedException { DirectChannel testChannel = new DirectChannel(); Subscriber> testSubscriber = (Subscriber>) Mockito.mock(Subscriber.class); @@ -117,7 +115,7 @@ public class ReactiveConsumerTests { .given(testSubscriber) .onNext(any(Message.class)); - ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber); + ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber); reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); reactiveConsumer.afterPropertiesSet(); reactiveConsumer.start(); @@ -163,7 +161,7 @@ public class ReactiveConsumerTests { @Test @SuppressWarnings("unchecked") - public void testReactiveConsumerPollableChannel() throws InterruptedException { + public void testReactiveStreamsConsumerPollableChannel() throws InterruptedException { QueueChannel testChannel = new QueueChannel(); Subscriber> testSubscriber = (Subscriber>) Mockito.mock(Subscriber.class); @@ -177,7 +175,7 @@ public class ReactiveConsumerTests { .given(testSubscriber) .onNext(any(Message.class)); - ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber); + ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber); reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); reactiveConsumer.afterPropertiesSet(); reactiveConsumer.start(); @@ -223,7 +221,7 @@ public class ReactiveConsumerTests { } @Test - public void testReactiveConsumerViaConsumerEndpointFactoryBean() throws Exception { + public void testReactiveStreamsConsumerViaConsumerEndpointFactoryBean() throws Exception { FluxMessageChannel testChannel = new FluxMessageChannel(); List> result = new LinkedList<>(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java index 2fe4896e92..fb904b8bfd 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java @@ -32,7 +32,6 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import org.apache.log4j.Level; import org.junit.Test; import org.junit.runner.RunWith; import org.reactivestreams.Publisher; @@ -49,7 +48,6 @@ import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.channel.MessageChannels; import org.springframework.integration.dsl.context.IntegrationFlowContext; -import org.springframework.integration.test.rule.Log4jLevelAdjuster; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; @@ -69,9 +67,6 @@ import reactor.core.publisher.Flux; @DirtiesContext public class ReactiveStreamsTests { -// @Rule - public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.DEBUG, "org.springframework.integration"); - @Autowired @Qualifier("reactiveFlow") private Publisher> publisher; diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java index c0cc5cf8ee..029a599e21 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java @@ -18,26 +18,29 @@ package org.springframework.integration.http.outbound; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; -import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeader; -import java.time.Duration; +import java.util.List; import org.junit.Test; +import org.reactivestreams.Subscriber; import org.springframework.http.HttpStatus; import org.springframework.http.client.reactive.ClientHttpConnector; -import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.http.HttpHeaders; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.support.ErrorMessage; import org.springframework.test.web.reactive.server.HttpHandlerConnector; import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; /** * @author Shiliang Li @@ -66,12 +69,15 @@ public class ReactiveHttpRequestExecutingMessageHandlerTests { FluxMessageChannel ackChannel = new FluxMessageChannel(); reactiveHandler.setOutputChannel(ackChannel); reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build()); + reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build()); - Message ack = Mono.from(ackChannel).block(Duration.ofSeconds(10)); - - assertNotNull(ack); - assertNotNull(ack.getHeaders()); - assertEquals(ack.getHeaders().get(HttpHeaders.STATUS_CODE), HttpStatus.OK); + StepVerifier.create(ackChannel, 2) + .assertNext(m -> assertThat(m, hasHeader(HttpHeaders.STATUS_CODE, HttpStatus.OK))) + .assertNext(m -> assertThat(m, hasHeader(HttpHeaders.STATUS_CODE, HttpStatus.OK))) + .then(() -> + ((Subscriber) TestUtils.getPropertyValue(ackChannel, "subscribers", List.class).get(0)) + .onComplete()) + .verifyComplete(); } @Test