Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use EmitterProcessor for Channels adaptation (#3100)

Browse Source 
* Use `EmitterProcessor` for Channels adaptation

Related https://github.com/spring-cloud/spring-cloud-stream/issues/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
This commit is contained in:
Artem Bilan
2019-11-01 14:49:29 -04:00
committed by Gary Russell
parent 69401c263c
commit 36c9f72a9d
4 changed files with 90 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java
View File

@@ -25,6 +25,7 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.scheduler.Schedulers;
@@ -60,37 +61,20 @@ public final class MessageChannelReactiveUtils {
}
private static <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;

2
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
View File

@@ -155,8 +155,8 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
@Override
public void hookOnSubscribe(Subscription s) {
this.delegate.onSubscribe(s);
ReactiveStreamsConsumer.this.subscription = s;
this.delegate.onSubscribe(s);
}
@Override

39
spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTest.java → spring-integration-core/src/test/java/org/springframework/integration/channel/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.
@@ -20,9 +20,8 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.time.Duration;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
import reactor.core.Disposable;
@@ -31,14 +30,19 @@ import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
import reactor.util.concurrent.Queues;
public class MessageChannelReactiveUtilsTest {
/**
* @author Sergei Egorov
* @author Artem Bilan
*
* @since 5.1.9
*/
class MessageChannelReactiveUtilsTests {
@Test
public void testBackpressureWithSubscribableChannel() {
void testBackpressureWithSubscribableChannel() {
Disposable.Composite compositeDisposable = Disposables.composite();
try {
DirectChannel channel = new DirectChannel();
assertThat(channel).isInstanceOf(SubscribableChannel.class);
int initialRequest = 10;
StepVerifier.create(MessageChannelReactiveUtils.toPublisher(channel), initialRequest)
.expectSubscription()
@@ -64,27 +68,25 @@ public class MessageChannelReactiveUtilsTest {
}
@Test
public void testOverproducingWithSubscribableChannel() {
void testOverproducingWithSubscribableChannel() {
DirectChannel channel = new DirectChannel();
channel.setCountsEnabled(true);
assertThat(channel).isInstanceOf(SubscribableChannel.class);
Disposable.Composite compositeDisposable = Disposables.composite();
try {
int initialRequest = 10;
StepVerifier.create(MessageChannelReactiveUtils.toPublisher(channel), initialRequest)
.expectSubscription()
.then(() -> {
compositeDisposable.add(
Schedulers.boundedElastic().schedule(() -> {
while (true) {
if (channel.getSubscriberCount() > 0) {
channel.send(new GenericMessage<>("foo"));
.then(() ->
compositeDisposable.add(
Schedulers.boundedElastic().schedule(() -> {
while (true) {
if (channel.getSubscriberCount() > 0) {
channel.send(new GenericMessage<>("foo"));
}
}
}
})
);
})
})
))
.expectNextCount(initialRequest)
.thenAwait(Duration.ofMillis(100))
.thenCancel()
@@ -98,4 +100,5 @@ public class MessageChannelReactiveUtilsTest {
.as("produced")
.isLessThanOrEqualTo(Queues.SMALL_BUFFER_SIZE);
}
}

125
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java
View File

@@ -17,12 +17,10 @@
package org.springframework.integration.channel.reactive;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.util.LinkedList;
@@ -33,7 +31,6 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
@@ -82,14 +79,10 @@ public class ReactiveStreamsConsumerTests {
reactiveConsumer.stop();
try {
testChannel.send(testMessage);
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageDeliveryException.class);
assertThat(e.getCause()).isInstanceOf(IllegalStateException.class);
assertThat(e.getMessage()).contains("doesn't have subscribers to accept messages");
}
assertThatExceptionOfType(MessageDeliveryException.class)
.isThrownBy(() -> testChannel.send(testMessage))
.withCauseInstanceOf(IllegalStateException.class)
.withMessageContaining("doesn't have subscribers to accept messages");
reactiveConsumer.start();
@@ -102,54 +95,53 @@ public class ReactiveStreamsConsumerTests {
@Test
@SuppressWarnings("unchecked")
public void testReactiveStreamsConsumerDirectChannel() throws InterruptedException {
DirectChannel testChannel = new DirectChannel();
Subscriber<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);
@@ -159,6 +151,8 @@ public class ReactiveStreamsConsumerTests {
verify(testSubscriber, never()).onComplete();
assertThat(messages.isEmpty()).isTrue();
reactiveConsumer.stop();
}
@Test
@@ -166,17 +160,31 @@ public class ReactiveStreamsConsumerTests {
public void testReactiveStreamsConsumerPollableChannel() throws InterruptedException {
QueueChannel testChannel = new QueueChannel();
Subscriber<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();
@@ -185,12 +193,6 @@ public class ReactiveStreamsConsumerTests {
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);
@@ -201,11 +203,6 @@ public class ReactiveStreamsConsumerTests {
reactiveConsumer.start();
verify(testSubscriber, times(2)).onSubscribe(subscriptionArgumentCaptor.capture());
subscription = subscriptionArgumentCaptor.getValue();
subscription.request(2);
Message<?> testMessage2 = new GenericMessage<>("test2");
testChannel.send(testMessage2);
@@ -247,14 +244,10 @@ public class ReactiveStreamsConsumerTests {
endpointFactoryBean.stop();
try {
testChannel.send(testMessage);
}
catch (Exception e) {
assertThat(e).isInstanceOf(MessageDeliveryException.class);
assertThat(e.getCause()).isInstanceOf(IllegalStateException.class);
assertThat(e.getMessage()).contains("doesn't have subscribers to accept messages");
}
assertThatExceptionOfType(MessageDeliveryException.class)
.isThrownBy(() -> testChannel.send(testMessage))
.withCauseInstanceOf(IllegalStateException.class)
.withMessageContaining("doesn't have subscribers to accept messages");
endpointFactoryBean.start();