Merge pull request #346 from benjchristensen/BlockingObservable

BlockingObservable Refactor
ReactiveX · Sep 5, 2013 · e81577b · e81577b
2 parents a698242 + 4850c86
commit e81577b
Show file tree

Hide file tree

Showing 3 changed files with 56 additions and 285 deletions.
diff --git a/language-adaptors/rxjava-clojure/src/test/clojure/rx/lang/clojure/interop_test.clj b/language-adaptors/rxjava-clojure/src/test/clojure/rx/lang/clojure/interop_test.clj
@@ -120,14 +120,16 @@
                                     (.onNext o 99)
                                     (.onCompleted o)
                                     (rx.subscriptions.Subscriptions/empty)))
-               (BlockingObservable/single)))))
+               .toBlockingObservable
+               .single))))
 
   (testing "can pass rx/fn to map and friends"
     (is (= (+ 1 4 9)
            (-> (Observable/from [1 2 3])
                (.map (rx/fn [v] (* v v)))
                (.reduce (rx/fn* +))
-               (BlockingObservable/single)))))
+           .toBlockingObservable
+           .single))))
 
   (testing "can pass rx/action to subscribe and friends"
     (let [finally-called (atom nil)

diff --git a/language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy b/language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy
@@ -324,8 +324,12 @@ def class ObservableTests {
             return Observable.from(1, 3, 2, 5, 4);
         }
 
-        public TestObservable getObservable() {
-            return new TestObservable(counter++);
+        public TestOnSubscribe getOnSubscribe() {
+            return new TestOnSubscribe(counter++);
+        }
+
+        public Observable getObservable() {
+            return Observable.create(getOnSubscribe());
         }
     }
 
@@ -335,14 +339,14 @@ def class ObservableTests {
         public void received(Object o);
     }
 
-    def class TestObservable extends Observable<String> {
+    def class TestOnSubscribe implements OnSubscribeFunc<String> {
         private final int count;
 
-        public TestObservable(int count) {
+        public TestOnSubscribe(int count) {
             this.count = count;
         }
 
-        public Subscription subscribe(Observer<String> observer) {
+        public Subscription onSubscribe(Observer<String> observer) {
 
             observer.onNext("hello_" + count);
             observer.onCompleted();