Decouple Dispose Function for Using

As per discussion at #1466

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -2906,12 +2906,45 @@ public final static Observable<Long> timer(long delay, TimeUnit unit, Scheduler
      *            the factory function to create a resource object that depends on the Observable
      * @param observableFactory
      *            the factory function to create an Observable
+     * @param disposeAction
+     *            the function that will dispose of the resource
      * @return the Observable whose lifetime controls the lifetime of the dependent resource object
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#using">RxJava wiki: using</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229585.aspx">MSDN: Observable.Using</a>
      */
-    public final static <T, Resource extends Subscription> Observable<T> using(Func0<Resource> resourceFactory, Func1<Resource, ? extends Observable<? extends T>> observableFactory) {
-        return create(new OnSubscribeUsing<T, Resource>(resourceFactory, observableFactory));
+    public final static <T, Resource> Observable<T> using(
+            final Func0<Resource> resourceFactory,
+            final Func1<? super Resource, ? extends Observable<? extends T>> observableFactory,
+            final Action1<? super Resource> disposeAction) {
+        return create(new OnSubscribeUsing<T, Resource>(resourceFactory, observableFactory, disposeAction));
+    }
+
+    /**
+     * Constructs an Observable that creates a dependent resource object.
+     * <p>
+     * <img width="640" height="400" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/using.png" alt="">
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code using} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * 
+     * @param resourceFactory
+     *            the factory function to create a resource object that depends on the Observable
+     * @param observableFactory
+     *            the factory function to create an Observable
+     * @return the Observable whose lifetime controls the lifetime of the dependent resource object
+     * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#using">RxJava wiki: using</a>
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229585.aspx">MSDN: Observable.Using</a>
+     */
+    public final static <T, Resource extends Subscription> Observable<T> using(Func0<Resource> resourceFactory, Func1<? super Resource, ? extends Observable<? extends T>> observableFactory) {
+        return create(new OnSubscribeUsing<T, Resource>(resourceFactory, observableFactory, new Action1<Resource>() {
+
+            @Override
+            public void call(Resource r) {
+                r.unsubscribe();
+            }
+
+        }));
     }
 
     /**

rxjava-core/src/main/java/rx/internal/operators/OnSubscribeUsing.java

@@ -18,34 +18,46 @@
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.Subscriber;
-import rx.Subscription;
+import rx.functions.Action0;
+import rx.functions.Action1;
 import rx.functions.Func0;
 import rx.functions.Func1;
+import rx.subscriptions.Subscriptions;
 
 /**
  * Constructs an observable sequence that depends on a resource object.
  */
-public final class OnSubscribeUsing<T, Resource extends Subscription> implements OnSubscribe<T> {
+public final class OnSubscribeUsing<T, Resource> implements OnSubscribe<T> {
 
     private final Func0<Resource> resourceFactory;
-    private final Func1<Resource, ? extends Observable<? extends T>> observableFactory;
+    private final Func1<? super Resource, ? extends Observable<? extends T>> observableFactory;
+    private final Action1<? super Resource> dispose;
 
-    public OnSubscribeUsing(Func0<Resource> resourceFactory, Func1<Resource, ? extends Observable<? extends T>> observableFactory) {
+    public OnSubscribeUsing(Func0<Resource> resourceFactory,
+            Func1<? super Resource, ? extends Observable<? extends T>> observableFactory,
+            Action1<? super Resource> dispose) {
         this.resourceFactory = resourceFactory;
         this.observableFactory = observableFactory;
+        this.dispose = dispose;
     }
 
     public void call(Subscriber<? super T> subscriber) {
-        Resource resource = null;
         try {
-            resource = resourceFactory.call();
-            subscriber.add(resource);
+            final Resource resource = resourceFactory.call();
+            subscriber.add(Subscriptions.create(new Action0() {
+
+                @Override
+                public void call() {
+                    dispose.call(resource);
+                }
+
+            }));
             Observable<? extends T> observable = observableFactory.call(resource);
             observable.subscribe(subscriber);
         } catch (Throwable e) {
-            if (resource != null) {
-                resource.unsubscribe();
-            }
+            // eagerly call unsubscribe since this operator is specifically about resource management
+            subscriber.unsubscribe();
+            // then propagate error
             subscriber.onError(e);
         }
     }

rxjava-core/src/test/java/rx/internal/operators/OnSubscribeUsingTest.java

@@ -32,18 +32,36 @@
 import rx.Subscription;
 import rx.exceptions.TestException;
 import rx.functions.Action0;
+import rx.functions.Action1;
 import rx.functions.Func0;
 import rx.functions.Func1;
 import rx.subscriptions.Subscriptions;
 
 public class OnSubscribeUsingTest {
 
-    private static interface Resource extends Subscription {
+    private static interface Resource {
         public String getTextFromWeb();
 
+        public void dispose();
+    }
+
+    private static class DisposeAction implements Action1<Resource> {
+
         @Override
-        public void unsubscribe();
+        public void call(Resource r) {
+            r.dispose();
+        }
+
     }
+    
+    private final Action1<Subscription> disposeSubscription = new Action1<Subscription>() {
+
+        @Override
+        public void call(Subscription s) {
+            s.unsubscribe();
+        }
+        
+    };
 
     @Test
     public void testUsing() {
@@ -66,7 +84,7 @@ public Observable<String> call(Resource resource) {
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = (Observer<String>) mock(Observer.class);
-        Observable<String> observable = Observable.using(resourceFactory, observableFactory);
+        Observable<String> observable = Observable.using(resourceFactory, observableFactory, new DisposeAction());
         observable.subscribe(observer);
 
         InOrder inOrder = inOrder(observer);
@@ -76,7 +94,7 @@ public Observable<String> call(Resource resource) {
         inOrder.verifyNoMoreInteractions();
 
         // The resouce should be closed
-        verify(resource, times(1)).unsubscribe();
+        verify(resource, times(1)).dispose();
     }
 
     @Test
@@ -97,14 +115,10 @@ public String getTextFromWeb() {
                         }
                         return "Nothing";
                     }
-
+                    
                     @Override
-                    public void unsubscribe() {
-                    }
-
-                    @Override
-                    public boolean isUnsubscribed() {
-                        return false;
+                    public void dispose() {
+                        // do nothing
                     }
 
                 };
@@ -120,7 +134,7 @@ public Observable<String> call(Resource resource) {
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = (Observer<String>) mock(Observer.class);
-        Observable<String> observable = Observable.using(resourceFactory, observableFactory);
+        Observable<String> observable = Observable.using(resourceFactory, observableFactory, new DisposeAction());
         observable.subscribe(observer);
         observable.subscribe(observer);
 
@@ -151,8 +165,8 @@ public Observable<Integer> call(Subscription subscription) {
                 return Observable.empty();
             }
         };
-
-        Observable.using(resourceFactory, observableFactory).toBlocking().last();
+        
+        Observable.using(resourceFactory, observableFactory, disposeSubscription).toBlocking().last();
     }
 
     @Test
@@ -171,9 +185,9 @@ public Observable<Integer> call(Subscription subscription) {
                 throw new TestException();
             }
         };
-
+        
         try {
-            Observable.using(resourceFactory, observableFactory).toBlocking().last();
+            Observable.using(resourceFactory, observableFactory, disposeSubscription).toBlocking().last();
             fail("Should throw a TestException when the observableFactory throws it");
         } catch (TestException e) {
             // Make sure that unsubscribe is called so that users can close
@@ -203,9 +217,11 @@ public void call(Subscriber<? super Integer> t1) {
                 });
             }
         };
+        
+        
 
         try {
-            Observable.using(resourceFactory, observableFactory).toBlocking().last();
+            Observable.using(resourceFactory, observableFactory, disposeSubscription).toBlocking().last();
             fail("Should throw a TestException when the observableFactory throws it");
         } catch (TestException e) {
             // Make sure that unsubscribe is called so that users can close