Skip to content

Commit

Permalink
Use EmitterProcessor for Channels adaptation (#3100)
Browse files Browse the repository at this point in the history 
* Use `EmitterProcessor` for Channels adaptation

Related spring-cloud/spring-cloud-stream#1835

To honor a back-pressure after `MessageChannel` adaptation it is better
to use an `EmitterProcessor.create(1)` instead of `Flux.create()`.
This way whenever an emitter buffer is full, we block upstream producer
and don't allow it to produce more messages

**Cherry-pick to 5.1.x**

* * Wrap every new subscription into a `Flux.defer()`
* Fix `ReactiveStreamsConsumerTests` to use a new `Subscription` after
each `stop()/start()` on the `ReactiveStreamsConsumer`

* * Remove unused imports
  • Loading branch information
@artembilan @garyrussell
artembilan authored and garyrussell committed Nov 1, 2019
1 parent 69401c2 commit 36c9f72
Show file tree
Hide file tree
Showing 4 changed files with 90 additions and 110 deletions.
34 changes: 9 additions & 25 deletions ...re/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java
View file
Expand Up @@ -25,6 +25,7 @@
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;

import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.scheduler.Schedulers;
Expand Down Expand Up @@ -60,37 +61,20 @@ else if (messageChannel instanceof PollableChannel) {
}

private static <T> Publisher<Message<T>> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) {
return new SubscribableChannelPublisherAdapter<>(inputChannel);
return Flux.defer(() -> {
EmitterProcessor<Message<T>> publisher = EmitterProcessor.create(1);
@SuppressWarnings("unchecked")
MessageHandler messageHandler = (message) -> publisher.onNext((Message<T>) message);
inputChannel.subscribe(messageHandler);
return publisher
.doOnCancel(() -> inputChannel.unsubscribe(messageHandler));
});
}

private static <T> Publisher<Message<T>> adaptPollableChannelToPublisher(PollableChannel inputChannel) {
return new PollableChannelPublisherAdapter<>(inputChannel);
}


private static final class SubscribableChannelPublisherAdapter<T> implements Publisher<Message<T>> {

private final SubscribableChannel channel;

SubscribableChannelPublisherAdapter(SubscribableChannel channel) {
this.channel = channel;
}

@Override
@SuppressWarnings("unchecked")
public void subscribe(Subscriber<? super Message<T>> subscriber) {
Flux.
<Message<?>>create(emitter -> {
MessageHandler messageHandler = emitter::next;
this.channel.subscribe(messageHandler);
emitter.onCancel(() -> this.channel.unsubscribe(messageHandler));
},
FluxSink.OverflowStrategy.IGNORE)
.subscribe((Subscriber<? super Message<?>>) subscriber);
}

}

private static final class PollableChannelPublisherAdapter<T> implements Publisher<Message<T>> {

private final PollableChannel channel;
Expand Down
2 changes: 1 addition & 1 deletion ...-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
View file
Expand Up @@ -155,8 +155,8 @@ private final class DelegatingSubscriber extends BaseSubscriber<Message<?>> {

@Override
public void hookOnSubscribe(Subscription s) {
this.delegate.onSubscribe(s);
ReactiveStreamsConsumer.this.subscription = s;
this.delegate.onSubscribe(s);
}

@Override
Expand Down
39 changes: 21 additions & 18 deletions ...nnel/MessageChannelReactiveUtilsTest.java → ...nel/MessageChannelReactiveUtilsTests.java
View file
@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand All @@ -20,9 +20,8 @@

import java.time.Duration;

import org.junit.Test;
import org.junit.jupiter.api.Test;

import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;

import reactor.core.Disposable;
Expand All @@ -31,14 +30,19 @@
import reactor.test.StepVerifier;
import reactor.util.concurrent.Queues;

public class MessageChannelReactiveUtilsTest {
/**
* @author Sergei Egorov
* @author Artem Bilan
*
* @since 5.1.9
*/
class MessageChannelReactiveUtilsTests {

@Test
public void testBackpressureWithSubscribableChannel() {
void testBackpressureWithSubscribableChannel() {
Disposable.Composite compositeDisposable = Disposables.composite();
try {
DirectChannel channel = new DirectChannel();
assertThat(channel).isInstanceOf(SubscribableChannel.class);
int initialRequest = 10;
StepVerifier.create(MessageChannelReactiveUtils.toPublisher(channel), initialRequest)
.expectSubscription()
Expand All @@ -64,27 +68,25 @@ public void testBackpressureWithSubscribableChannel() {
}

@Test
public void testOverproducingWithSubscribableChannel() {
void testOverproducingWithSubscribableChannel() {
DirectChannel channel = new DirectChannel();
channel.setCountsEnabled(true);
assertThat(channel).isInstanceOf(SubscribableChannel.class);

Disposable.Composite compositeDisposable = Disposables.composite();
try {
int initialRequest = 10;
StepVerifier.create(MessageChannelReactiveUtils.toPublisher(channel), initialRequest)
.expectSubscription()
.then(() -> {
compositeDisposable.add(
Schedulers.boundedElastic().schedule(() -> {
while (true) {
if (channel.getSubscriberCount() > 0) {
channel.send(new GenericMessage<>("foo"));
.then(() ->
compositeDisposable.add(
Schedulers.boundedElastic().schedule(() -> {
while (true) {
if (channel.getSubscriberCount() > 0) {
channel.send(new GenericMessage<>("foo"));
}
}
}
})
);
})
})
))
.expectNextCount(initialRequest)
.thenAwait(Duration.ofMillis(100))
.thenCancel()
Expand All @@ -98,4 +100,5 @@ public void testOverproducingWithSubscribableChannel() {
.as("produced")
.isLessThanOrEqualTo(Queues.SMALL_BUFFER_SIZE);
}

}
125 changes: 59 additions & 66 deletions ...t/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java
View file
Expand Up @@ -17,12 +17,10 @@
package org.springframework.integration.channel.reactive;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

import java.util.LinkedList;
Expand All @@ -33,7 +31,6 @@
import java.util.concurrent.TimeUnit;

import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
Expand Down Expand Up @@ -82,14 +79,10 @@ public void testReactiveStreamsConsumerFluxMessageChannel() throws InterruptedEx

reactiveConsumer.stop();

try {
testChannel.send(testMessage);
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageDeliveryException.class);
assertThat(e.getCause()).isInstanceOf(IllegalStateException.class);
assertThat(e.getMessage()).contains("doesn't have subscribers to accept messages");
}
assertThatExceptionOfType(MessageDeliveryException.class)
.isThrownBy(() -> testChannel.send(testMessage))
.withCauseInstanceOf(IllegalStateException.class)
.withMessageContaining("doesn't have subscribers to accept messages");

reactiveConsumer.start();

Expand All @@ -102,54 +95,53 @@ public void testReactiveStreamsConsumerFluxMessageChannel() throws InterruptedEx


@Test
@SuppressWarnings("unchecked")
public void testReactiveStreamsConsumerDirectChannel() throws InterruptedException {
DirectChannel testChannel = new DirectChannel();

Subscriber<Message<?>> testSubscriber = (Subscriber<Message<?>>) Mockito.mock(Subscriber.class);

BlockingQueue<Message<?>> messages = new LinkedBlockingQueue<>();

willAnswer(i -> {
messages.put(i.getArgument(0));
return null;
})
.given(testSubscriber)
.onNext(any(Message.class));
Subscriber<Message<?>> testSubscriber = Mockito.spy(new Subscriber<Message<?>>() {

@Override
public void onSubscribe(Subscription subscription) {
subscription.request(1);
}

@Override
public void onNext(Message<?> message) {
messages.offer(message);
}

@Override
public void onError(Throwable t) {

}

@Override
public void onComplete() {

}

});

ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber);
reactiveConsumer.setBeanFactory(mock(BeanFactory.class));
reactiveConsumer.afterPropertiesSet();
reactiveConsumer.start();

Message<?> testMessage = new GenericMessage<>("test");
final Message<?> testMessage = new GenericMessage<>("test");
testChannel.send(testMessage);

ArgumentCaptor<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);
assertThat(message).isSameAs(testMessage);

reactiveConsumer.stop();

try {
testChannel.send(testMessage);
fail("MessageDeliveryException");
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageDeliveryException.class);
}
assertThatExceptionOfType(MessageDeliveryException.class)
.isThrownBy(() -> testChannel.send(testMessage));

reactiveConsumer.start();

subscription.request(1);

testMessage = new GenericMessage<>("test2");

testChannel.send(testMessage);

message = messages.poll(10, TimeUnit.SECONDS);
Expand All @@ -159,24 +151,40 @@ public void testReactiveStreamsConsumerDirectChannel() throws InterruptedExcepti
verify(testSubscriber, never()).onComplete();

assertThat(messages.isEmpty()).isTrue();

reactiveConsumer.stop();
}

@Test
@SuppressWarnings("unchecked")
public void testReactiveStreamsConsumerPollableChannel() 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.getArgument(0));
return null;
})
.given(testSubscriber)
.onNext(any(Message.class));
Subscriber<Message<?>> testSubscriber = Mockito.spy(new Subscriber<Message<?>>() {

@Override
public void onSubscribe(Subscription subscription) {
subscription.request(2);
}

@Override
public void onNext(Message<?> message) {
messages.offer(message);
}

@Override
public void onError(Throwable t) {

}

@Override
public void onComplete() {

}

});
ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber);
reactiveConsumer.setBeanFactory(mock(BeanFactory.class));
reactiveConsumer.afterPropertiesSet();
Expand All @@ -185,12 +193,6 @@ public void testReactiveStreamsConsumerPollableChannel() throws InterruptedExcep
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);
assertThat(message).isSameAs(testMessage);

Expand All @@ -201,11 +203,6 @@ public void testReactiveStreamsConsumerPollableChannel() throws InterruptedExcep

reactiveConsumer.start();

verify(testSubscriber, times(2)).onSubscribe(subscriptionArgumentCaptor.capture());
subscription = subscriptionArgumentCaptor.getValue();

subscription.request(2);

Message<?> testMessage2 = new GenericMessage<>("test2");

testChannel.send(testMessage2);
Expand Down Expand Up @@ -247,14 +244,10 @@ public void testReactiveStreamsConsumerViaConsumerEndpointFactoryBean() throws E

endpointFactoryBean.stop();

try {
testChannel.send(testMessage);
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageDeliveryException.class);
assertThat(e.getCause()).isInstanceOf(IllegalStateException.class);
assertThat(e.getMessage()).contains("doesn't have subscribers to accept messages");
}
assertThatExceptionOfType(MessageDeliveryException.class)
.isThrownBy(() -> testChannel.send(testMessage))
.withCauseInstanceOf(IllegalStateException.class)
.withMessageContaining("doesn't have subscribers to accept messages");

endpointFactoryBean.start();

Expand Down

0 comments on commit 36c9f72

Please sign in to comment.