BeforeFinallyHttpOperator dereference subscriber after termination (#…

…2327)

Motivation:
After terminating the Single subscriber we need to dereference the Single subscriber as otherwise
we may retain a reference to a downstream Subscriber and violate RS 3.13.
https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.4/README.md#3.13

servicetalk-http-utils/src/main/java/io/servicetalk/http/utils/BeforeFinallyHttpOperator.java

@@ -117,8 +117,22 @@ private static final class ResponseCompletionSubscriber implements SingleSource.
         private static final int TERMINATED = 4;
         private static final AtomicIntegerFieldUpdater<ResponseCompletionSubscriber> stateUpdater =
                 newUpdater(ResponseCompletionSubscriber.class, "state");
+        private static final SingleSource.Subscriber<StreamingHttpResponse> NOOP_SUBSCRIBER =
+                new SingleSource.Subscriber<StreamingHttpResponse>() {
+            @Override
+            public void onSubscribe(final Cancellable cancellable) {
+            }
+
+            @Override
+            public void onSuccess(@Nullable final StreamingHttpResponse result) {
+            }
 
-        private final SingleSource.Subscriber<? super StreamingHttpResponse> subscriber;
+            @Override
+            public void onError(final Throwable t) {
+            }
+        };
+
+        private SingleSource.Subscriber<? super StreamingHttpResponse> subscriber;
         private final TerminalSignalConsumer beforeFinally;
         private final boolean discardEventsAfterCancel;
         private volatile int state;
@@ -388,6 +402,7 @@ private void cancel0(final boolean propagateCancel) {
                     return Publisher.failed(new CancellationException("Received response post cancel."));
                 }));
             }
+            dereferenceSubscriber();
         }
 
         @Override
@@ -402,6 +417,7 @@ public void onError(final Throwable t) {
                 addSuppressed(t, cause);
             }
             subscriber.onError(t);
+            dereferenceSubscriber();
         }
 
         private void sendNullResponse() {
@@ -414,9 +430,18 @@ private void sendNullResponse() {
                 }
             } catch (Throwable cause) {
                 subscriber.onError(cause);
+                dereferenceSubscriber();
                 return;
             }
             subscriber.onSuccess(null);
+            dereferenceSubscriber();
+        }
+
+        private void dereferenceSubscriber() {
+            // After terminating the Single subscriber we need to dereference the Single subscriber as otherwise
+            // we may retain a reference to a downstream Subscriber and violate RS 3.13.
+            // https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.4/README.md#3.13
+            subscriber = NOOP_SUBSCRIBER;
         }
     }
 }