OperatorUsing

Author: zsxwing

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

@@ -76,7 +76,7 @@
 import rx.operators.OperationTimer;
 import rx.operators.OperationToMap;
 import rx.operators.OperationToMultimap;
-import rx.operators.OperationUsing;
+import rx.operators.OperatorUsing;
 import rx.operators.OperationWindow;
 import rx.operators.OperatorAll;
 import rx.operators.OperatorAmb;
@@ -2589,8 +2589,8 @@ public final static Observable<Long> timer(long delay, TimeUnit unit, Scheduler
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-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, Observable<T>> observableFactory) {
-        return create(OperationUsing.using(resourceFactory, observableFactory));
+    public final static <T, Resource extends Subscription> Observable<T> using(Func0<Resource> resourceFactory, Func1<Resource, ? extends Observable<? extends T>> observableFactory) {
+        return create(new OperatorUsing<T, Resource>(resourceFactory, observableFactory));
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationUsing.java

@@ -0,0 +1,53 @@
+/**
+ * Copyright 2014 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.operators;
+
+import rx.Observable;
+import rx.Observable.OnSubscribe;
+import rx.Subscriber;
+import rx.Subscription;
+import rx.functions.Func0;
+import rx.functions.Func1;
+
+/**
+ * Constructs an observable sequence that depends on a resource object.
+ */
+public final class OperatorUsing<T, Resource extends Subscription> implements OnSubscribe<T> {
+
+    private final Func0<Resource> resourceFactory;
+    private final Func1<Resource, ? extends Observable<? extends T>> observableFactory;
+
+    public OperatorUsing(Func0<Resource> resourceFactory, Func1<Resource, ? extends Observable<? extends T>> observableFactory) {
+        this.resourceFactory = resourceFactory;
+        this.observableFactory = observableFactory;
+    }
+
+    public void call(Subscriber<? super T> subscriber) {
+        Resource resource = null;
+        try {
+            resource = resourceFactory.call();
+            subscriber.add(resource);
+            Observable<? extends T> observable = observableFactory.call(resource);
+            observable.subscribe(subscriber);
+        } catch (Throwable e) {
+            if (resource != null) {
+                resource.unsubscribe();
+            }
+            subscriber.onError(e);
+        }
+    }
+
+}

rxjava-core/src/main/java/rx/operators/OperatorUsing.java

@@ -21,7 +21,6 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
-import static rx.operators.OperationUsing.using;
 
 import org.junit.Test;
 import org.mockito.InOrder;
@@ -35,7 +34,7 @@
 import rx.functions.Func1;
 import rx.subscriptions.Subscriptions;
 
-public class OperationUsingTest {
+public class OperatorUsingTest {
 
     @SuppressWarnings("serial")
     private static class TestException extends RuntimeException {
@@ -69,8 +68,7 @@ public Observable<String> call(Resource resource) {
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = (Observer<String>) mock(Observer.class);
-        Observable<String> observable = Observable.create(using(
-                resourceFactory, observableFactory));
+        Observable<String> observable = Observable.using(resourceFactory, observableFactory);
         observable.subscribe(observer);
 
         InOrder inOrder = inOrder(observer);
@@ -124,8 +122,7 @@ public Observable<String> call(Resource resource) {
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = (Observer<String>) mock(Observer.class);
-        Observable<String> observable = Observable.create(using(
-                resourceFactory, observableFactory));
+        Observable<String> observable = Observable.using(resourceFactory, observableFactory);
         observable.subscribe(observer);
         observable.subscribe(observer);
 
@@ -157,8 +154,7 @@ public Observable<Integer> call(Subscription subscription) {
             }
         };
 
-        Observable.create(using(resourceFactory, observableFactory))
-                .toBlockingObservable().last();
+        Observable.using(resourceFactory, observableFactory).toBlockingObservable().last();
     }
 
     @Test
@@ -179,8 +175,7 @@ public Observable<Integer> call(Subscription subscription) {
         };
 
         try {
-            Observable.create(using(resourceFactory, observableFactory))
-                    .toBlockingObservable().last();
+            Observable.using(resourceFactory, observableFactory).toBlockingObservable().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
@@ -212,8 +207,7 @@ public Subscription onSubscribe(Observer<? super Integer> t1) {
         };
 
         try {
-            Observable.create(using(resourceFactory, observableFactory))
-                    .toBlockingObservable().last();
+            Observable.using(resourceFactory, observableFactory).toBlockingObservable().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