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 0f900228e0..a08f51cafa 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 @@ -25,6 +25,7 @@ import org.springframework.messaging.MessageHandler; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; +import reactor.core.publisher.EmitterProcessor; import reactor.core.publisher.Flux; import reactor.core.publisher.FluxSink; import reactor.core.scheduler.Schedulers; @@ -60,37 +61,20 @@ public final class MessageChannelReactiveUtils { } private static Publisher> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) { - return new SubscribableChannelPublisherAdapter<>(inputChannel); + return Flux.defer(() -> { + EmitterProcessor> publisher = EmitterProcessor.create(1); + @SuppressWarnings("unchecked") + MessageHandler messageHandler = (message) -> publisher.onNext((Message) message); + inputChannel.subscribe(messageHandler); + return publisher + .doOnCancel(() -> inputChannel.unsubscribe(messageHandler)); + }); } private static Publisher> adaptPollableChannelToPublisher(PollableChannel inputChannel) { return new PollableChannelPublisherAdapter<>(inputChannel); } - - private static final class SubscribableChannelPublisherAdapter implements Publisher> { - - private final SubscribableChannel channel; - - SubscribableChannelPublisherAdapter(SubscribableChannel channel) { - this.channel = channel; - } - - @Override - @SuppressWarnings("unchecked") - public void subscribe(Subscriber> subscriber) { - Flux. - >create(emitter -> { - MessageHandler messageHandler = emitter::next; - this.channel.subscribe(messageHandler); - emitter.onCancel(() -> this.channel.unsubscribe(messageHandler)); - }, - FluxSink.OverflowStrategy.IGNORE) - .subscribe((Subscriber>) subscriber); - } - - } - private static final class PollableChannelPublisherAdapter implements Publisher> { private final PollableChannel channel; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java index 1df8807c43..f52a017a22 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java @@ -155,8 +155,8 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra @Override public void hookOnSubscribe(Subscription s) { - this.delegate.onSubscribe(s); ReactiveStreamsConsumer.this.subscription = s; + this.delegate.onSubscribe(s); } @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTest.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java similarity index 77% rename from spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTest.java rename to spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java index 8fd69c5929..7b0b90f04d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTest.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2019 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. @@ -20,9 +20,8 @@ import static org.assertj.core.api.Assertions.assertThat; import java.time.Duration; -import org.junit.Test; +import org.junit.jupiter.api.Test; -import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.GenericMessage; import reactor.core.Disposable; @@ -31,14 +30,19 @@ import reactor.core.scheduler.Schedulers; import reactor.test.StepVerifier; import reactor.util.concurrent.Queues; -public class MessageChannelReactiveUtilsTest { +/** + * @author Sergei Egorov + * @author Artem Bilan + * + * @since 5.1.9 + */ +class MessageChannelReactiveUtilsTests { @Test - public void testBackpressureWithSubscribableChannel() { + void testBackpressureWithSubscribableChannel() { Disposable.Composite compositeDisposable = Disposables.composite(); try { DirectChannel channel = new DirectChannel(); - assertThat(channel).isInstanceOf(SubscribableChannel.class); int initialRequest = 10; StepVerifier.create(MessageChannelReactiveUtils.toPublisher(channel), initialRequest) .expectSubscription() @@ -64,27 +68,25 @@ public class MessageChannelReactiveUtilsTest { } @Test - public void testOverproducingWithSubscribableChannel() { + void testOverproducingWithSubscribableChannel() { DirectChannel channel = new DirectChannel(); channel.setCountsEnabled(true); - assertThat(channel).isInstanceOf(SubscribableChannel.class); Disposable.Composite compositeDisposable = Disposables.composite(); try { int initialRequest = 10; StepVerifier.create(MessageChannelReactiveUtils.toPublisher(channel), initialRequest) .expectSubscription() - .then(() -> { - compositeDisposable.add( - Schedulers.boundedElastic().schedule(() -> { - while (true) { - if (channel.getSubscriberCount() > 0) { - channel.send(new GenericMessage<>("foo")); + .then(() -> + compositeDisposable.add( + Schedulers.boundedElastic().schedule(() -> { + while (true) { + if (channel.getSubscriberCount() > 0) { + channel.send(new GenericMessage<>("foo")); + } } - } - }) - ); - }) + }) + )) .expectNextCount(initialRequest) .thenAwait(Duration.ofMillis(100)) .thenCancel() @@ -98,4 +100,5 @@ public class MessageChannelReactiveUtilsTest { .as("produced") .isLessThanOrEqualTo(Queues.SMALL_BUFFER_SIZE); } + } 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 a30bf791ba..fdb612f852 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 @@ -17,12 +17,10 @@ package org.springframework.integration.channel.reactive; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; 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; import static org.mockito.Mockito.verify; import java.util.LinkedList; @@ -33,7 +31,6 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import org.junit.Test; -import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; @@ -82,14 +79,10 @@ public class ReactiveStreamsConsumerTests { reactiveConsumer.stop(); - try { - testChannel.send(testMessage); - } - catch (Exception e) { - assertThat(e).isInstanceOf(MessageDeliveryException.class); - assertThat(e.getCause()).isInstanceOf(IllegalStateException.class); - assertThat(e.getMessage()).contains("doesn't have subscribers to accept messages"); - } + assertThatExceptionOfType(MessageDeliveryException.class) + .isThrownBy(() -> testChannel.send(testMessage)) + .withCauseInstanceOf(IllegalStateException.class) + .withMessageContaining("doesn't have subscribers to accept messages"); reactiveConsumer.start(); @@ -102,54 +95,53 @@ public class ReactiveStreamsConsumerTests { @Test - @SuppressWarnings("unchecked") public void testReactiveStreamsConsumerDirectChannel() throws InterruptedException { DirectChannel testChannel = new DirectChannel(); - Subscriber> testSubscriber = (Subscriber>) Mockito.mock(Subscriber.class); - BlockingQueue> messages = new LinkedBlockingQueue<>(); - willAnswer(i -> { - messages.put(i.getArgument(0)); - return null; - }) - .given(testSubscriber) - .onNext(any(Message.class)); + Subscriber> testSubscriber = Mockito.spy(new Subscriber>() { + + @Override + public void onSubscribe(Subscription subscription) { + subscription.request(1); + } + + @Override + public void onNext(Message message) { + messages.offer(message); + } + + @Override + public void onError(Throwable t) { + + } + + @Override + public void onComplete() { + + } + + }); ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber); reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); reactiveConsumer.afterPropertiesSet(); reactiveConsumer.start(); - Message testMessage = new GenericMessage<>("test"); + final Message testMessage = new GenericMessage<>("test"); testChannel.send(testMessage); - ArgumentCaptor subscriptionArgumentCaptor = ArgumentCaptor.forClass(Subscription.class); - verify(testSubscriber).onSubscribe(subscriptionArgumentCaptor.capture()); - Subscription subscription = subscriptionArgumentCaptor.getValue(); - - subscription.request(1); - Message message = messages.poll(10, TimeUnit.SECONDS); assertThat(message).isSameAs(testMessage); reactiveConsumer.stop(); - try { - testChannel.send(testMessage); - fail("MessageDeliveryException"); - } - catch (Exception e) { - assertThat(e).isInstanceOf(MessageDeliveryException.class); - } + assertThatExceptionOfType(MessageDeliveryException.class) + .isThrownBy(() -> testChannel.send(testMessage)); reactiveConsumer.start(); - subscription.request(1); - - testMessage = new GenericMessage<>("test2"); - testChannel.send(testMessage); message = messages.poll(10, TimeUnit.SECONDS); @@ -159,6 +151,8 @@ public class ReactiveStreamsConsumerTests { verify(testSubscriber, never()).onComplete(); assertThat(messages.isEmpty()).isTrue(); + + reactiveConsumer.stop(); } @Test @@ -166,17 +160,31 @@ public class ReactiveStreamsConsumerTests { public void testReactiveStreamsConsumerPollableChannel() throws InterruptedException { QueueChannel testChannel = new QueueChannel(); - Subscriber> testSubscriber = (Subscriber>) Mockito.mock(Subscriber.class); - BlockingQueue> messages = new LinkedBlockingQueue<>(); - willAnswer(i -> { - messages.put(i.getArgument(0)); - return null; - }) - .given(testSubscriber) - .onNext(any(Message.class)); + Subscriber> testSubscriber = Mockito.spy(new Subscriber>() { + @Override + public void onSubscribe(Subscription subscription) { + subscription.request(2); + } + + @Override + public void onNext(Message message) { + messages.offer(message); + } + + @Override + public void onError(Throwable t) { + + } + + @Override + public void onComplete() { + + } + + }); ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber); reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); reactiveConsumer.afterPropertiesSet(); @@ -185,12 +193,6 @@ public class ReactiveStreamsConsumerTests { Message testMessage = new GenericMessage<>("test"); testChannel.send(testMessage); - ArgumentCaptor subscriptionArgumentCaptor = ArgumentCaptor.forClass(Subscription.class); - verify(testSubscriber).onSubscribe(subscriptionArgumentCaptor.capture()); - Subscription subscription = subscriptionArgumentCaptor.getValue(); - - subscription.request(1); - Message message = messages.poll(10, TimeUnit.SECONDS); assertThat(message).isSameAs(testMessage); @@ -201,11 +203,6 @@ public class ReactiveStreamsConsumerTests { reactiveConsumer.start(); - verify(testSubscriber, times(2)).onSubscribe(subscriptionArgumentCaptor.capture()); - subscription = subscriptionArgumentCaptor.getValue(); - - subscription.request(2); - Message testMessage2 = new GenericMessage<>("test2"); testChannel.send(testMessage2); @@ -247,14 +244,10 @@ public class ReactiveStreamsConsumerTests { endpointFactoryBean.stop(); - try { - testChannel.send(testMessage); - } - catch (Exception e) { - assertThat(e).isInstanceOf(MessageDeliveryException.class); - assertThat(e.getCause()).isInstanceOf(IllegalStateException.class); - assertThat(e.getMessage()).contains("doesn't have subscribers to accept messages"); - } + assertThatExceptionOfType(MessageDeliveryException.class) + .isThrownBy(() -> testChannel.send(testMessage)) + .withCauseInstanceOf(IllegalStateException.class) + .withMessageContaining("doesn't have subscribers to accept messages"); endpointFactoryBean.start();