From d61a247f4bde5545e8dc1bbfc658f18a9075faeb Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Tue, 18 Feb 2020 12:37:24 -0500
Subject: [PATCH] Fix ReactiveStreamsConsumer for PollableChannel

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

* Move `onSubscribe()` callback to the `subscribe()` operator
to honor a `request` contract from the target subscriber
---
 .../endpoint/ReactiveStreamsConsumer.java     | 19 ++++++++++---------
 1 file changed, 10 insertions(+), 9 deletions(-)

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
index 5a791ec654..1c81fe2d5c 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
@@ -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);
 		}
 	}