3.x: Make using() resource disposal order consistent with eager-mode (#6534)

akarnokd · web-flow · commit 62d2b3153362 · 2019-06-24T10:06:10.000+02:00
diff --git a/src/main/java/io/reactivex/Completable.java b/src/main/java/io/reactivex/Completable.java
@@ -1032,8 +1032,11 @@ public static <R> Completable using(Supplier<R> resourceSupplier,
      * @param completableFunction the function that given a resource returns a non-null
      * Completable instance that will be subscribed to
      * @param disposer the consumer that disposes the resource created by the resource supplier
-     * @param eager if true, the resource is disposed before the terminal event is emitted, if false, the
-     * resource is disposed after the terminal event has been emitted
+     * @param eager
+     *            If {@code true} then resource disposal will happen either on a {@code dispose()} call before the upstream is disposed
+     *            or just before the emission of a terminal event ({@code onComplete} or {@code onError}).
+     *            If {@code false} the resource disposal will happen either on a {@code dispose()} call after the upstream is disposed
+     *            or just after the emission of a terminal event ({@code onComplete} or {@code onError}).
      * @return the new Completable instance
      */
     @CheckReturnValue
diff --git a/src/main/java/io/reactivex/Flowable.java b/src/main/java/io/reactivex/Flowable.java
@@ -4609,8 +4609,10 @@ public static <T, D> Flowable<T> using(Supplier<? extends D> resourceSupplier,
      * @param resourceDisposer
      *            the function that will dispose of the resource
      * @param eager
-     *            if {@code true} then disposal will happen either on cancellation or just before emission of
-     *            a terminal event ({@code onComplete} or {@code onError}).
+     *            If {@code true} then resource disposal will happen either on a {@code cancel()} call before the upstream is disposed
+     *            or just before the emission of a terminal event ({@code onComplete} or {@code onError}).
+     *            If {@code false} the resource disposal will happen either on a {@code cancel()} call after the upstream is disposed
+     *            or just after the emission of a terminal event ({@code onComplete} or {@code onError}).
      * @return the Publisher whose lifetime controls the lifetime of the dependent resource object
      * @see <a href="https://reactivex.io/documentation/operators/using.html">ReactiveX operators documentation: Using</a>
      * @since 2.0
diff --git a/src/main/java/io/reactivex/Maybe.java b/src/main/java/io/reactivex/Maybe.java
@@ -1772,8 +1772,10 @@ public static <T, D> Maybe<T> using(Supplier<? extends D> resourceSupplier,
      * @param resourceDisposer
      *            the function that will dispose of the resource
      * @param eager
-     *            if {@code true} then disposal will happen either on a dispose() call or just before emission of
-     *            a terminal event ({@code onComplete} or {@code onError}).
+     *            If {@code true} then resource disposal will happen either on a {@code dispose()} call before the upstream is disposed
+     *            or just before the emission of a terminal event ({@code onSuccess}, {@code onComplete} or {@code onError}).
+     *            If {@code false} the resource disposal will happen either on a {@code dispose()} call after the upstream is disposed
+     *            or just after the emission of a terminal event ({@code onSuccess}, {@code onComplete} or {@code onError}).
      * @return the Maybe whose lifetime controls the lifetime of the dependent resource object
      * @see <a href="https://reactivex.io/documentation/operators/using.html">ReactiveX operators documentation: Using</a>
      */
diff --git a/src/main/java/io/reactivex/Observable.java b/src/main/java/io/reactivex/Observable.java
@@ -4095,8 +4095,10 @@ public static <T, D> Observable<T> using(Supplier<? extends D> resourceSupplier,
      * @param disposer
      *            the function that will dispose of the resource
      * @param eager
-     *            if {@code true} then disposal will happen either on a dispose() call or just before emission of
-     *            a terminal event ({@code onComplete} or {@code onError}).
+     *            If {@code true} then resource disposal will happen either on a {@code dispose()} call before the upstream is disposed
+     *            or just before the emission of a terminal event ({@code onComplete} or {@code onError}).
+     *            If {@code false} the resource disposal will happen either on a {@code dispose()} call after the upstream is disposed
+     *            or just after the emission of a terminal event ({@code onComplete} or {@code onError}).
      * @return the ObservableSource whose lifetime controls the lifetime of the dependent resource object
      * @see <a href="https://reactivex.io/documentation/operators/using.html">ReactiveX operators documentation: Using</a>
      * @since 2.0
diff --git a/src/main/java/io/reactivex/Single.java b/src/main/java/io/reactivex/Single.java
@@ -1483,8 +1483,10 @@ public static <T, U> Single<T> using(Supplier<U> resourceSupplier,
      *                  that particular resource when the generated SingleSource terminates
      *                  (successfully or with an error) or gets disposed.
      * @param eager
-     *                 if true, the disposer is called before the terminal event is signalled
-     *                 if false, the disposer is called after the terminal event is delivered to downstream
+     *            If {@code true} then resource disposal will happen either on a {@code dispose()} call before the upstream is disposed
+     *            or just before the emission of a terminal event ({@code onSuccess} or {@code onError}).
+     *            If {@code false} the resource disposal will happen either on a {@code dispose()} call after the upstream is disposed
+     *            or just after the emission of a terminal event ({@code onSuccess} or {@code onError}).
      * @return the new Single instance
      * @since 2.0
      */
diff --git a/src/main/java/io/reactivex/internal/operators/completable/CompletableUsing.java b/src/main/java/io/reactivex/internal/operators/completable/CompletableUsing.java
@@ -106,13 +106,19 @@ static final class UsingObserver<R>
 
         @Override
         public void dispose() {
-            upstream.dispose();
-            upstream = DisposableHelper.DISPOSED;
-            disposeResourceAfter();
+            if (eager) {
+                disposeResource();
+                upstream.dispose();
+                upstream = DisposableHelper.DISPOSED;
+            } else {
+                upstream.dispose();
+                upstream = DisposableHelper.DISPOSED;
+                disposeResource();
+            }
         }
 
         @SuppressWarnings("unchecked")
-        void disposeResourceAfter() {
+        void disposeResource() {
             Object resource = getAndSet(this);
             if (resource != this) {
                 try {
@@ -159,7 +165,7 @@ public void onError(Throwable e) {
             downstream.onError(e);
 
             if (!eager) {
-                disposeResourceAfter();
+                disposeResource();
             }
         }
 
@@ -185,7 +191,7 @@ public void onComplete() {
             downstream.onComplete();
 
             if (!eager) {
-                disposeResourceAfter();
+                disposeResource();
             }
         }
     }
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableUsing.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableUsing.java
@@ -126,7 +126,7 @@ public void onError(Throwable t) {
             } else {
                 downstream.onError(t);
                 upstream.cancel();
-                disposeAfter();
+                disposeResource();
             }
         }
 
@@ -148,7 +148,7 @@ public void onComplete() {
             } else {
                 downstream.onComplete();
                 upstream.cancel();
-                disposeAfter();
+                disposeResource();
             }
         }
 
@@ -159,11 +159,18 @@ public void request(long n) {
 
         @Override
         public void cancel() {
-            disposeAfter();
-            upstream.cancel();
+            if (eager) {
+                disposeResource();
+                upstream.cancel();
+                upstream = SubscriptionHelper.CANCELLED;
+            } else {
+                upstream.cancel();
+                upstream = SubscriptionHelper.CANCELLED;
+                disposeResource();
+            }
         }
 
-        void disposeAfter() {
+        void disposeResource() {
             if (compareAndSet(false, true)) {
                 try {
                     disposer.accept(resource);
diff --git a/src/main/java/io/reactivex/internal/operators/maybe/MaybeUsing.java b/src/main/java/io/reactivex/internal/operators/maybe/MaybeUsing.java
@@ -117,13 +117,19 @@ static final class UsingObserver<T, D>
 
         @Override
         public void dispose() {
-            upstream.dispose();
-            upstream = DisposableHelper.DISPOSED;
-            disposeResourceAfter();
+            if (eager) {
+                disposeResource();
+                upstream.dispose();
+                upstream = DisposableHelper.DISPOSED;
+            } else {
+                upstream.dispose();
+                upstream = DisposableHelper.DISPOSED;
+                disposeResource();
+            }
         }
 
         @SuppressWarnings("unchecked")
-        void disposeResourceAfter() {
+        void disposeResource() {
             Object resource = getAndSet(this);
             if (resource != this) {
                 try {
@@ -171,7 +177,7 @@ public void onSuccess(T value) {
             downstream.onSuccess(value);
 
             if (!eager) {
-                disposeResourceAfter();
+                disposeResource();
             }
         }
 
@@ -196,7 +202,7 @@ public void onError(Throwable e) {
             downstream.onError(e);
 
             if (!eager) {
-                disposeResourceAfter();
+                disposeResource();
             }
         }
 
@@ -222,7 +228,7 @@ public void onComplete() {
             downstream.onComplete();
 
             if (!eager) {
-                disposeResourceAfter();
+                disposeResource();
             }
         }
     }
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableUsing.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableUsing.java
@@ -120,7 +120,7 @@ public void onError(Throwable t) {
             } else {
                 downstream.onError(t);
                 upstream.dispose();
-                disposeAfter();
+                disposeResource();
             }
         }
 
@@ -142,22 +142,29 @@ public void onComplete() {
             } else {
                 downstream.onComplete();
                 upstream.dispose();
-                disposeAfter();
+                disposeResource();
             }
         }
 
         @Override
         public void dispose() {
-            disposeAfter();
-            upstream.dispose();
+            if (eager) {
+                disposeResource();
+                upstream.dispose();
+                upstream = DisposableHelper.DISPOSED;
+            } else {
+                upstream.dispose();
+                upstream = DisposableHelper.DISPOSED;
+                disposeResource();
+            }
         }
 
         @Override
         public boolean isDisposed() {
             return get();
         }
 
-        void disposeAfter() {
+        void disposeResource() {
             if (compareAndSet(false, true)) {
                 try {
                     disposer.accept(resource);
diff --git a/src/main/java/io/reactivex/internal/operators/single/SingleUsing.java b/src/main/java/io/reactivex/internal/operators/single/SingleUsing.java
@@ -106,9 +106,15 @@ static final class UsingSingleObserver<T, U> extends
 
         @Override
         public void dispose() {
-            upstream.dispose();
-            upstream = DisposableHelper.DISPOSED;
-            disposeAfter();
+            if (eager) {
+                disposeResource();
+                upstream.dispose();
+                upstream = DisposableHelper.DISPOSED;
+            } else {
+                upstream.dispose();
+                upstream = DisposableHelper.DISPOSED;
+                disposeResource();
+            }
         }
 
         @Override
@@ -148,7 +154,7 @@ public void onSuccess(T value) {
             downstream.onSuccess(value);
 
             if (!eager) {
-                disposeAfter();
+                disposeResource();
             }
         }
 
@@ -174,12 +180,12 @@ public void onError(Throwable e) {
             downstream.onError(e);
 
             if (!eager) {
-                disposeAfter();
+                disposeResource();
             }
         }
 
         @SuppressWarnings("unchecked")
-        void disposeAfter() {
+        void disposeResource() {
             Object u = getAndSet(this);
             if (u != this) {
                 try {
diff --git a/src/test/java/io/reactivex/internal/operators/completable/CompletableUsingTest.java b/src/test/java/io/reactivex/internal/operators/completable/CompletableUsingTest.java
@@ -570,4 +570,67 @@ public void run() {
         }
     }
 
+    @Test
+    public void eagerDisposeResourceThenDisposeUpstream() {
+        final StringBuilder sb = new StringBuilder();
+
+        TestObserver<Void> to = Completable.using(Functions.justSupplier(1),
+            new Function<Integer, Completable>() {
+                @Override
+                public Completable apply(Integer t) throws Throwable {
+                    return Completable.never()
+                            .doOnDispose(new Action() {
+                                @Override
+                                public void run() throws Throwable {
+                                    sb.append("Dispose");
+                                }
+                            })
+                            ;
+                }
+            }, new Consumer<Integer>() {
+                @Override
+                public void accept(Integer t) throws Throwable {
+                    sb.append("Resource");
+                }
+            }, true)
+        .test()
+        ;
+        to.assertEmpty();
+
+        to.dispose();
+
+        assertEquals("ResourceDispose", sb.toString());
+    }
+
+    @Test
+    public void nonEagerDisposeUpstreamThenDisposeResource() {
+        final StringBuilder sb = new StringBuilder();
+
+        TestObserver<Void> to = Completable.using(Functions.justSupplier(1),
+            new Function<Integer, Completable>() {
+                @Override
+                public Completable apply(Integer t) throws Throwable {
+                    return Completable.never()
+                            .doOnDispose(new Action() {
+                                @Override
+                                public void run() throws Throwable {
+                                    sb.append("Dispose");
+                                }
+                            })
+                            ;
+                }
+            }, new Consumer<Integer>() {
+                @Override
+                public void accept(Integer t) throws Throwable {
+                    sb.append("Resource");
+                }
+            }, false)
+        .test()
+        ;
+        to.assertEmpty();
+
+        to.dispose();
+
+        assertEquals("DisposeResource", sb.toString());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableUsingTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableUsingTest.java
diff --git a/src/test/java/io/reactivex/internal/operators/maybe/MaybeUsingTest.java b/src/test/java/io/reactivex/internal/operators/maybe/MaybeUsingTest.java
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableUsingTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableUsingTest.java
diff --git a/src/test/java/io/reactivex/internal/operators/single/SingleUsingTest.java b/src/test/java/io/reactivex/internal/operators/single/SingleUsingTest.java

Original file line number	Diff line number	Diff line change
`@@ -117,13 +117,19 @@ static final class UsingObserver<T, D>`
`117`	`117`
`118`	`118`	`@Override`
`119`	`119`	`public void dispose() {`
`120`		`- upstream.dispose();`
`121`		`- upstream = DisposableHelper.DISPOSED;`
`122`		`- disposeResourceAfter();`
	`120`	`+ if (eager) {`
	`121`	`+ disposeResource();`
	`122`	`+ upstream.dispose();`
	`123`	`+ upstream = DisposableHelper.DISPOSED;`
	`124`	`+ } else {`
	`125`	`+ upstream.dispose();`
	`126`	`+ upstream = DisposableHelper.DISPOSED;`
	`127`	`+ disposeResource();`
	`128`	`+ }`
`123`	`129`	`}`
`124`	`130`
`125`	`131`	`@SuppressWarnings("unchecked")`
`126`		`- void disposeResourceAfter() {`
	`132`	`+ void disposeResource() {`
`127`	`133`	`Object resource = getAndSet(this);`
`128`	`134`	`if (resource != this) {`
`129`	`135`	`try {`
`@@ -171,7 +177,7 @@ public void onSuccess(T value) {`
`171`	`177`	`downstream.onSuccess(value);`
`172`	`178`
`173`	`179`	`if (!eager) {`
`174`		`- disposeResourceAfter();`
	`180`	`+ disposeResource();`
`175`	`181`	`}`
`176`	`182`	`}`
`177`	`183`
`@@ -196,7 +202,7 @@ public void onError(Throwable e) {`
`196`	`202`	`downstream.onError(e);`
`197`	`203`
`198`	`204`	`if (!eager) {`
`199`		`- disposeResourceAfter();`
	`205`	`+ disposeResource();`
`200`	`206`	`}`
`201`	`207`	`}`
`202`	`208`
`@@ -222,7 +228,7 @@ public void onComplete() {`
`222`	`228`	`downstream.onComplete();`
`223`	`229`
`224`	`230`	`if (!eager) {`
`225`		`- disposeResourceAfter();`
	`231`	`+ disposeResource();`
`226`	`232`	`}`
`227`	`233`	`}`
`228`	`234`	`}`