Merge branch '0.9.x'

reactor · Jun 29, 2020 · 10c7f70 · 10c7f70
2 parents 02eaf62 + 49387fa
commit 10c7f70
Showing 1 changed file with 10 additions and 8 deletions.
diff --git a/src/main/java/reactor/netty/http/client/HttpClientOperations.java b/src/main/java/reactor/netty/http/client/HttpClientOperations.java
@@ -63,6 +63,7 @@
 import io.netty.handler.stream.ChunkedWriteHandler;
 import io.netty.util.ReferenceCountUtil;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
 import reactor.core.Exceptions;
@@ -765,6 +766,13 @@ static final class SendForm extends Mono<Void> {
 
 		@Override
 		public void subscribe(CoreSubscriber<? super Void> s) {
+			if (!parent.markSentHeaders()) {
+				Operators.error(s,
+						new IllegalStateException("headers have already been sent"));
+				return;
+			}
+			Subscription subscription = Operators.emptySubscription();
+			s.onSubscribe(subscription);
 			if (parent.channel()
 			          .eventLoop()
 			          .inEventLoop()) {
@@ -779,12 +787,6 @@ public void subscribe(CoreSubscriber<? super Void> s) {
 
 		@SuppressWarnings("FutureReturnValueIgnored")
 		void _subscribe(CoreSubscriber<? super Void> s) {
-			if (!parent.markSentHeaders()) {
-				Operators.error(s,
-						new IllegalStateException("headers have already been sent"));
-				return;
-			}
-
 			HttpDataFactory df = DEFAULT_FACTORY;
 
 			try {
@@ -841,14 +843,14 @@ void _subscribe(CoreSubscriber<? super Void> s) {
 						                                  .flux());
 					}
 				}
-				Operators.complete(s);
+				s.onComplete();
 
 
 			}
 			catch (Throwable e) {
 				Exceptions.throwIfJvmFatal(e);
 				df.cleanRequestHttpData(parent.nettyRequest);
-				Operators.error(s, Exceptions.unwrap(e));
+				s.onError(Exceptions.unwrap(e));
 			}
 		}
 	}