Merge pull request #188 from benjchristensen/issue-173

Subscriptions cleanup

Author: benjchristensen

language-adaptors/rxjava-clojure/src/examples/clojure/rx/lang/clojure/examples/rx_examples.clj

@@ -1,5 +1,5 @@
 (ns rx.lang.clojure.examples.rx-examples
-  (:import rx.Observable)
+  (:import rx.Observable rx.subscriptions.Subscriptions)
   (:require [clj-http.client :as http]))
 
 ; NOTE on naming conventions. I'm using camelCase names (against clojure convention)
@@ -62,7 +62,7 @@
       (-> observer .onCompleted)
       ; return a NoOpSubsription since this blocks and thus
       ; can't be unsubscribed from
-      (Observable/noOpSubscription))))
+      (Subscriptions/empty))))
 
 ; To see output
 (comment
@@ -81,7 +81,7 @@
                 ; after sending all values we complete the sequence
                 (-> observer .onCompleted))]
         ; return a subscription that cancels the future
-        (Observable/createSubscription #(future-cancel f))))))
+        (Subscriptions/create #(future-cancel f))))))
 
 ; To see output
 (comment
@@ -100,7 +100,7 @@
                 ; after sending response to onnext we complete the sequence
                 (-> observer .onCompleted))]
         ; a subscription that cancels the future if unsubscribed
-        (Observable/createSubscription #(future-cancel f))))))
+        (Subscriptions/create #(future-cancel f))))))
 
 ; To see output
 (comment
@@ -147,7 +147,7 @@
                   (-> observer .onCompleted)
                   (catch Exception e (-> observer (.onError e))))) ]
         ; a subscription that cancels the future if unsubscribed
-        (Observable/createSubscription #(future-cancel f))))))
+        (Subscriptions/create #(future-cancel f))))))
 
 (defn getVideoBookmark [userId, videoId]
   "Asynchronously fetch bookmark for video
@@ -165,7 +165,7 @@
                   (-> observer .onCompleted)
                   (catch Exception e (-> observer (.onError e)))))]
         ; a subscription that cancels the future if unsubscribed
-        (Observable/createSubscription #(future-cancel f))))))
+        (Subscriptions/create #(future-cancel f))))))
 
 (defn getVideoMetadata [videoId, preferredLanguage]
   "Asynchronously fetch movie metadata for a given language
@@ -190,7 +190,7 @@
                   (-> observer .onCompleted)
                   (catch Exception e (-> observer (.onError e))))) ]
         ; a subscription that cancels the future if unsubscribed
-        (Observable/createSubscription #(future-cancel f))))))
+        (Subscriptions/create #(future-cancel f))))))
 
 
 (defn getVideoForUser [userId videoId]
@@ -256,7 +256,7 @@
                 ; after sending response to onNext we complete the sequence
                 (-> observer .onCompleted))]
         ; a subscription that cancels the future if unsubscribed
-        (Observable/createSubscription #(future-cancel f))))))
+        (Subscriptions/create #(future-cancel f))))))
 
 ; To see output
 (comment

language-adaptors/rxjava-clojure/src/examples/clojure/rx/lang/clojure/examples/video_example.clj

@@ -1,5 +1,5 @@
 (ns rx.lang.clojure.examples.video-example
-  (:import [rx Observable Observer Subscription]))
+  (:import [rx Observable Observer Subscription] rx.subscriptions.Subscriptions))
 
 ; Adapted from language-adaptors/rxjava-groovy/src/examples/groovy/rx/lang/groovy/examples/VideoExample.groovy
 
@@ -93,7 +93,7 @@
   [f]
   (Observable/create (fn [^Observer observer]
                        (let [f (future (f observer))]
-                         (Observable/createSubscription #(future-cancel f))))))
+                         (Subscriptions/create #(future-cancel f))))))
 
 (defn ^Observable get-list-of-lists
   "
@@ -122,7 +122,7 @@
                        (dotimes [i 50]
                          (.onNext observer (+ (* position 1000) i)))
                        (.onCompleted observer)
-                       (Observable/noOpSubscription))))
+                       (Subscriptions/empty))))
 
 (comment (.subscribe (video-list->videos (video-list 2)) println))
 
@@ -133,7 +133,7 @@
                                           :actors ["actor1" "actor2"]
                                           :duration 5428 })
                        (.onCompleted observer)
-                       (Observable/noOpSubscription))))
+                       (Subscriptions/empty))))
 
 (comment (.subscribe (video->metadata 10) println))
 

language-adaptors/rxjava-groovy/src/examples/groovy/rx/lang/groovy/examples/RxExamples.groovy

@@ -18,6 +18,7 @@ package rx.lang.groovy.examples;
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.subscriptions.Subscriptions;
 import rx.util.functions.Func1;
 
 // --------------------------------------------------
@@ -86,9 +87,9 @@ def customObservableBlocking() {
             }
             // after sending all values we complete the sequence
             observer.onCompleted();
-            // return a NoOpSubsription since this blocks and thus
+            // return an empty subscription since this blocks and thus
             // can't be unsubscribed from
-            return Observable.noOpSubscription();
+            return Subscriptions.empty();
         };
     });
 }
@@ -151,7 +152,7 @@ def fetchWikipediaArticleAsynchronously(String... wikipediaArticleNames) {
             }
             observer.onCompleted();
         }
-        return Observable.noOpSubscription();
+        return Subscriptions.empty();
     });
 }
 
@@ -218,7 +219,7 @@ def fetchWikipediaArticleAsynchronouslyWithErrorHandling(String... wikipediaArti
                 observer.onError(e);
             }
         }
-            return Observable.noOpSubscription();
+            return Subscriptions.empty();
     });
 }
 

rxjava-core/src/main/java/rx/subscriptions/Subscriptions.java

@@ -12,15 +12,15 @@ public class Subscriptions {
      * @return {@link Subscription}
      */
     public static Subscription empty() {
-        return new EmptySubscription();
+        return EMPTY;
     }
 
     /**
      * A {@link Subscription} implemented via a Func
      * 
      * @return {@link Subscription}
      */
-    public static Subscription createSubscription(final Action0 unsubscribe) {
+    public static Subscription create(final Action0 unsubscribe) {
         return new Subscription() {
 
             @Override
@@ -36,7 +36,7 @@ public void unsubscribe() {
      * 
      * @return {@link Subscription}
      */
-    public static Subscription createSubscription(final Object unsubscribe) {
+    public static Subscription create(final Object unsubscribe) {
         final FuncN<?> f = Functions.from(unsubscribe);
         return new Subscription() {
 
@@ -51,8 +51,8 @@ public void unsubscribe() {
     /**
      * A {@link Subscription} that does nothing when its unsubscribe method is called.
      */
-    private static class EmptySubscription implements Subscription {
+    private static Subscription EMPTY = new Subscription() {
         public void unsubscribe() {
         }
-    }
+    };
 }