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

Fix ReactiveStreamsConsumer for PollableChannel

Browse Source 
https://build.spring.io/browse/INT-MASTERSPRING40-985/

* Move `onSubscribe()` callback to the `subscribe()` operator
to honor a `request` contract from the target subscriber
This commit is contained in:
Artem Bilan
2020-02-18 12:37:24 -05:00
parent 6d936d9774
commit d61a247f4b
1 changed files with 10 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -172,16 +172,17 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
else if (this.subscriber != null) {
this.subscription =
Flux.from(this.publisher)
.doOnComplete(this.subscriber::onComplete)
.doOnSubscribe(this.subscriber::onSubscribe)
.subscribe((data) -> {
try {
this.subscriber.onNext(data);
}
catch (Exception ex) {
this.errorHandler.handleError(ex);
}
});
try {
this.subscriber.onNext(data);
}
catch (Exception ex) {
this.errorHandler.handleError(ex);
}
},
null,
this.subscriber::onComplete,
this.subscriber::onSubscribe);
}
}