2.x: add onTerminateDetach to Single and Completable (#5624)

* 2.x: add onTerminateDetach to Single and Completable

* Improve coverage, fix lack of nulling out the downstream

Author: akarnokd

src/main/java/io/reactivex/Completable.java

@@ -1380,6 +1380,24 @@ public final Completable onErrorResumeNext(final Function<? super Throwable, ? e
         return RxJavaPlugins.onAssembly(new CompletableResumeNext(this, errorMapper));
     }
 
+    /**
+     * Nulls out references to the upstream producer and downstream CompletableObserver if
+     * the sequence is terminated or downstream calls dispose().
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code onTerminateDetach} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @return a Completable which nulls out references to the upstream producer and downstream CompletableObserver if
+     * the sequence is terminated or downstream calls dispose()
+     * @since 2.1.5 - experimental
+     */
+    @Experimental
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final Completable onTerminateDetach() {
+        return RxJavaPlugins.onAssembly(new CompletableDetach(this));
+    }
+
     /**
      * Returns a Completable that repeatedly subscribes to this Completable until cancelled.
      * <dl>

src/main/java/io/reactivex/Maybe.java

@@ -3316,14 +3316,15 @@ public final Maybe<T> onExceptionResumeNext(final MaybeSource<? extends T> next)
         ObjectHelper.requireNonNull(next, "next is null");
         return RxJavaPlugins.onAssembly(new MaybeOnErrorNext<T>(this, Functions.justFunction(next), false));
     }
+
     /**
      * Nulls out references to the upstream producer and downstream MaybeObserver if
      * the sequence is terminated or downstream calls dispose().
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code onTerminateDetach} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
-     * @return a Maybe which out references to the upstream producer and downstream MaybeObserver if
+     * @return a Maybe which nulls out references to the upstream producer and downstream MaybeObserver if
      * the sequence is terminated or downstream calls dispose()
      */
     @CheckReturnValue

src/main/java/io/reactivex/Single.java

@@ -2468,6 +2468,24 @@ public final Single<T> onErrorResumeNext(
         return RxJavaPlugins.onAssembly(new SingleResumeNext<T>(this, resumeFunctionInCaseOfError));
     }
 
+    /**
+     * Nulls out references to the upstream producer and downstream SingleObserver if
+     * the sequence is terminated or downstream calls dispose().
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code onTerminateDetach} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @return a Single which nulls out references to the upstream producer and downstream SingleObserver if
+     * the sequence is terminated or downstream calls dispose()
+     * @since 2.1.5 - experimental
+     */
+    @Experimental
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final Single<T> onTerminateDetach() {
+        return RxJavaPlugins.onAssembly(new SingleDetach<T>(this));
+    }
+
     /**
      * Repeatedly re-subscribes to the current Single and emits each success value.
      * <dl>

src/main/java/io/reactivex/internal/operators/completable/CompletableDetach.java

@@ -0,0 +1,91 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.operators.completable;
+
+import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+
+/**
+ * Breaks the references between the upstream and downstream when the Completable terminates.
+ * 
+ * @since 2.1.5 - experimental
+ */
+@Experimental
+public final class CompletableDetach extends Completable {
+
+    final CompletableSource source;
+
+    public CompletableDetach(CompletableSource source) {
+        this.source = source;
+    }
+
+    @Override
+    protected void subscribeActual(CompletableObserver observer) {
+        source.subscribe(new DetachCompletableObserver(observer));
+    }
+
+    static final class DetachCompletableObserver implements CompletableObserver, Disposable {
+
+        CompletableObserver actual;
+
+        Disposable d;
+
+        DetachCompletableObserver(CompletableObserver actual) {
+            this.actual = actual;
+        }
+
+        @Override
+        public void dispose() {
+            actual = null;
+            d.dispose();
+            d = DisposableHelper.DISPOSED;
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return d.isDisposed();
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.validate(this.d, d)) {
+                this.d = d;
+
+                actual.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            d = DisposableHelper.DISPOSED;
+            CompletableObserver a = actual;
+            if (a != null) {
+                actual = null;
+                a.onError(e);
+            }
+        }
+
+        @Override
+        public void onComplete() {
+            d = DisposableHelper.DISPOSED;
+            CompletableObserver a = actual;
+            if (a != null) {
+                actual = null;
+                a.onComplete();
+            }
+        }
+    }
+}

src/main/java/io/reactivex/internal/operators/maybe/MaybeDetach.java

@@ -69,6 +69,7 @@ public void onSuccess(T value) {
             d = DisposableHelper.DISPOSED;
             MaybeObserver<? super T> a = actual;
             if (a != null) {
+                actual = null;
                 a.onSuccess(value);
             }
         }
@@ -78,6 +79,7 @@ public void onError(Throwable e) {
             d = DisposableHelper.DISPOSED;
             MaybeObserver<? super T> a = actual;
             if (a != null) {
+                actual = null;
                 a.onError(e);
             }
         }
@@ -87,6 +89,7 @@ public void onComplete() {
             d = DisposableHelper.DISPOSED;
             MaybeObserver<? super T> a = actual;
             if (a != null) {
+                actual = null;
                 a.onComplete();
             }
         }

src/main/java/io/reactivex/internal/operators/single/SingleDetach.java

@@ -0,0 +1,92 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.operators.single;
+
+import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+
+/**
+ * Breaks the references between the upstream and downstream when the Maybe terminates.
+ *
+ * @param <T> the value type
+ * @since 2.1.5 - experimental
+ */
+@Experimental
+public final class SingleDetach<T> extends Single<T> {
+
+    final SingleSource<T> source;
+
+    public SingleDetach(SingleSource<T> source) {
+        this.source = source;
+    }
+
+    @Override
+    protected void subscribeActual(SingleObserver<? super T> observer) {
+        source.subscribe(new DetachSingleObserver<T>(observer));
+    }
+
+    static final class DetachSingleObserver<T> implements SingleObserver<T>, Disposable {
+
+        SingleObserver<? super T> actual;
+
+        Disposable d;
+
+        DetachSingleObserver(SingleObserver<? super T> actual) {
+            this.actual = actual;
+        }
+
+        @Override
+        public void dispose() {
+            actual = null;
+            d.dispose();
+            d = DisposableHelper.DISPOSED;
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return d.isDisposed();
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.validate(this.d, d)) {
+                this.d = d;
+
+                actual.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onSuccess(T value) {
+            d = DisposableHelper.DISPOSED;
+            SingleObserver<? super T> a = actual;
+            if (a != null) {
+                actual = null;
+                a.onSuccess(value);
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            d = DisposableHelper.DISPOSED;
+            SingleObserver<? super T> a = actual;
+            if (a != null) {
+                actual = null;
+                a.onError(e);
+            }
+        }
+    }
+}