From ba576bbc0d37a91a041271afb0608de8ac1c1a0f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 25 Aug 2016 12:13:08 -0400 Subject: [PATCH] ReactiveConsumer Improvements * Add `ConsumerSubscriber` to adapt `Consumer` to the `Subscriber` * Add `SubscribableChannelPublisherAdapter` to adapt `SubscribableChannel` to `Publisher` via `Flux.create()` on subscription * Add `PollableChannelPublisherAdapter` to adapt `PollableChannel` to `Publisher` via `Mono.delayMillis()` and `flatMap()` for `channel.receive()` Use `concatMap()` for `PollableChannel` Iterate `PollableChannel` until there is a data Fix Lambda signature error via explicit class declaration: ``` java.lang.ClassFormatError: Duplicate field name&signature in class file org/springframework/integration/endpoint/ReactiveConsumer$PollableChannelPublisherAdapter$1 at java.lang.ClassLoader.defineClass1(Native Method) at java.lang.ClassLoader.defineClass(ClassLoader.java:760) at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142) at java.net.URLClassLoader.defineClass(URLClassLoader.java:467) at java.net.URLClassLoader.access$100(URLClassLoader.java:73) at java.net.URLClassLoader$1.run(URLClassLoader.java:368) at java.net.URLClassLoader$1.run(URLClassLoader.java:362) at java.security.AccessController.doPrivileged(Native Method) at java.net.URLClassLoader.findClass(URLClassLoader.java:361) at java.lang.ClassLoader.loadClass(ClassLoader.java:424) at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) at org.springframework.integration.endpoint.ReactiveConsumer$PollableChannelPublisherAdapter.lambda$new$2(ReactiveConsumer.java:163) ``` Looks like Java bug: Move `Iterator>` instance to the `subscribe()` to avoid race conditions when we have several subscribers. In other words make `Iterator>` subscriber-specific, but at the same time avoid re-instantiation for each `Flux.concatMap()` call caused by the previous `repeat()` Upgrade to Reactor 3.0.3 and others * Make fixes according Reactor 3.0.3 changes * Remove redundant `TestSubscriber` in favor of `StepVerifier` and `mock(Subscriber)` * Rework `PublisherIntegrationFlow` Reactive Streams implementation to the out-of-the-box `ReactiveConsumer` and `ReactiveChannel` --- build.gradle | 4 +- .../dsl/IntegrationFlowDefinition.java | 49 +- .../dsl/PublisherIntegrationFlow.java | 209 +-- .../endpoint/ReactiveConsumer.java | 192 ++- .../handler/AbstractMessageHandler.java | 8 +- .../reactive/ReactiveChannelTests.java | 19 +- .../reactive/ReactiveConsumerTests.java | 111 +- .../reactivestreams/ReactiveStreamsTests.java | 14 +- .../test/reactive/TestSubscriber.java | 1164 ----------------- 9 files changed, 280 insertions(+), 1490 deletions(-) delete mode 100644 spring-integration-test/src/main/java/org/springframework/integration/test/reactive/TestSubscriber.java diff --git a/build.gradle b/build.gradle index 53fa3e04cd..38f992bf95 100644 --- a/build.gradle +++ b/build.gradle @@ -116,7 +116,7 @@ subprojects { subproject -> nettyVersion = '4.1.4.Final' pahoMqttClientVersion = '1.0.2' postgresVersion = '9.1-901-1.jdbc4' - reactorVersion = '3.0.0.BUILD-SNAPSHOT' + reactorVersion = '3.0.3.RELEASE' romeToolsVersion = '1.7.0' servletApiVersion = '3.1.0' slf4jVersion = "1.7.21" @@ -261,7 +261,6 @@ project('spring-integration-test') { compile "org.springframework:spring-context:$springVersion" compile "org.springframework:spring-messaging:$springVersion" compile "org.springframework:spring-test:$springVersion" - compile ("io.projectreactor:reactor-core:$reactorVersion", optional) compile ("log4j:log4j:$log4jVersion", optional) } } @@ -303,6 +302,7 @@ project('spring-integration-core') { compile("com.esotericsoftware:kryo-shaded:$kryoShadedVersion", optional) testCompile ("org.aspectj:aspectjweaver:$aspectjVersion") + compile "io.projectreactor.addons:reactor-test:$reactorVersion" } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index f57450e21f..6e9e331656 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -22,7 +22,6 @@ import java.util.LinkedHashSet; import java.util.Map; import java.util.Set; import java.util.concurrent.Executor; -import java.util.concurrent.Executors; import java.util.function.Consumer; import java.util.function.Function; @@ -39,7 +38,7 @@ import org.springframework.integration.aggregator.ResequencingMessageHandler; import org.springframework.integration.channel.ChannelInterceptorAware; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.FixedSubscriberChannel; -import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.integration.channel.ReactiveChannel; import org.springframework.integration.channel.interceptor.WireTap; import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; @@ -49,6 +48,7 @@ import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.channel.WireTapSpec; import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype; import org.springframework.integration.dsl.support.MessageChannelReference; +import org.springframework.integration.endpoint.ReactiveConsumer; import org.springframework.integration.expression.ControlBusMethodFilter; import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.filter.ExpressionEvaluatingSelector; @@ -89,7 +89,6 @@ import org.springframework.integration.transformer.Transformer; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.PollableChannel; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; @@ -2762,32 +2761,36 @@ public abstract class IntegrationFlowDefinition the {@code payload} type * @return the Reactive Streams {@link Publisher} */ + @SuppressWarnings("unchecked") public Publisher> toReactivePublisher() { - return toReactivePublisher(Executors.newSingleThreadExecutor()); - } - - /** - * Represent an Integration Flow as a Reactive Streams {@link Publisher} bean. - * @param executor the managed {@link Executor} to be used for the background task to - * poll messages from the {@link PollableChannel}. - * Defaults to {@link Executors#newSingleThreadExecutor()}. - * @param the {@code payload} type - * @return the Reactive Streams {@link Publisher} - */ - public Publisher> toReactivePublisher(Executor executor) { - Assert.notNull(executor); MessageChannel channelForPublisher = this.currentMessageChannel; - if (channelForPublisher == null) { - PublishSubscribeChannel publishSubscribeChannel = new PublishSubscribeChannel(); - publishSubscribeChannel.setMinSubscribers(1); - channelForPublisher = publishSubscribeChannel; - channel(channelForPublisher); + Publisher> publisher; + if (channelForPublisher instanceof Publisher) { + publisher = (Publisher>) channelForPublisher; } + else { + MessageChannel reactiveChannel = new ReactiveChannel(); + publisher = (Publisher>) reactiveChannel; + + if (channelForPublisher != null) { + BridgeHandler bridge = new BridgeHandler(); + bridge.setOutputChannel(reactiveChannel); + + addComponent(bridge) + .addComponent(new ReactiveConsumer(channelForPublisher, bridge)) + .addComponent(reactiveChannel); + } + else { + channel(reactiveChannel); + } + + } + get(); - return new PublisherIntegrationFlow(this.integrationComponents, channelForPublisher, executor); + + return new PublisherIntegrationFlow(this.integrationComponents, publisher); } private > B register(S endpointSpec, diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublisherIntegrationFlow.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublisherIntegrationFlow.java index fe65e766f1..9bd5f01f94 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublisherIntegrationFlow.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublisherIntegrationFlow.java @@ -16,24 +16,12 @@ package org.springframework.integration.dsl; -import java.util.Queue; import java.util.Set; -import java.util.concurrent.Executor; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.atomic.AtomicReference; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageDeliveryException; -import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessagingException; -import org.springframework.messaging.PollableChannel; -import org.springframework.messaging.SubscribableChannel; /** * @@ -45,205 +33,16 @@ import org.springframework.messaging.SubscribableChannel; */ class PublisherIntegrationFlow extends StandardIntegrationFlow implements Publisher> { - private static final Subscription NO_OP_SUBSCRIPTION = new Subscription() { + private final Publisher> delegate; - @Override - public void request(long n) { - } - - @Override - public void cancel() { - } - - }; - - private final Queue>> subscribers = new LinkedBlockingQueue<>(); - - private final MessageChannel messageChannel; - - private final Executor executor; - - PublisherIntegrationFlow(Set integrationComponents, MessageChannel messageChannel, Executor executor) { + PublisherIntegrationFlow(Set integrationComponents, Publisher> publisher) { super(integrationComponents); - this.messageChannel = messageChannel; - this.executor = executor; - start(); + this.delegate = publisher; } @Override - @SuppressWarnings("unchecked") public void subscribe(Subscriber> subscriber) { - if (!isRunning()) { - //Reactive Streams Specification: https://github.com/reactive-streams/reactive-streams-jvm#1.4 - subscriber.onSubscribe(NO_OP_SUBSCRIPTION); - subscriber.onError( - new IllegalStateException("The Publisher must be started ('Lifecycle.start()') " + - "before accepting subscription.")); - return; - } - - this.subscribers.add(subscriber); - if (this.messageChannel instanceof SubscribableChannel) { - subscriber.onSubscribe(new MessageHandlerSubscription((Subscriber>) subscriber)); - } - else if (this.messageChannel instanceof PollableChannel) { - subscriber.onSubscribe(new PollableSubscription((Subscriber>) subscriber)); - } - else { - //Reactive Streams Specification: https://github.com/reactive-streams/reactive-streams-jvm#1.4 - subscriber.onSubscribe(NO_OP_SUBSCRIPTION); - subscriber.onError( - new IllegalStateException("Unsupported MessageChannel type [" - + this.messageChannel + "]. Must be 'SubscribableChannel' or 'PollableChannel'.")); - } - } - - @Override - public void stop() { - super.stop(); - shutdown(); - } - - public void shutdown() { - Subscriber> subscriber; - while ((subscriber = this.subscribers.poll()) != null) { - subscriber.onComplete(); - } - } - - - private abstract class SubscriberSubscription implements Subscription { - - final Subscriber> subscriber; - - volatile boolean terminated; - - SubscriberSubscription(Subscriber> subscriber) { - this.subscriber = subscriber; - } - - @Override - public void request(long n) { - //Reactive Streams Specification: https://github.com/reactive-streams/reactive-streams-jvm#3.9 - if (n <= 0L) { - this.subscriber.onError( - new IllegalArgumentException("Spec. Rule 3.9 - " + - "Cannot request a non strictly positive number: " + n)); - } - //Reactive Streams Specification: https://github.com/reactive-streams/reactive-streams-jvm#3.6 - else if (!this.terminated && isRunning()) { - onRequest(n); - } - } - - @Override - public void cancel() { - PublisherIntegrationFlow.this.subscribers.remove(this.subscriber); - this.terminated = true; - } - - protected abstract void onRequest(long n); - - } - - private final class MessageHandlerSubscription extends SubscriberSubscription implements MessageHandler { - - private final Queue pendingRequests = new LinkedBlockingQueue<>(); - - private final AtomicReference currentRequest = new AtomicReference<>(); - - private final AtomicLong count = new AtomicLong(); - - private volatile boolean unbounded; - - MessageHandlerSubscription(Subscriber> subscriber) { - super(subscriber); - } - - @Override - public void onRequest(long n) { - if (n == Long.MAX_VALUE) { - this.unbounded = true; - this.pendingRequests.clear(); - this.currentRequest.set(null); - this.count.set(0); - } - else if (!this.unbounded) { - if (this.currentRequest.get() != null) { - this.pendingRequests.offer(n); - } - else { - this.currentRequest.set(n); - this.count.set(0); - } - } - ((SubscribableChannel) PublisherIntegrationFlow.this.messageChannel).subscribe(this); - } - - @Override - public void handleMessage(Message message) throws MessagingException { - if (this.terminated || !PublisherIntegrationFlow.this.isRunning()) { - ((SubscribableChannel) PublisherIntegrationFlow.this.messageChannel).unsubscribe(this); - throw new MessageDeliveryException(message); - } - - if (this.unbounded) { - this.subscriber.onNext(message); - } - else { - if (this.currentRequest.get() == null || this.count.getAndIncrement() == this.currentRequest.get()) { - this.currentRequest.set(this.pendingRequests.poll()); - this.count.set(0); - if (this.currentRequest.get() == null) { - ((SubscribableChannel) PublisherIntegrationFlow.this.messageChannel).unsubscribe(this); - throw new MessageDeliveryException(message); - } - } - this.subscriber.onNext(message); - } - } - - @Override - public void cancel() { - ((SubscribableChannel) PublisherIntegrationFlow.this.messageChannel).unsubscribe(this); - super.cancel(); - } - - } - - - private final class PollableSubscription extends SubscriberSubscription { - - PollableSubscription(Subscriber> subscriber) { - super(subscriber); - } - - @Override - public void onRequest(final long n) { - PublisherIntegrationFlow.this.executor.execute(() -> { - if (n == Long.MAX_VALUE) { - while (!terminated && isRunning()) { - Message receive = - ((PollableChannel) PublisherIntegrationFlow.this.messageChannel).receive(50); - if (receive != null) { - subscriber.onNext(receive); - } - } - } - else { - long i = 0; - while (!terminated && isRunning() && i < n) { - Message receive = - ((PollableChannel) PublisherIntegrationFlow.this.messageChannel).receive(50); - if (receive != null) { - subscriber.onNext(receive); - i++; - } - } - } - }); - } - + this.delegate.subscribe(subscriber); } } 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/ReactiveConsumer.java index acd9ef245d..45b2a589c6 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/ReactiveConsumer.java @@ -16,6 +16,7 @@ package org.springframework.integration.endpoint; +import java.util.Iterator; import java.util.function.Consumer; import org.reactivestreams.Publisher; @@ -32,8 +33,14 @@ import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; -import reactor.core.publisher.DirectProcessor; +import reactor.core.Cancellation; +import reactor.core.Exceptions; +import reactor.core.Receiver; +import reactor.core.Trackable; import reactor.core.publisher.Flux; +import reactor.core.publisher.FluxSink; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Operators; /** @@ -42,48 +49,43 @@ import reactor.core.publisher.Flux; */ public class ReactiveConsumer extends AbstractEndpoint { - private final Subscriber> subscriber; + private final Operators.SubscriberAdapter, Message> subscriber; - private final Consumer> consumer; - - private volatile Flux> publisher; - - private volatile Subscription subscription; + private volatile Publisher> publisher; private ErrorHandler errorHandler; - public ReactiveConsumer(MessageChannel inputChannel, Subscriber> subscriber) { - this(inputChannel, subscriber, null); - Assert.notNull(subscriber); - - } - public ReactiveConsumer(MessageChannel inputChannel, Consumer> consumer) { - this(inputChannel, null, consumer); - Assert.notNull(consumer); + this(inputChannel, new ConsumerSubscriber(consumer)); } @SuppressWarnings("unchecked") - private ReactiveConsumer(MessageChannel inputChannel, Subscriber> subscriber, - Consumer> consumer) { + public ReactiveConsumer(MessageChannel inputChannel, Subscriber> subscriber) { Assert.notNull(inputChannel); + Assert.notNull(subscriber); - Publisher> publisher; if (inputChannel instanceof Publisher) { - publisher = (Publisher>) inputChannel; + this.publisher = (Publisher>) inputChannel; } else { - publisher = adaptToPublisher(inputChannel); + this.publisher = adaptToPublisher(inputChannel); } - this.publisher = Flux.from(publisher) - .doOnError(t -> this.errorHandler.handleError(t)) // NPE if method reference - .doOnSubscribe(s -> this.subscription = s) - .retry(); + this.subscriber = new Operators.SubscriberAdapter, Message>(subscriber) { - this.subscriber = subscriber; - this.consumer = consumer; + @Override + protected void doNext(Message message) { + try { + super.doNext(message); + } + catch (Exception e) { + ReactiveConsumer.this.errorHandler.handleError(e); + doOnSubscriberError(e); + } + } + + }; } public void setErrorHandler(ErrorHandler errorHandler) { @@ -95,26 +97,18 @@ public class ReactiveConsumer extends AbstractEndpoint { super.onInit(); if (this.errorHandler == null) { Assert.notNull(getBeanFactory(), "BeanFactory is required"); - this.errorHandler = new MessagePublishingErrorHandler( - new BeanFactoryChannelResolver(getBeanFactory())); + this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(getBeanFactory())); } } @Override protected void doStart() { - if (this.subscriber != null) { - this.publisher.subscribe(this.subscriber); - } - else { - this.publisher.subscribe(this.consumer); - } + this.publisher.subscribe(this.subscriber); } @Override protected void doStop() { - if (this.subscription != null) { - this.subscription.cancel(); - } + this.subscriber.cancel(); } private Publisher> adaptToPublisher(MessageChannel inputChannel) { @@ -135,65 +129,137 @@ public class ReactiveConsumer extends AbstractEndpoint { } private Publisher> adaptPollableChannelToPublisher(PollableChannel inputChannel) { - return null; + return new PollableChannelPublisherAdapter(inputChannel); } - private final static class SubscribableChannelPublisherAdapter - implements Publisher>, Subscriber>, Subscription { - - private final DirectProcessor> delegate = DirectProcessor.create(); - - private final MessageHandler subscriberAdapter = this.delegate.connectSink()::accept; + private final static class SubscribableChannelPublisherAdapter implements Publisher> { private final SubscribableChannel channel; - private Subscriber> actualSubscriber; - - private Subscription actualSubscription; - - SubscribableChannelPublisherAdapter(SubscribableChannel channel) { this.channel = channel; } @Override public void subscribe(Subscriber> subscriber) { - this.actualSubscriber = subscriber; - this.delegate.subscribe(this); - this.channel.subscribe(this.subscriberAdapter); + Flux. + >create(emitter -> { + MessageHandler messageHandler = emitter::next; + this.channel.subscribe(messageHandler); + emitter.setCancellation(() -> this.channel.unsubscribe(messageHandler)); + }, + FluxSink.OverflowStrategy.IGNORE) + .subscribe(subscriber); + } + + } + + private final static class PollableChannelPublisherAdapter implements Publisher> { + + private final PollableChannel channel; + + + PollableChannelPublisherAdapter(final PollableChannel channel) { + this.channel = channel; } @Override - public void onSubscribe(Subscription subscription) { - this.actualSubscription = subscription; - this.actualSubscriber.onSubscribe(this); + 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.>delayMillis(100) + .repeat() + .concatMap(value -> Flux.fromIterable(() -> messageIterator)) + .subscribe(subscriber); + } + + } + + private static final class ConsumerSubscriber implements Subscriber>, Receiver, Cancellation, Trackable { + + private final Consumer> consumer; + + private Subscription subscription; + + ConsumerSubscriber(Consumer> consumer) { + Assert.notNull(consumer); + this.consumer = consumer; + } + + @Override + public void onSubscribe(Subscription s) { + this.subscription = s; + s.request(Long.MAX_VALUE); } @Override public void onNext(Message message) { - this.actualSubscriber.onNext(message); + this.consumer.accept(message); } @Override public void onError(Throwable t) { - this.actualSubscriber.onError(t); + if (t == null) { + throw Exceptions.argumentIsNullException(); + } + onComplete(); + Operators.onErrorDropped(t); } @Override public void onComplete() { - this.actualSubscriber.onComplete(); + if (this.subscription != null) { + this.subscription = null; + } } @Override - public void request(long n) { - this.actualSubscription.request(n); + public Object upstream() { + return this.subscription; } @Override - public void cancel() { - this.channel.unsubscribe(this.subscriberAdapter); - this.actualSubscription.cancel(); + public void dispose() { + Subscription s = this.subscription; + if (s != null) { + this.subscription = null; + s.cancel(); + } + } + + @Override + public long getCapacity() { + return Long.MAX_VALUE; + } + + @Override + public boolean isStarted() { + return this.subscription != null; + } + + @Override + public boolean isTerminated() { + return false; } } 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 37402d59b2..3144b06285 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 @@ -37,7 +37,7 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; -import reactor.core.Exceptions; +import reactor.core.publisher.Operators; /** * Base class for MessageHandler implementations that provides basic validation @@ -159,11 +159,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im @Override public void onError(Throwable throwable) { - Exceptions.throwIfFatal(throwable); - if (throwable instanceof MessagingException) { - throw (MessagingException) throwable; - } - throw new MessagingException("Error occurred in message handler [" + this + "]", throwable); + Operators.onErrorDropped(throwable); } @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java index af3f7d06d8..f37c4e17c2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java @@ -16,6 +16,8 @@ package org.springframework.integration.channel.reactive; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.isOneOf; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; @@ -32,6 +34,7 @@ import org.springframework.integration.channel.ReactiveChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.support.MessageBuilder; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; @@ -53,13 +56,20 @@ public class ReactiveChannelTests { QueueChannel replyChannel = new QueueChannel(); for (int i = 0; i < 10; i++) { - this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build()); + try { + this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build()); + } + catch (Exception e) { + assertThat(e.getCause(), instanceOf(MessageHandlingException.class)); + assertThat(e.getCause().getCause(), instanceOf(IllegalStateException.class)); + assertThat(e.getMessage(), containsString("intentional")); + } } - for (int i = 0; i < 10; i++) { + for (int i = 0; i < 9; i++) { Message receive = replyChannel.receive(10000); assertNotNull(receive); - assertThat(receive.getPayload(), isOneOf("0", "1", "2", "3", "4", "5", "6", "7", "8", "9")); + assertThat(receive.getPayload(), isOneOf("0", "1", "2", "3", "4", "6", "7", "8", "9")); } } @@ -74,10 +84,9 @@ public class ReactiveChannelTests { @ServiceActivator(inputChannel = "reactiveChannel") public String handle(int payload) { - /* TODO doesn't work yet if (payload == 5) { throw new IllegalStateException("intentional"); - }*/ + } return "" + payload; } 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/ReactiveConsumerTests.java index 7b0e59691b..d10c39f6be 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/ReactiveConsumerTests.java @@ -18,29 +18,38 @@ package org.springframework.integration.channel.reactive; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.instanceOf; +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.Matchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import java.util.LinkedList; import java.util.List; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import org.hamcrest.Matchers; import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; -import org.springframework.beans.DirectFieldAccessor; 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.ReactiveChannel; import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.endpoint.ReactiveConsumer; import org.springframework.integration.handler.MethodInvokingMessageHandler; -import org.springframework.integration.test.reactive.TestSubscriber; -import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; @@ -91,10 +100,20 @@ public class ReactiveConsumerTests { @Test - public void testReactiveConsumerDirectChannel() { + @SuppressWarnings("unchecked") + public void testReactiveConsumerDirectChannel() throws InterruptedException { DirectChannel testChannel = new DirectChannel(); - TestSubscriber> testSubscriber = TestSubscriber.create(); + Subscriber> testSubscriber = (Subscriber>) Mockito.mock(Subscriber.class); + + BlockingQueue> messages = new LinkedBlockingQueue<>(); + + willAnswer(i -> { + messages.put(i.getArgumentAt(0, Message.class)); + return null; + }) + .given(testSubscriber) + .onNext(any(Message.class)); ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber); reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); @@ -104,11 +123,14 @@ public class ReactiveConsumerTests { Message testMessage = new GenericMessage<>("test"); testChannel.send(testMessage); - testSubscriber.assertSubscribed(); - testSubscriber.assertNoError(); - testSubscriber.assertNotComplete(); + ArgumentCaptor subscriptionArgumentCaptor = ArgumentCaptor.forClass(Subscription.class); + verify(testSubscriber).onSubscribe(subscriptionArgumentCaptor.capture()); + Subscription subscription = subscriptionArgumentCaptor.getValue(); - testSubscriber.assertValues(testMessage); + subscription.request(1); + + Message message = messages.poll(10, TimeUnit.SECONDS); + assertSame(testMessage, message); reactiveConsumer.stop(); @@ -120,18 +142,79 @@ public class ReactiveConsumerTests { assertThat(e, instanceOf(MessageDeliveryException.class)); } - new DirectFieldAccessor(testSubscriber).setPropertyValue("s", null); - TestUtils.getPropertyValue(testSubscriber, "values", List.class).clear(); - reactiveConsumer.start(); - testSubscriber.request(1); + subscription.request(1); testMessage = new GenericMessage<>("test2"); testChannel.send(testMessage); - testSubscriber.assertValues(testMessage); + message = messages.poll(10, TimeUnit.SECONDS); + assertSame(testMessage, message); + + verify(testSubscriber, never()).onError(any(Throwable.class)); + verify(testSubscriber, never()).onComplete(); + + assertTrue(messages.isEmpty()); + } + + @Test + @SuppressWarnings("unchecked") + public void testReactiveConsumerPollableChannel() throws InterruptedException { + QueueChannel testChannel = new QueueChannel(); + + Subscriber> testSubscriber = (Subscriber>) Mockito.mock(Subscriber.class); + + BlockingQueue> messages = new LinkedBlockingQueue<>(); + + willAnswer(i -> { + messages.put(i.getArgumentAt(0, Message.class)); + return null; + }) + .given(testSubscriber) + .onNext(any(Message.class)); + + ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber); + reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); + reactiveConsumer.afterPropertiesSet(); + reactiveConsumer.start(); + + 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); + assertSame(testMessage, message); + + reactiveConsumer.stop(); + + + testChannel.send(testMessage); + + reactiveConsumer.start(); + + subscription.request(2); + + Message testMessage2 = new GenericMessage<>("test2"); + + testChannel.send(testMessage2); + + message = messages.poll(10, TimeUnit.SECONDS); + assertSame(testMessage, message); + + message = messages.poll(10, TimeUnit.SECONDS); + assertSame(testMessage2, message); + + verify(testSubscriber, never()).onError(any(Throwable.class)); + verify(testSubscriber, never()).onComplete(); + + assertTrue(messages.isEmpty()); } @Test 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 207f49aef5..ffc1f5cce2 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 @@ -33,9 +33,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.stream.Collectors; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.reactivestreams.Publisher; @@ -100,7 +98,6 @@ public class ReactiveStreamsTests { } @Test - @Ignore("Until Reactor 3.0.x solution") public void testPollableReactiveFlow() throws InterruptedException, TimeoutException, ExecutionException { this.inputChannel.send(new GenericMessage<>("1,2,3,4,5")); @@ -109,19 +106,20 @@ public class ReactiveStreamsTests { Flux.from(this.pollablePublisher) .filter(m -> m.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)) .doOnNext(p -> latch.countDown()) - .subscribe(6); + .take(6) + .subscribe(); Future> future = Executors.newSingleThreadExecutor().submit(() -> - Flux.fromArray(new String[] { "11,12,13" }) + Flux.just("11,12,13") .map(v -> v.split(",")) - .map(Arrays::asList) - .flatMapIterable(data -> data) + .flatMapIterable(Arrays::asList) .map(Integer::parseInt) .>map(GenericMessage::new) .concatWith(this.pollablePublisher) .map(Message::getPayload) - .collect(Collectors.toList()) + .take(7) + .collectList() .block(Duration.ofSeconds(5))); this.inputChannel.send(new GenericMessage<>("6,7,8,9,10")); diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/reactive/TestSubscriber.java b/spring-integration-test/src/main/java/org/springframework/integration/test/reactive/TestSubscriber.java deleted file mode 100644 index c6af316c6e..0000000000 --- a/spring-integration-test/src/main/java/org/springframework/integration/test/reactive/TestSubscriber.java +++ /dev/null @@ -1,1164 +0,0 @@ -/* - * Copyright 2016 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 - * - * http://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.test.reactive; - -import java.time.Duration; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Iterator; -import java.util.LinkedList; -import java.util.List; -import java.util.Objects; -import java.util.Set; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLongFieldUpdater; -import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; -import java.util.function.BooleanSupplier; -import java.util.function.Consumer; -import java.util.function.Supplier; - -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; - -import reactor.core.Fuseable; -import reactor.core.Receiver; -import reactor.core.Trackable; -import reactor.core.publisher.Operators; - -/** - * A Reactor based Subscriber implementation that hosts assertion tests for its state and - * allows asynchronous cancellation and requesting. - * - *

To create a new instance of {@link TestSubscriber}, you have the choice between - * these static methods: - *

    - *
  • {@link TestSubscriber#subscribe(Publisher)}: create a new {@link TestSubscriber}, - * subscribe to it with the specified {@link Publisher} and requests an unbounded - * number of elements.
    • - *
  • {@link TestSubscriber#subscribe(Publisher, long)}: create a new {@link TestSubscriber}, - * subscribe to it with the specified {@link Publisher} and requests {@code n} elements - * (can be 0 if you want no initial demand). - *
  • {@link TestSubscriber#create()}: create a new {@link TestSubscriber} and requests - * an unbounded number of elements.
    • - *
  • {@link TestSubscriber#create(long)}: create a new {@link TestSubscriber} and - * requests {@code n} elements (can be 0 if you want no initial demand). - *
- * - *

If you are testing asynchronous publishers, don't forget to use one of the - * {@code await*()} methods to wait for the data to assert. - * - *

You can extend this class but only the onNext, onError and onComplete can be overridden. - * You can call {@link #request(long)} and {@link #cancel()} from any thread or from within - * the overridable methods but you should avoid calling the assertXXX methods asynchronously. - * - *

Usage: - * 

- * {@code
- * TestSubscriber
- *   .subscribe(publisher)
- *   .await()
- *   .assertValues("ABC", "DEF");
- * }
- *
- * - * @param the value type. - * - * @author Sebastien Deleuze - * @author David Karnok - * @author Anatoly Kadyshev - * @author Stephane Maldini - * @author Brian Clozel - */ -public class TestSubscriber - implements Subscriber, Subscription, Trackable, Receiver { - - /** - * Default timeout for waiting next values to be received - */ - public static final Duration DEFAULT_VALUES_TIMEOUT = Duration.ofSeconds(3); - - @SuppressWarnings("rawtypes") - private static final AtomicLongFieldUpdater REQUESTED = - AtomicLongFieldUpdater.newUpdater(TestSubscriber.class, "requested"); - - @SuppressWarnings("rawtypes") - private static final AtomicReferenceFieldUpdater NEXT_VALUES = - AtomicReferenceFieldUpdater.newUpdater(TestSubscriber.class, List.class, - "values"); - - @SuppressWarnings("rawtypes") - private static final AtomicReferenceFieldUpdater S = - AtomicReferenceFieldUpdater.newUpdater(TestSubscriber.class, Subscription.class, "s"); - - - private final List errors = new LinkedList<>(); - - private final CountDownLatch cdl = new CountDownLatch(1); - - volatile Subscription s; - - volatile long requested; - - volatile List values = new LinkedList<>(); - - /** - * The fusion mode to request. - */ - private int requestedFusionMode = -1; - - /** - * The established fusion mode. - */ - private volatile int establishedFusionMode = -1; - - /** - * The fuseable QueueSubscription in case a fusion mode was specified. - */ - private Fuseable.QueueSubscription qs; - - private int subscriptionCount = 0; - - private int completionCount = 0; - - private volatile long valueCount = 0L; - - private volatile long nextValueAssertedCount = 0L; - - private Duration valuesTimeout = DEFAULT_VALUES_TIMEOUT; - - private boolean valuesStorage = true; - -// ============================================================================================================== -// Static methods -// ============================================================================================================== - - /** - * Blocking method that waits until {@code conditionSupplier} returns true, or if it - * does not before the specified timeout, throws an {@link AssertionError} with the - * specified error message supplier. - * - * @param timeout the timeout duration - * @param errorMessageSupplier the error message supplier - * @param conditionSupplier condition to break out of the wait loop - */ - public static void await(Duration timeout, Supplier errorMessageSupplier, - BooleanSupplier conditionSupplier) { - - Objects.requireNonNull(errorMessageSupplier); - Objects.requireNonNull(conditionSupplier); - Objects.requireNonNull(timeout); - - long timeoutNs = timeout.toNanos(); - long startTime = System.nanoTime(); - do { - if (conditionSupplier.getAsBoolean()) { - return; - } - try { - Thread.sleep(100); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new RuntimeException(e); - } - } - while (System.nanoTime() - startTime < timeoutNs); - throw new AssertionError(errorMessageSupplier.get()); - } - - /** - * Blocking method that waits until {@code conditionSupplier} returns true, or if it - * does not before the specified timeout, throw an {@link AssertionError} with the - * specified error message. - * - * @param timeout the timeout duration - * @param errorMessage the error message - * @param conditionSupplier condition to break out of the wait loop - */ - public static void await(Duration timeout, - final String errorMessage, - BooleanSupplier conditionSupplier) { - await(timeout, () -> errorMessage, conditionSupplier); - } - - /** - * Create a new {@link TestSubscriber} that requests an unbounded number of elements. - *

Be sure at least a publisher has subscribed to it via {@link Publisher#subscribe(Subscriber)} - * before use assert methods. - * @see #subscribe(Publisher) - * @param the observed value type - * @return a fresh TestSubscriber instance - */ - public static TestSubscriber create() { - return new TestSubscriber<>(); - } - - /** - * Create a new {@link TestSubscriber} that requests initially {@code n} elements. You - * can then manage the demand with {@link Subscription#request(long)}. - *

Be sure at least a publisher has subscribed to it via {@link Publisher#subscribe(Subscriber)} - * before use assert methods. - * @param n Number of elements to request (can be 0 if you want no initial demand). - * @see #subscribe(Publisher, long) - * @param the observed value type - * @return a fresh TestSubscriber instance - */ - public static TestSubscriber create(long n) { - return new TestSubscriber<>(n); - } - - /** - * Create a new {@link TestSubscriber} that requests an unbounded number of elements, - * and make the specified {@code publisher} subscribe to it. - * @param publisher The publisher to subscribe with - * @param the observed value type - * @return a fresh TestSubscriber instance - */ - public static TestSubscriber subscribe(Publisher publisher) { - TestSubscriber subscriber = new TestSubscriber<>(); - publisher.subscribe(subscriber); - return subscriber; - } - - /** - * Create a new {@link TestSubscriber} that requests initially {@code n} elements, - * and make the specified {@code publisher} subscribe to it. You can then manage the - * demand with {@link Subscription#request(long)}. - * @param publisher The publisher to subscribe with - * @param n Number of elements to request (can be 0 if you want no initial demand). - * @param the observed value type - * @return a fresh TestSubscriber instance - */ - public static TestSubscriber subscribe(Publisher publisher, long n) { - TestSubscriber subscriber = new TestSubscriber<>(n); - publisher.subscribe(subscriber); - return subscriber; - } - -// ============================================================================================================== -// Private constructors -// ============================================================================================================== - - private TestSubscriber() { - this(Long.MAX_VALUE); - } - - private TestSubscriber(long n) { - if (n < 0) { - throw new IllegalArgumentException("initialRequest >= required but it was " + n); - } - REQUESTED.lazySet(this, n); - } - -// ============================================================================================================== -// Configuration -// ============================================================================================================== - - - /** - * Enable or disabled the values storage. It is enabled by default, and can be disable - * in order to be able to perform performance benchmarks or tests with a huge amount - * values. - * @param enabled enable value storage? - * @return this - */ - public final TestSubscriber configureValuesStorage(boolean enabled) { - this.valuesStorage = enabled; - return this; - } - - /** - * Configure the timeout in seconds for waiting next values to be received (3 seconds - * by default). - * @param timeout the new default value timeout duration - * @return this - */ - public final TestSubscriber configureValuesTimeout(Duration timeout) { - this.valuesTimeout = timeout; - return this; - } - - /** - * Returns the established fusion mode or -1 if it was not enabled - * - * @return the fusion mode, see Fuseable constants - */ - public final int establishedFusionMode() { - return establishedFusionMode; - } - -// ============================================================================================================== -// Assertions -// ============================================================================================================== - - /** - * Assert a complete successfully signal has been received. - * @return this - */ - public final TestSubscriber assertComplete() { - assertNoError(); - int c = completionCount; - if (c == 0) { - throw new AssertionError("Not completed", null); - } - if (c > 1) { - throw new AssertionError("Multiple completions: " + c, null); - } - return this; - } - - /** - * Assert the specified values have been received. Values storage should be enabled to - * use this method. - * @param expectedValues the values to assert - * @see #configureValuesStorage(boolean) - * @return this - */ - public final TestSubscriber assertContainValues(Set expectedValues) { - if (!valuesStorage) { - throw new IllegalStateException( - "Using assertNoValues() requires enabling values storage"); - } - if (expectedValues.size() > values.size()) { - throw new AssertionError("Actual contains fewer elements" + values, null); - } - - Iterator expected = expectedValues.iterator(); - - while (true) { - boolean n2 = expected.hasNext(); - if (n2) { - T t2 = expected.next(); - if (!values.contains(t2)) { - throw new AssertionError("The element is not contained in the " + - "received resuls" + - " = " + valueAndClass(t2), null); - } - } - else { - break; - } - } - return this; - } - - /** - * Assert an error signal has been received. - * @return this - */ - public final TestSubscriber assertError() { - assertNotComplete(); - int s = errors.size(); - if (s == 0) { - throw new AssertionError("No error", null); - } - if (s > 1) { - throw new AssertionError("Multiple errors: " + s, null); - } - return this; - } - - /** - * Assert an error signal has been received. - * @param clazz The class of the exception contained in the error signal - * @return this - */ - public final TestSubscriber assertError(Class clazz) { - assertNotComplete(); - int s = errors.size(); - if (s == 0) { - throw new AssertionError("No error", null); - } - if (s == 1) { - Throwable e = errors.get(0); - if (!clazz.isInstance(e)) { - throw new AssertionError("Error class incompatible: expected = " + - clazz + ", actual = " + e, null); - } - } - if (s > 1) { - throw new AssertionError("Multiple errors: " + s, null); - } - return this; - } - - public final TestSubscriber assertErrorMessage(String message) { - assertNotComplete(); - int s = errors.size(); - if (s == 0) { - assertionError("No error", null); - } - if (s == 1) { - if (!Objects.equals(message, - errors.get(0) - .getMessage())) { - assertionError("Error class incompatible: expected = \"" + message + - "\", actual = \"" + errors.get(0).getMessage() + "\"", null); - } - } - if (s > 1) { - assertionError("Multiple errors: " + s, null); - } - - return this; - } - - /** - * Assert an error signal has been received. - * @param expectation A method that can verify the exception contained in the error signal - * and throw an exception (like an {@link AssertionError}) if the exception is not valid. - * @return this - */ - public final TestSubscriber assertErrorWith(Consumer expectation) { - assertNotComplete(); - int s = errors.size(); - if (s == 0) { - throw new AssertionError("No error", null); - } - if (s == 1) { - expectation.accept(errors.get(0)); - } - if (s > 1) { - throw new AssertionError("Multiple errors: " + s, null); - } - return this; - } - - /** - * Assert that the upstream was a Fuseable source. - * - * @return this - */ - public final TestSubscriber assertFuseableSource() { - if (qs == null) { - throw new AssertionError("Upstream was not Fuseable"); - } - return this; - } - - /** - * Assert that the fusion mode was granted. - * - * @return this - */ - public final TestSubscriber assertFusionEnabled() { - if (establishedFusionMode != Fuseable.SYNC && establishedFusionMode != Fuseable.ASYNC) { - throw new AssertionError("Fusion was not enabled"); - } - return this; - } - - public final TestSubscriber assertFusionMode(int expectedMode) { - if (establishedFusionMode != expectedMode) { - throw new AssertionError("Wrong fusion mode: expected: " + fusionModeName( - expectedMode) + ", actual: " + fusionModeName(establishedFusionMode)); - } - return this; - } - - /** - * Assert that the fusion mode was granted. - * - * @return this - */ - public final TestSubscriber assertFusionRejected() { - if (establishedFusionMode != Fuseable.NONE) { - throw new AssertionError("Fusion was granted"); - } - return this; - } - - /** - * Assert no error signal has been received. - * @return this - */ - public final TestSubscriber assertNoError() { - int s = errors.size(); - if (s == 1) { - Throwable e = errors.get(0); - String valueAndClass = e == null ? null : e + " (" + e.getClass().getSimpleName() + ")"; - throw new AssertionError("Error present: " + valueAndClass, null); - } - if (s > 1) { - throw new AssertionError("Multiple errors: " + s, null); - } - return this; - } - - /** - * Assert no values have been received. - * - * @return this - */ - public final TestSubscriber assertNoValues() { - if (valueCount != 0) { - throw new AssertionError("No values expected but received: [length = " + values.size() + "] " + values, - null); - } - return this; - } - - /** - * Assert that the upstream was not a Fuseable source. - * @return this - */ - public final TestSubscriber assertNonFuseableSource() { - if (qs != null) { - throw new AssertionError("Upstream was Fuseable"); - } - return this; - } - - /** - * Assert no complete successfully signal has been received. - * @return this - */ - public final TestSubscriber assertNotComplete() { - int c = completionCount; - if (c == 1) { - throw new AssertionError("Completed", null); - } - if (c > 1) { - throw new AssertionError("Multiple completions: " + c, null); - } - return this; - } - - /** - * Assert no subscription occurred. - * - * @return this - */ - public final TestSubscriber assertNotSubscribed() { - int s = subscriptionCount; - - if (s == 1) { - throw new AssertionError("OnSubscribe called once", null); - } - if (s > 1) { - throw new AssertionError("OnSubscribe called multiple times: " + s, null); - } - - return this; - } - - /** - * Assert no complete successfully or error signal has been received. - * @return this - */ - public final TestSubscriber assertNotTerminated() { - if (cdl.getCount() == 0) { - throw new AssertionError("Terminated", null); - } - return this; - } - - /** - * Assert subscription occurred (once). - * @return this - */ - public final TestSubscriber assertSubscribed() { - int s = subscriptionCount; - - if (s == 0) { - throw new AssertionError("OnSubscribe not called", null); - } - if (s > 1) { - throw new AssertionError("OnSubscribe called multiple times: " + s, null); - } - - return this; - } - - /** - * Assert either complete successfully or error signal has been received. - * @return this - */ - public final TestSubscriber assertTerminated() { - if (cdl.getCount() != 0) { - throw new AssertionError("Not terminated", null); - } - return this; - } - - /** - * Assert {@code n} values has been received. - * - * @param n the expected value count - * - * @return this - */ - public final TestSubscriber assertValueCount(long n) { - if (valueCount != n) { - throw new AssertionError("Different value count: expected = " + n + ", actual = " + valueCount, - null); - } - return this; - } - - /** - * Assert the specified values have been received in the same order read by the - * passed {@link Iterable}. Values storage - * should be enabled to - * use this method. - * @param expectedSequence the values to assert - * @see #configureValuesStorage(boolean) - * @return this - */ - public final TestSubscriber assertValueSequence(Iterable expectedSequence) { - if (!valuesStorage) { - throw new IllegalStateException("Using assertNoValues() requires enabling values storage"); - } - Iterator actual = values.iterator(); - Iterator expected = expectedSequence.iterator(); - int i = 0; - while (true) { - boolean n1 = actual.hasNext(); - boolean n2 = expected.hasNext(); - if (n1 && n2) { - T t1 = actual.next(); - T t2 = expected.next(); - if (!Objects.equals(t1, t2)) { - throw new AssertionError("The element with index " + i + " does not match: expected = " + valueAndClass(t2) + ", actual = " - + valueAndClass( - t1), null); - } - i++; - } - else if (n1 && !n2) { - throw new AssertionError("Actual contains more elements" + values, null); - } - else if (!n1 && n2) { - throw new AssertionError("Actual contains fewer elements: " + values, null); - } - else { - break; - } - } - return this; - } - - /** - * Assert the specified values have been received in the declared order. Values - * storage should be enabled to use this method. - * - * @param expectedValues the values to assert - * - * @return this - * - * @see #configureValuesStorage(boolean) - */ - @SafeVarargs - @SuppressWarnings("varargs") - public final TestSubscriber assertValues(T... expectedValues) { - return assertValueSequence(Arrays.asList(expectedValues)); - } - - /** - * Assert the specified values have been received in the declared order. Values - * storage should be enabled to use this method. - * - * @param expectations One or more methods that can verify the values and throw a - * exception (like an {@link AssertionError}) if the value is not valid. - * - * @return this - * - * @see #configureValuesStorage(boolean) - */ - @SafeVarargs - public final TestSubscriber assertValuesWith(Consumer... expectations) { - if (!valuesStorage) { - throw new IllegalStateException( - "Using assertNoValues() requires enabling values storage"); - } - final int expectedValueCount = expectations.length; - if (expectedValueCount != values.size()) { - throw new AssertionError("Different value count: expected = " + expectedValueCount + ", actual = " + valueCount, null); - } - for (int i = 0; i < expectedValueCount; i++) { - Consumer consumer = expectations[i]; - T actualValue = values.get(i); - consumer.accept(actualValue); - } - return this; - } - -// ============================================================================================================== -// Await methods -// ============================================================================================================== - - /** - * Blocking method that waits until a complete successfully or error signal is received. - * @return this - */ - public final TestSubscriber await() { - if (cdl.getCount() == 0) { - return this; - } - try { - cdl.await(); - } - catch (InterruptedException ex) { - throw new AssertionError("Wait interrupted", ex); - } - return this; - } - - /** - * Blocking method that waits until a complete successfully or error signal is received - * or until a timeout occurs. - * @param timeout The timeout value - * @return this - */ - public final TestSubscriber await(Duration timeout) { - if (cdl.getCount() == 0) { - return this; - } - try { - if (!cdl.await(timeout.toMillis(), TimeUnit.MILLISECONDS)) { - throw new AssertionError("No complete or error signal before timeout"); - } - return this; - } - catch (InterruptedException ex) { - throw new AssertionError("Wait interrupted", ex); - } - } - - /** - * Blocking method that waits until {@code n} next values have been received. - * - * @param n the value count to assert - * - * @return this - */ - public final TestSubscriber awaitAndAssertNextValueCount(final long n) { - await(valuesTimeout, () -> { - if (valuesStorage) { - return String.format("%d out of %d next values received within %d, " + - "values : %s", - valueCount - nextValueAssertedCount, - n, - valuesTimeout.toMillis(), - values.toString() - ); - } - return String.format("%d out of %d next values received within %d", - valueCount - nextValueAssertedCount, - n, - valuesTimeout.toMillis()); - }, () -> valueCount >= (nextValueAssertedCount + n)); - nextValueAssertedCount += n; - return this; - } - - /** - * Blocking method that waits until {@code n} next values have been received (n is the - * number of values provided) to assert them. - * - * @param values the values to assert - * - * @return this - */ - @SafeVarargs - @SuppressWarnings({ "unchecked", "rawtypes" }) - public final TestSubscriber awaitAndAssertNextValues(T... values) { - final int expectedNum = values.length; - final List> expectations = new ArrayList<>(); - for (int i = 0; i < expectedNum; i++) { - final T expectedValue = values[i]; - expectations.add(actualValue -> { - if (!actualValue.equals(expectedValue)) { - throw new AssertionError(String.format( - "Expected Next signal: %s, but got: %s", - expectedValue, - actualValue)); - } - }); - } - awaitAndAssertNextValuesWith(expectations.toArray((Consumer[]) new Consumer[0])); - return this; - } - - /** - * Blocking method that waits until {@code n} next values have been received - * (n is the number of expectations provided) to assert them. - * @param expectations One or more methods that can verify the values and throw a - * exception (like an {@link AssertionError}) if the value is not valid. - * @return this - */ - @SafeVarargs - public final TestSubscriber awaitAndAssertNextValuesWith(Consumer... expectations) { - valuesStorage = true; - final int expectedValueCount = expectations.length; - await(valuesTimeout, () -> { - if (valuesStorage) { - return String.format("%d out of %d next values received within %d, " + - "values : %s", - valueCount - nextValueAssertedCount, - expectedValueCount, - valuesTimeout.toMillis(), - values.toString() - ); - } - return String.format("%d out of %d next values received within %d ms", - valueCount - nextValueAssertedCount, - expectedValueCount, - valuesTimeout.toMillis()); - }, () -> valueCount >= (nextValueAssertedCount + expectedValueCount)); - List nextValuesSnapshot; - List empty = new ArrayList<>(); - while (true) { - nextValuesSnapshot = values; - if (NEXT_VALUES.compareAndSet(this, values, empty)) { - break; - } - } - if (nextValuesSnapshot.size() < expectedValueCount) { - throw new AssertionError(String.format("Expected %d number of signals but received %d", - expectedValueCount, - nextValuesSnapshot.size())); - } - for (int i = 0; i < expectedValueCount; i++) { - Consumer consumer = expectations[i]; - T actualValue = nextValuesSnapshot.get(i); - consumer.accept(actualValue); - } - nextValueAssertedCount += expectedValueCount; - return this; - } - -// ============================================================================================================== -// Overrides -// ============================================================================================================== - - @Override - public void cancel() { - Subscription a = s; - if (a != Operators.cancelledSubscription()) { - a = S.getAndSet(this, Operators.cancelledSubscription()); - if (a != null && a != Operators.cancelledSubscription()) { - a.cancel(); - } - } - } - - @Override - public final boolean isCancelled() { - return s == Operators.cancelledSubscription(); - } - - @Override - public final boolean isStarted() { - return s != null; - } - - @Override - public final boolean isTerminated() { - return isCancelled(); - } - - @Override - public void onComplete() { - completionCount++; - cdl.countDown(); - } - - @Override - public void onError(Throwable t) { - errors.add(t); - cdl.countDown(); - } - - @Override - public void onNext(T t) { - if (establishedFusionMode == Fuseable.ASYNC) { - while (true) { - t = qs.poll(); - if (t == null) { - break; - } - valueCount++; - if (valuesStorage) { - List nextValuesSnapshot; - while (true) { - nextValuesSnapshot = values; - nextValuesSnapshot.add(t); - if (NEXT_VALUES.compareAndSet(this, - nextValuesSnapshot, - nextValuesSnapshot)) { - break; - } - } - } - } - } - else { - valueCount++; - if (valuesStorage) { - List nextValuesSnapshot; - while (true) { - nextValuesSnapshot = values; - nextValuesSnapshot.add(t); - if (NEXT_VALUES.compareAndSet(this, - nextValuesSnapshot, - nextValuesSnapshot)) { - break; - } - } - } - } - } - - @Override - @SuppressWarnings("unchecked") - public void onSubscribe(Subscription s) { - subscriptionCount++; - int requestMode = requestedFusionMode; - if (requestMode >= 0) { - if (!setWithoutRequesting(s)) { - if (!isCancelled()) { - errors.add(new IllegalStateException("Subscription already set: " + - subscriptionCount)); - } - } - else { - if (s instanceof Fuseable.QueueSubscription) { - this.qs = (Fuseable.QueueSubscription) s; - - int m = qs.requestFusion(requestMode); - establishedFusionMode = m; - - if (m == Fuseable.SYNC) { - while (true) { - T v = qs.poll(); - if (v == null) { - onComplete(); - break; - } - - onNext(v); - } - } - else { - requestDeferred(); - } - } - else { - requestDeferred(); - } - } - } - else { - if (!set(s)) { - if (!isCancelled()) { - errors.add(new IllegalStateException("Subscription already set: " + - subscriptionCount)); - } - } - } - } - - @Override - public void request(long n) { - if (Operators.validate(n)) { - if (establishedFusionMode != Fuseable.SYNC) { - normalRequest(n); - } - } - } - - @Override - public final long requestedFromDownstream() { - return requested; - } - - /** - * Setup what fusion mode should be requested from the incomining - * Subscription if it happens to be QueueSubscription - * @param requestMode the mode to request, see Fuseable constants - * @return this - */ - public final TestSubscriber requestedFusionMode(int requestMode) { - this.requestedFusionMode = requestMode; - return this; - } - - @Override - public Subscription upstream() { - return s; - } - - -// ============================================================================================================== -// Non public methods -// ============================================================================================================== - - protected final void normalRequest(long n) { - Subscription a = s; - if (a != null) { - a.request(n); - } - else { - Operators.addAndGet(REQUESTED, this, n); - - a = s; - - if (a != null) { - long r = REQUESTED.getAndSet(this, 0L); - - if (r != 0L) { - a.request(r); - } - } - } - } - - /** - * Requests the deferred amount if not zero. - */ - protected final void requestDeferred() { - long r = REQUESTED.getAndSet(this, 0L); - - if (r != 0L) { - s.request(r); - } - } - - /** - * Atomically sets the single subscription and requests the missed amount from it. - * - * @param s the Subscription to set. - * @return false if this arbiter is cancelled or there was a subscription already set - */ - protected final boolean set(Subscription s) { - Objects.requireNonNull(s, "s"); - Subscription a = this.s; - if (a == Operators.cancelledSubscription()) { - s.cancel(); - return false; - } - if (a != null) { - s.cancel(); - Operators.reportSubscriptionSet(); - return false; - } - - if (S.compareAndSet(this, null, s)) { - - long r = REQUESTED.getAndSet(this, 0L); - - if (r != 0L) { - s.request(r); - } - - return true; - } - - a = this.s; - - if (a != Operators.cancelledSubscription()) { - s.cancel(); - return false; - } - - Operators.reportSubscriptionSet(); - return false; - } - - /** - * Sets the Subscription once but does not request anything. - * @param s the Subscription to set - * @return true if successful, false if the current subscription is not null - */ - protected final boolean setWithoutRequesting(Subscription s) { - Objects.requireNonNull(s, "s"); - while (true) { - Subscription a = this.s; - if (a == Operators.cancelledSubscription()) { - s.cancel(); - return false; - } - if (a != null) { - s.cancel(); - Operators.reportSubscriptionSet(); - return false; - } - - if (S.compareAndSet(this, null, s)) { - return true; - } - } - } - - /** - * Prepares and throws an AssertionError exception based on the message, cause, the - * active state and the potential errors so far. - * - * @param message the message - * @param cause the optional Throwable cause - * - * @throws AssertionError as expected - */ - protected final void assertionError(String message, Throwable cause) { - StringBuilder b = new StringBuilder(); - - if (cdl.getCount() != 0) { - b.append("(active) "); - } - b.append(message); - - List err = errors; - if (!err.isEmpty()) { - b.append(" (+ ") - .append(err.size()) - .append(" errors)"); - } - AssertionError e = new AssertionError(b.toString(), cause); - - for (Throwable t : err) { - e.addSuppressed(t); - } - - throw e; - } - - protected final String fusionModeName(int mode) { - switch (mode) { - case -1: - return "Disabled"; - case Fuseable.NONE: - return "None"; - case Fuseable.SYNC: - return "Sync"; - case Fuseable.ASYNC: - return "Async"; - default: - return "Unknown(" + mode + ")"; - } - } - - protected final String valueAndClass(Object o) { - if (o == null) { - return null; - } - return o + " (" + o.getClass().getSimpleName() + ")"; - } - -}