1.x: proposal: onTerminateDetach - detach upstream/downstream for GC (#3888)

akarnokd · akarnokd · commit 4e5c17c6186d · 2016-05-03T22:12:02.000+02:00
diff --git a/src/main/java/rx/Observable.java b/src/main/java/rx/Observable.java
@@ -6611,6 +6611,23 @@ public final Observable<T> onExceptionResumeNext(final Observable<? extends T> r
         return lift((Operator<T, T>)OperatorOnErrorResumeNextViaFunction.withException(resumeSequence));
     }
 
+    
+    /**
+     * Nulls out references to the upstream producer and downstream Subscriber if
+     * the sequence is terminated or downstream unsubscribes.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code onTerminateDetach} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @return an Observable which out references to the upstream producer and downstream Subscriber if
+     * the sequence is terminated or downstream unsubscribes
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Experimental
+    public final Observable<T> onTerminateDetach() {
+        return create(new OnSubscribeDetach<T>(this));
+    }
+    
     /**
      * Returns a {@link ConnectableObservable}, which is a variety of Observable that waits until its
      * {@link ConnectableObservable#connect connect} method is called before it begins emitting items to those
diff --git a/src/main/java/rx/internal/operators/OnSubscribeDetach.java b/src/main/java/rx/internal/operators/OnSubscribeDetach.java
@@ -0,0 +1,173 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * 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 rx.internal.operators;
+
+import java.util.concurrent.atomic.*;
+
+import rx.*;
+import rx.Observable.OnSubscribe;
+import rx.internal.util.RxJavaPluginUtils;
+
+/**
+ * Nulls out references to upstream data structures when the source terminates or 
+ * the child unsubscribes.
+ * @param <T> the value type
+ */
+public final class OnSubscribeDetach<T> implements OnSubscribe<T> {
+    
+    final Observable<T> source;
+    
+    public OnSubscribeDetach(Observable<T> source) {
+        this.source = source;
+    }
+    
+    @Override
+    public void call(Subscriber<? super T> t) {
+        DetachSubscriber<T> parent = new DetachSubscriber<T>(t);
+        DetachProducer<T> producer = new DetachProducer<T>(parent);
+        
+        t.add(producer);
+        t.setProducer(producer);
+        
+        source.unsafeSubscribe(parent);
+    }
+    
+    /**
+     * The parent subscriber that forwards events and cleans up on a terminal state.
+     * @param <T> the value type
+     */
+    static final class DetachSubscriber<T> extends Subscriber<T> {
+        
+        final AtomicReference<Subscriber<? super T>> actual;
+        
+        final AtomicReference<Producer> producer;
+        
+        final AtomicLong requested;
+
+        public DetachSubscriber(Subscriber<? super T> actual) {
+            this.actual = new AtomicReference<Subscriber<? super T>>(actual);
+            this.producer = new AtomicReference<Producer>();
+            this.requested = new AtomicLong();
+        }
+        
+        @Override
+        public void onNext(T t) {
+            Subscriber<? super T> a = actual.get();
+            
+            if (a != null) {
+                a.onNext(t);
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            producer.lazySet(TerminatedProducer.INSTANCE);
+            Subscriber<? super T> a = actual.getAndSet(null);
+            
+            if (a != null) {
+                a.onError(e);
+            } else {
+                RxJavaPluginUtils.handleException(e);
+            }
+        }
+
+        
+        @Override
+        public void onCompleted() {
+            producer.lazySet(TerminatedProducer.INSTANCE);
+            Subscriber<? super T> a = actual.getAndSet(null);
+            
+            if (a != null) {
+                a.onCompleted();
+            }
+        }
+        
+        void innerRequest(long n) {
+            if (n < 0L) {
+                throw new IllegalArgumentException("n >= 0 required but it was " + n);
+            }
+            Producer p = producer.get();
+            if (p != null) {
+                p.request(n);
+            } else {
+                BackpressureUtils.getAndAddRequest(requested, n);
+                p = producer.get();
+                if (p != null && p != TerminatedProducer.INSTANCE) {
+                    long r = requested.getAndSet(0L);
+                    p.request(r);
+                }
+            }
+        }
+        
+        @Override
+        public void setProducer(Producer p) {
+            if (producer.compareAndSet(null, p)) {
+                long r = requested.getAndSet(0L);
+                p.request(r);
+            } else {
+                if (producer.get() != TerminatedProducer.INSTANCE) {
+                    throw new IllegalStateException("Producer already set!");
+                }
+            }
+        }
+        
+        void innerUnsubscribe() {
+            producer.lazySet(TerminatedProducer.INSTANCE);
+            actual.lazySet(null);
+            // full barrier in unsubscribe()
+            unsubscribe();
+        }
+    }
+    
+    /**
+     * Callbacks from the child Subscriber.
+     * @param <T> the value type
+     */
+    static final class DetachProducer<T> implements Producer, Subscription {
+        final DetachSubscriber<T> parent;
+
+        public DetachProducer(DetachSubscriber<T> parent) {
+            this.parent = parent;
+        }
+        
+        @Override
+        public void request(long n) {
+            parent.innerRequest(n);
+        }
+        
+        @Override
+        public boolean isUnsubscribed() {
+            return parent.isUnsubscribed();
+        }
+        
+        @Override
+        public void unsubscribe() {
+            parent.innerUnsubscribe();
+        }
+    }
+    
+    /**
+     * Singleton instance via enum.
+     */
+    enum TerminatedProducer implements Producer {
+        INSTANCE;
+        
+        @Override
+        public void request(long n) {
+            // ignored
+        }
+    }
+}
diff --git a/src/test/java/rx/internal/operators/OnSubscribeDetachTest.java b/src/test/java/rx/internal/operators/OnSubscribeDetachTest.java
@@ -0,0 +1,160 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * 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 rx.internal.operators;
+
+import java.lang.ref.WeakReference;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.*;
+
+import rx.*;
+import rx.Observable.OnSubscribe;
+import rx.exceptions.TestException;
+import rx.observers.TestSubscriber;
+
+public class OnSubscribeDetachTest {
+
+    Object o;
+    
+    @Test
+    public void just() throws Exception {
+        o = new Object();
+        
+        WeakReference<Object> wr = new WeakReference<Object>(o);
+        
+        TestSubscriber<Object> ts = new TestSubscriber<Object>();
+        
+        Observable.just(o).count().onTerminateDetach().subscribe(ts);
+        
+        ts.assertValue(1);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+        
+        o = null;
+        
+        System.gc();
+        Thread.sleep(200);
+        
+        Assert.assertNull("Object retained!", wr.get());
+        
+    }
+    
+    @Test
+    public void error() {
+        TestSubscriber<Object> ts = new TestSubscriber<Object>();
+        
+        Observable.error(new TestException()).onTerminateDetach().subscribe(ts);
+        
+        ts.assertNoValues();
+        ts.assertError(TestException.class);
+        ts.assertNotCompleted();
+    }
+    
+    @Test
+    public void empty() {
+        TestSubscriber<Object> ts = new TestSubscriber<Object>();
+        
+        Observable.empty().onTerminateDetach().subscribe(ts);
+        
+        ts.assertNoValues();
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
+
+    @Test
+    public void range() {
+        TestSubscriber<Object> ts = new TestSubscriber<Object>();
+        
+        Observable.range(1, 1000).onTerminateDetach().subscribe(ts);
+        
+        ts.assertValueCount(1000);
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
+
+    
+    @Test
+    public void backpressured() throws Exception {
+        o = new Object();
+        
+        WeakReference<Object> wr = new WeakReference<Object>(o);
+        
+        TestSubscriber<Object> ts = new TestSubscriber<Object>(0L);
+        
+        Observable.just(o).count().onTerminateDetach().subscribe(ts);
+
+        ts.assertNoValues();
+
+        ts.requestMore(1);
+        
+        ts.assertValue(1);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+        
+        o = null;
+        
+        System.gc();
+        Thread.sleep(200);
+        
+        Assert.assertNull("Object retained!", wr.get());
+    }
+
+    @Test
+    public void justUnsubscribed() throws Exception {
+        o = new Object();
+        
+        WeakReference<Object> wr = new WeakReference<Object>(o);
+        
+        TestSubscriber<Object> ts = new TestSubscriber<Object>(0);
+        
+        Observable.just(o).count().onTerminateDetach().subscribe(ts);
+        
+        ts.unsubscribe();
+        o = null;
+        
+        System.gc();
+        Thread.sleep(200);
+        
+        Assert.assertNull("Object retained!", wr.get());
+        
+    }
+
+    @Test
+    public void deferredUpstreamProducer() {
+        final AtomicReference<Subscriber<? super Object>> subscriber = new AtomicReference<Subscriber<? super Object>>();
+        
+        TestSubscriber<Object> ts = new TestSubscriber<Object>(0);
+        
+        Observable.create(new OnSubscribe<Object>() {
+            @Override
+            public void call(Subscriber<? super Object> t) {
+                subscriber.set(t);
+            }
+        }).onTerminateDetach().subscribe(ts);
+        
+        ts.requestMore(2);
+        
+        new OnSubscribeRange(1, 3).call(subscriber.get());
+        
+        ts.assertValues(1, 2);
+        
+        ts.requestMore(1);
+        
+        ts.assertValues(1, 2, 3);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
+}
