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

Use EmitterProcessor in the FluxMessageChannel (#3104)

Browse Source 
* Use EmitterProcessor in the FluxMessageChannel

The `EmitterProcessor` has a good logic to block upstream producer
when its downstream subscriber cannot keep up with overproducing.

* Rework `FluxMessageChannel` logic to rely on the `EmitterProcessor`
instead of `Flux.create()`
* Cancel `FluxMessageChannel` internal subscriptions in the `destroy()`
* Fix `ReactiveStreamsTests.testFluxTransform()` for the splitter's
delimiter
* Ensure in the `FluxMessageChannelTests.testFluxMessageChannel`
that we can have several concurrent subscribers to the
`FluxMessageChannel`

* * Use `flux.onComplete()` instead of iteration over subscribers
* Change `subscribers` list into just `AtomicInteger` count marker
* fix `DefaultSplitterTests` according a new logic in the `FluxMessageChannel`

* GH-3107: Add errorOnTimeout for TcpInboundGateway

Fixes https://github.com/spring-projects/spring-integration/issues/3107

The `MessagingGatewaySupport` has an `errorOnTimeout` option to throw
a `MessageTimeoutException` when downstream reply doesn't come back in
time for configured reply timeout

* Expose an `errorOnTimeout` option as a `TcpInboundGateway` ctor
property
* Add new factory methods into a `Tcp` factory for Java DSL
* Ensure a property works as expected in the `IpIntegrationTests`
* Document a new option

* * Use `delaySubscription()` for subscribing publishers in the `FluxMessageChannel`
to wait until this one subscribed.
* Use an `EmitterProcessor` to catch subscriptions and pass them as a
signal to delayed upstream publishers
* Fix  `FluxMessageChannelTests.testFluxMessageChannelCleanUp` to
verify an actual property instead of removed.
* Fix `RSocketOutboundGatewayIntegrationTests` for the proper subscription
into a `FluxMessageChannel` before actual interaction with an RSocket
gateway.
This should help us also to avoid some race conditions in the future

* Revert "GH-3107: Add errorOnTimeout for TcpInboundGateway"

This reverts commit fa6119ddc4e423e8b15ca3fdc9fc4c79ea9d56af.

* * Refactor `FluxMessageChannel` to use `ReplayProcessor` for `subscribedSignal`.
This one is used `delaySubscription` for the upstream publishers
* Use a `AtomicBoolean` for subscription state since `doOnSubscribe()`
is called before `EmitterProcessor` adds subscribers for its `downstreams`
* Use `publishOn(Schedulers.boundedElastic())` for upstream publishers
to avoid blocking over there when our `EmitterProcessor` doesn't have
enough demand
* Refactor reactive tests to have a subscription into the `FluxMessageChannel`
earlier than emission happens for it

* * Use `Flux.subscribe(Consumer)` instead of `doOnNext(Consumer).subscribe()`

* * Emit `subscribedSignal` value after `.subscribe(subscriber)`
instead of `doOnSubscribe`
* Check for `this.processor.hasDownstreams()` before emitting such an event

* * Use `this.processor.hasDownstreams()` as a value to emit for `subscribedSignal`.
This way we are less vulnerable race conditions when subscribers are changed
actively
This commit is contained in:
Artem Bilan
2019-12-03 12:19:14 -05:00
committed by Gary Russell
parent 762f839602
commit 94e08169a5
2 changed files with 57 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
View File

@@ -16,20 +16,17 @@
package org.springframework.integration.channel;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.ReplayProcessor;
import reactor.core.scheduler.Schedulers;
/**
* The {@link AbstractMessageChannel} implementation for the
@@ -37,30 +34,27 @@ import reactor.core.publisher.FluxSink;
*
* @author Artem Bilan
* @author Gary Russell
* @author Sergei Egorov
*
* @since 5.0
*/
public class FluxMessageChannel extends AbstractMessageChannel
implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel {
private final List<Subscriber<? super Message<?>>> subscribers = new ArrayList<>();
private final EmitterProcessor<Message<?>> processor;
private final Map<Publisher<? extends Message<?>>, ConnectableFlux<?>> publishers = new ConcurrentHashMap<>();
private final FluxSink<Message<?>> sink;
private final Flux<Message<?>> flux;
private FluxSink<Message<?>> sink;
private final ReplayProcessor<Boolean> subscribedSignal = ReplayProcessor.create(1);
public FluxMessageChannel() {
this.flux =
Flux.<Message<?>>create(emitter -> this.sink = emitter, FluxSink.OverflowStrategy.IGNORE)
.publish()
.autoConnect();
this.processor = EmitterProcessor.create(1, false);
this.sink = this.processor.sink(FluxSink.OverflowStrategy.BUFFER);
}
@Override
protected boolean doSend(Message<?> message, long timeout) {
Assert.state(this.subscribers.size() > 0,
Assert.state(this.processor.hasDownstreams(),
() -> "The [" + this + "] doesn't have subscribers to accept messages");
this.sink.next(message);
return true;
@@ -68,30 +62,33 @@ public class FluxMessageChannel extends AbstractMessageChannel
@Override
public void subscribe(Subscriber<? super Message<?>> subscriber) {
this.subscribers.add(subscriber);
this.flux.doOnCancel(() -> this.subscribers.remove(subscriber))
.retry()
this.processor
.doFinally((s) -> this.subscribedSignal.onNext(this.processor.hasDownstreams()))
.subscribe(subscriber);
this.publishers.values().forEach(ConnectableFlux::connect);
this.subscribedSignal.onNext(this.processor.hasDownstreams());
}
@Override
public void subscribeTo(Publisher<? extends Message<?>> publisher) {
ConnectableFlux<?> connectableFlux =
Flux.from(publisher)
.handle((message, sink) -> sink.next(send(message)))
.onErrorContinue((throwable, event) ->
logger.warn("Error during processing event: " + event, throwable))
.doOnComplete(() -> this.publishers.remove(publisher))
.publish();
Flux.from(publisher)
.delaySubscription(this.subscribedSignal.filter(Boolean::booleanValue).next())
.publishOn(Schedulers.boundedElastic())
.doOnNext((message) -> {
try {
send(message);
}
catch (Exception e) {
logger.warn("Error during processing event: " + message, e);
}
})
.subscribe();
}
this.publishers.put(publisher, connectableFlux);
if (!this.subscribers.isEmpty()) {
connectableFlux.connect();
}
@Override
public void destroy() {
this.subscribedSignal.onNext(false);
this.processor.onComplete();
super.destroy();
}
}

42
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java
View File

@@ -20,17 +20,17 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.BridgeFrom;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.MessageChannelReactiveUtils;
@@ -47,8 +47,10 @@ import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import reactor.core.Disposable;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
/**
@@ -56,7 +58,7 @@ import reactor.core.publisher.Flux;
*
* @since 5.0
*/
@RunWith(SpringRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class FluxMessageChannelTests {
@@ -64,7 +66,7 @@ public class FluxMessageChannelTests {
private MessageChannel fluxMessageChannel;
@Autowired
private MessageChannel queueChannel;
private QueueChannel queueChannel;
@Autowired
private PollableChannel errorChannel;
@@ -73,7 +75,7 @@ public class FluxMessageChannelTests {
private IntegrationFlowContext integrationFlowContext;
@Test
public void testFluxMessageChannel() {
void testFluxMessageChannel() {
QueueChannel replyChannel = new QueueChannel();
for (int i = 0; i < 10; i++) {
@@ -90,28 +92,35 @@ public class FluxMessageChannelTests {
Message<?> error = this.errorChannel.receive(0);
assertThat(error).isNotNull();
assertThat(((MessagingException) error.getPayload()).getFailedMessage().getPayload()).isEqualTo(5);
List<Message<?>> messages = this.queueChannel.clear();
assertThat(messages).extracting((message) -> (Integer) message.getPayload())
.containsAll(IntStream.range(0, 10).boxed().collect(Collectors.toList()));
}
@Test
public void testMessageChannelReactiveAdaptation() throws InterruptedException {
void testMessageChannelReactiveAdaptation() throws InterruptedException {
CountDownLatch done = new CountDownLatch(2);
List<String> results = new ArrayList<>();
Flux.from(MessageChannelReactiveUtils.<String>toPublisher(this.queueChannel))
.map(Message::getPayload)
.map(String::toUpperCase)
.doOnNext(results::add)
.subscribe(v -> done.countDown());
Disposable disposable =
Flux.from(MessageChannelReactiveUtils.<String>toPublisher(this.queueChannel))
.map(Message::getPayload)
.map(String::toUpperCase)
.doOnNext(results::add)
.subscribe(v -> done.countDown());
this.queueChannel.send(new GenericMessage<>("foo"));
this.queueChannel.send(new GenericMessage<>("bar"));
assertThat(done.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(results).containsExactly("FOO", "BAR");
disposable.dispose();
}
@Test
public void testFluxMessageChannelCleanUp() throws InterruptedException {
void testFluxMessageChannelCleanUp() throws InterruptedException {
FluxMessageChannel flux = MessageChannels.flux().get();
CountDownLatch finishLatch = new CountDownLatch(1);
@@ -130,9 +139,9 @@ public class FluxMessageChannelTests {
assertThat(finishLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(TestUtils.getPropertyValue(flux, "publishers", Map.class).isEmpty()).isTrue();
flowRegistration.destroy();
assertThat(TestUtils.getPropertyValue(flux, "processor", EmitterProcessor.class).isTerminated()).isTrue();
}
@Configuration
@@ -158,6 +167,7 @@ public class FluxMessageChannelTests {
}
@Bean
@BridgeFrom("fluxMessageChannel")
public MessageChannel queueChannel() {
return new QueueChannel();
}