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

Upgrade dependencies; prepare for release

Browse Source 
* Fix MongoDB tests for changed return type from Spring Data MongoDB
This commit is contained in:
Artem Bilan
2019-12-04 13:06:04 -05:00
parent d5e3148618
commit aa895e7365
4 changed files with 77 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -22,6 +22,8 @@ import org.reactivestreams.Subscriber;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
@@ -47,6 +49,8 @@ public class FluxMessageChannel extends AbstractMessageChannel
private final ReplayProcessor<Boolean> subscribedSignal = ReplayProcessor.create(1);
private final Disposable.Composite upstreamSubscriptions = Disposables.composite();
public FluxMessageChannel() {
this.processor = EmitterProcessor.create(1, false);
this.sink = this.processor.sink(FluxSink.OverflowStrategy.BUFFER);
@@ -70,23 +74,25 @@ public class FluxMessageChannel extends AbstractMessageChannel
@Override
public void subscribeTo(Publisher<? extends Message<?>> publisher) {
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.upstreamSubscriptions.add(
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());
}
@Override
public void destroy() {
this.subscribedSignal.onNext(false);
this.upstreamSubscriptions.dispose();
this.processor.onComplete();
super.destroy();
}