Fixes HttpClientOperations to don't offload `onSubscribe(subscription…

…)` (#1189)

Signed-off-by: Oleh Dokuka <shadowgun@i.ua>

src/main/java/reactor/netty/http/client/HttpClientOperations.java

@@ -64,6 +64,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;
@@ -761,6 +762,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()) {
@@ -775,12 +783,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 {
@@ -837,14 +839,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));
 			}
 		}
 	}