Fix forwarding null when CancellableMonoSink is cancelled (#11221)

Might fix #11209, will see.

http-client/src/main/java/io/micronaut/http/client/netty/CancellableMonoSink.java

@@ -44,6 +44,7 @@ final class CancellableMonoSink<T> implements Publisher<T>, Sinks.One<T>, Subscr
     private T value;
     private Throwable failure;
     private boolean complete = false;
+    private boolean cancelled = false;
     private Subscriber<? super T> subscriber = null;
     private boolean subscriberWaiting = false;
 
@@ -72,7 +73,7 @@ public void subscribe(Subscriber<? super T> s) {
     }
 
     private void tryForward() {
-        if (subscriberWaiting && complete) {
+        if (subscriberWaiting && complete && !cancelled) {
             if (failure == null) {
                 if (value != EMPTY) {
                     subscriber.onNext(value);
@@ -181,6 +182,7 @@ public void cancel() {
         lock.lock();
         try {
             complete = true;
+            cancelled = true;
         } finally {
             lock.unlock();
         }

http-client/src/test/groovy/io/micronaut/http/client/netty/CancellableMonoSinkSpec.groovy

@@ -0,0 +1,41 @@
+package io.micronaut.http.client.netty
+
+import org.reactivestreams.Subscriber
+import org.reactivestreams.Subscription
+import reactor.core.publisher.Mono
+import spock.lang.Specification
+
+class CancellableMonoSinkSpec extends Specification {
+    def "cancel before request"() {
+        given:
+        def sink = new CancellableMonoSink<String>(null)
+        def result = "unset"
+        Subscription subscription = null
+        sink.subscribe(new Subscriber<String>() {
+            @Override
+            void onSubscribe(Subscription s) {
+                subscription = s
+            }
+
+            @Override
+            void onNext(String s) {
+                result = s
+            }
+
+            @Override
+            void onError(Throwable t) {
+            }
+
+            @Override
+            void onComplete() {
+            }
+        })
+
+        when:
+        sink.cancel()
+        subscription.request(1)
+
+        then:
+        result == "unset"
+    }
+}