Merge pull request ReactiveX#3585 from akarnokd/UsingDisposeFix

1.x: fix Completable.using not disposing the resource if the factory crashes during the subscription phase.

Author: akarnokd

src/main/java/rx/Completable.java

@@ -16,13 +16,13 @@
 
 package rx;
 
-import java.util.Iterator;
+import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import rx.Observable.OnSubscribe;
 import rx.annotations.Experimental;
-import rx.exceptions.Exceptions;
+import rx.exceptions.*;
 import rx.functions.*;
 import rx.internal.operators.*;
 import rx.internal.util.*;
@@ -864,12 +864,33 @@ public void call(final CompletableSubscriber s) {
                 try {
                     cs = completableFunc1.call(resource);
                 } catch (Throwable e) {
+                    try {
+                        disposer.call(resource);
+                    } catch (Throwable ex) {
+                        Exceptions.throwIfFatal(e);
+                        Exceptions.throwIfFatal(ex);
+
+                        s.onSubscribe(Subscriptions.unsubscribed());
+                        s.onError(new CompositeException(Arrays.asList(e, ex)));
+                        return;
+                    }
+                    Exceptions.throwIfFatal(e);
+                    
                     s.onSubscribe(Subscriptions.unsubscribed());
                     s.onError(e);
                     return;
                 }
 
                 if (cs == null) {
+                    try {
+                        disposer.call(resource);
+                    } catch (Throwable ex) {
+                        Exceptions.throwIfFatal(ex);
+
+                        s.onSubscribe(Subscriptions.unsubscribed());
+                        s.onError(new CompositeException(Arrays.asList(new NullPointerException("The completable supplied is null"), ex)));
+                        return;
+                    }
                     s.onSubscribe(Subscriptions.unsubscribed());
                     s.onError(new NullPointerException("The completable supplied is null"));
                     return;

src/test/java/rx/CompletableTest.java

@@ -32,6 +32,9 @@
 import rx.subjects.PublishSubject;
 import rx.subscriptions.*;
 
+import static org.mockito.Mockito.*;
+import static org.junit.Assert.*;
+
 /**
  * Test Completable methods and operators.
  */
@@ -3469,4 +3472,135 @@ public void endWithFlowableError() {
         ts.assertError(TestException.class);
         ts.assertNotCompleted();
     }
+    
+    @Test
+    public void usingFactoryThrows() {
+        @SuppressWarnings("unchecked")
+        Action1<Integer> onDispose = mock(Action1.class);
+        
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        
+        Completable.using(new Func0<Integer>() {
+            @Override
+            public Integer call() {
+                return 1;
+            }
+        },
+        new Func1<Integer, Completable>() {
+            @Override
+            public Completable call(Integer t) {
+                throw new TestException();
+            }
+        }, onDispose).subscribe(ts);
+        
+        verify(onDispose).call(1);
+        
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+        ts.assertError(TestException.class);
+    }
+
+    @Test
+    public void usingFactoryAndDisposerThrow() {
+        Action1<Integer> onDispose = new Action1<Integer>() {
+            @Override
+            public void call(Integer t) {
+                throw new TestException();
+            }
+        };
+        
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        
+        Completable.using(new Func0<Integer>() {
+            @Override
+            public Integer call() {
+                return 1;
+            }
+        },
+        new Func1<Integer, Completable>() {
+            @Override
+            public Completable call(Integer t) {
+                throw new TestException();
+            }
+        }, onDispose).subscribe(ts);
+        
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+        ts.assertError(CompositeException.class);
+        
+        CompositeException ex = (CompositeException)ts.getOnErrorEvents().get(0);
+        
+        List<Throwable> listEx = ex.getExceptions();
+        
+        assertEquals(2, listEx.size());
+        
+        assertTrue(listEx.get(0).toString(), listEx.get(0) instanceof TestException);
+        assertTrue(listEx.get(1).toString(), listEx.get(1) instanceof TestException);
+    }
+
+    @Test
+    public void usingFactoryReturnsNull() {
+        @SuppressWarnings("unchecked")
+        Action1<Integer> onDispose = mock(Action1.class);
+        
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        
+        Completable.using(new Func0<Integer>() {
+            @Override
+            public Integer call() {
+                return 1;
+            }
+        },
+        new Func1<Integer, Completable>() {
+            @Override
+            public Completable call(Integer t) {
+                return null;
+            }
+        }, onDispose).subscribe(ts);
+        
+        verify(onDispose).call(1);
+        
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+        ts.assertError(NullPointerException.class);
+    }
+
+    @Test
+    public void usingFactoryReturnsNullAndDisposerThrows() {
+        Action1<Integer> onDispose = new Action1<Integer>() {
+            @Override
+            public void call(Integer t) {
+                throw new TestException();
+            }
+        };
+        
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        
+        Completable.using(new Func0<Integer>() {
+            @Override
+            public Integer call() {
+                return 1;
+            }
+        },
+        new Func1<Integer, Completable>() {
+            @Override
+            public Completable call(Integer t) {
+                return null;
+            }
+        }, onDispose).subscribe(ts);
+        
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+        ts.assertError(CompositeException.class);
+        
+        CompositeException ex = (CompositeException)ts.getOnErrorEvents().get(0);
+        
+        List<Throwable> listEx = ex.getExceptions();
+        
+        assertEquals(2, listEx.size());
+        
+        assertTrue(listEx.get(0).toString(), listEx.get(0) instanceof NullPointerException);
+        assertTrue(listEx.get(1).toString(), listEx.get(1) instanceof TestException);
+    }
+
 }