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<Message<?>>` instance to the `subscribe()` to avoid race conditions when we have several subscribers. In other words make `Iterator<Message<?>>` 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`
This commit is contained in:
Artem Bilan
committed by
Gary Russell
Gary Russell
parent
8070f72c44
commit
ba576bbc0d
@@ -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"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<B extends IntegrationFlowDefinit
|
||||
|
||||
/**
|
||||
* Represent an Integration Flow as a Reactive Streams {@link Publisher} bean.
|
||||
* @param <T> the {@code payload} type
|
||||
* @return the Reactive Streams {@link Publisher}
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public <T> Publisher<Message<T>> 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 <T> the {@code payload} type
|
||||
* @return the Reactive Streams {@link Publisher}
|
||||
*/
|
||||
public <T> Publisher<Message<T>> 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<Message<T>> publisher;
|
||||
if (channelForPublisher instanceof Publisher) {
|
||||
publisher = (Publisher<Message<T>>) channelForPublisher;
|
||||
}
|
||||
else {
|
||||
MessageChannel reactiveChannel = new ReactiveChannel();
|
||||
publisher = (Publisher<Message<T>>) 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<T>(this.integrationComponents, channelForPublisher, executor);
|
||||
|
||||
return new PublisherIntegrationFlow<T>(this.integrationComponents, publisher);
|
||||
}
|
||||
|
||||
private <S extends ConsumerEndpointSpec<S, ? extends MessageHandler>> B register(S endpointSpec,
|
||||
|
||||
@@ -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<T> extends StandardIntegrationFlow implements Publisher<Message<T>> {
|
||||
|
||||
private static final Subscription NO_OP_SUBSCRIPTION = new Subscription() {
|
||||
private final Publisher<Message<T>> delegate;
|
||||
|
||||
@Override
|
||||
public void request(long n) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel() {
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
private final Queue<Subscriber<? super Message<T>>> subscribers = new LinkedBlockingQueue<>();
|
||||
|
||||
private final MessageChannel messageChannel;
|
||||
|
||||
private final Executor executor;
|
||||
|
||||
PublisherIntegrationFlow(Set<Object> integrationComponents, MessageChannel messageChannel, Executor executor) {
|
||||
PublisherIntegrationFlow(Set<Object> integrationComponents, Publisher<Message<T>> publisher) {
|
||||
super(integrationComponents);
|
||||
this.messageChannel = messageChannel;
|
||||
this.executor = executor;
|
||||
start();
|
||||
this.delegate = publisher;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public void subscribe(Subscriber<? super Message<T>> 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<Message<?>>) subscriber));
|
||||
}
|
||||
else if (this.messageChannel instanceof PollableChannel) {
|
||||
subscriber.onSubscribe(new PollableSubscription((Subscriber<Message<?>>) 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<? super Message<T>> subscriber;
|
||||
while ((subscriber = this.subscribers.poll()) != null) {
|
||||
subscriber.onComplete();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private abstract class SubscriberSubscription implements Subscription {
|
||||
|
||||
final Subscriber<Message<?>> subscriber;
|
||||
|
||||
volatile boolean terminated;
|
||||
|
||||
SubscriberSubscription(Subscriber<Message<?>> 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<Long> pendingRequests = new LinkedBlockingQueue<>();
|
||||
|
||||
private final AtomicReference<Long> currentRequest = new AtomicReference<>();
|
||||
|
||||
private final AtomicLong count = new AtomicLong();
|
||||
|
||||
private volatile boolean unbounded;
|
||||
|
||||
MessageHandlerSubscription(Subscriber<Message<?>> 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<Message<?>> 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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<Message<?>> subscriber;
|
||||
private final Operators.SubscriberAdapter<Message<?>, Message<?>> subscriber;
|
||||
|
||||
private final Consumer<Message<?>> consumer;
|
||||
|
||||
private volatile Flux<Message<?>> publisher;
|
||||
|
||||
private volatile Subscription subscription;
|
||||
private volatile Publisher<Message<?>> publisher;
|
||||
|
||||
private ErrorHandler errorHandler;
|
||||
|
||||
|
||||
public ReactiveConsumer(MessageChannel inputChannel, Subscriber<Message<?>> subscriber) {
|
||||
this(inputChannel, subscriber, null);
|
||||
Assert.notNull(subscriber);
|
||||
|
||||
}
|
||||
|
||||
public ReactiveConsumer(MessageChannel inputChannel, Consumer<Message<?>> consumer) {
|
||||
this(inputChannel, null, consumer);
|
||||
Assert.notNull(consumer);
|
||||
this(inputChannel, new ConsumerSubscriber(consumer));
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private ReactiveConsumer(MessageChannel inputChannel, Subscriber<Message<?>> subscriber,
|
||||
Consumer<Message<?>> consumer) {
|
||||
public ReactiveConsumer(MessageChannel inputChannel, Subscriber<Message<?>> subscriber) {
|
||||
Assert.notNull(inputChannel);
|
||||
Assert.notNull(subscriber);
|
||||
|
||||
Publisher<Message<?>> publisher;
|
||||
if (inputChannel instanceof Publisher) {
|
||||
publisher = (Publisher<Message<?>>) inputChannel;
|
||||
this.publisher = (Publisher<Message<?>>) 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<?>, 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<Message<?>> adaptToPublisher(MessageChannel inputChannel) {
|
||||
@@ -135,65 +129,137 @@ public class ReactiveConsumer extends AbstractEndpoint {
|
||||
}
|
||||
|
||||
private Publisher<Message<?>> adaptPollableChannelToPublisher(PollableChannel inputChannel) {
|
||||
return null;
|
||||
return new PollableChannelPublisherAdapter(inputChannel);
|
||||
}
|
||||
|
||||
|
||||
private final static class SubscribableChannelPublisherAdapter
|
||||
implements Publisher<Message<?>>, Subscriber<Message<?>>, Subscription {
|
||||
|
||||
private final DirectProcessor<Message<?>> delegate = DirectProcessor.create();
|
||||
|
||||
private final MessageHandler subscriberAdapter = this.delegate.connectSink()::accept;
|
||||
private final static class SubscribableChannelPublisherAdapter implements Publisher<Message<?>> {
|
||||
|
||||
private final SubscribableChannel channel;
|
||||
|
||||
private Subscriber<? super Message<?>> actualSubscriber;
|
||||
|
||||
private Subscription actualSubscription;
|
||||
|
||||
|
||||
SubscribableChannelPublisherAdapter(SubscribableChannel channel) {
|
||||
this.channel = channel;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void subscribe(Subscriber<? super Message<?>> subscriber) {
|
||||
this.actualSubscriber = subscriber;
|
||||
this.delegate.subscribe(this);
|
||||
this.channel.subscribe(this.subscriberAdapter);
|
||||
Flux.
|
||||
<Message<?>>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<Message<?>> {
|
||||
|
||||
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<? super Message<?>> subscriber) {
|
||||
Iterator<Message<?>> messageIterator = new Iterator<Message<?>>() {
|
||||
|
||||
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.<Message<?>>delayMillis(100)
|
||||
.repeat()
|
||||
.concatMap(value -> Flux.fromIterable(() -> messageIterator))
|
||||
.subscribe(subscriber);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static final class ConsumerSubscriber implements Subscriber<Message<?>>, Receiver, Cancellation, Trackable {
|
||||
|
||||
private final Consumer<Message<?>> consumer;
|
||||
|
||||
private Subscription subscription;
|
||||
|
||||
ConsumerSubscriber(Consumer<Message<?>> 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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Message<?>> testSubscriber = TestSubscriber.create();
|
||||
Subscriber<Message<?>> testSubscriber = (Subscriber<Message<?>>) Mockito.mock(Subscriber.class);
|
||||
|
||||
BlockingQueue<Message<?>> 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<Subscription> 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<Message<?>> testSubscriber = (Subscriber<Message<?>>) Mockito.mock(Subscriber.class);
|
||||
|
||||
BlockingQueue<Message<?>> 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<Subscription> 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
|
||||
|
||||
@@ -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<List<Integer>> 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)
|
||||
.<Message<Integer>>map(GenericMessage<Integer>::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"));
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user