Make rx/action implement new OnSubscribe interface

Author: daveray

language-adaptors/rxjava-clojure/README.md

@@ -46,6 +46,19 @@ The `rx/action` macro is identical to `rx/fn` except that the object returned im
                 (rx/action [] (println "Sequence complete"))))
 ```
 
+## Using Observable/create
+As of 0.17, `rx.Observable/create` takes an implementation of `rx.Observable$OnSubscribe` which is basically an alias for `rx.util.functions.Action1` that takes an `rx.Subscriber` as its argument. Thus, you can just use `rx/action` when creating new observables:
+
+```clojure
+; A simple observable that emits 0..9 taking unsubscribe into account
+(Observable/create (rx/action [^rx.Subscriber s]
+                     (loop [i 0]
+                       (when (and (< i 10) (.isUnsubscribed s))
+                         (.onNext s i)
+                         (recur (inc i))))
+                     (.onCompleted s)))
+```
+
 # Gotchas
 Here are a few things to keep in mind when using this interop:
 

language-adaptors/rxjava-clojure/src/main/clojure/rx/lang/clojure/interop.clj

@@ -19,12 +19,19 @@
        (reify
          ; If they want Func1, give them onSubscribe as well so Observable/create can be
          ; used seemlessly with rx/fn.
+         ; TODO remove this when OnSubscriberFunc is removed
          ~@(if (and (= prefix "rx.util.functions.Func")
                     (some #{1} arities))
              `(rx.Observable$OnSubscribeFunc
                 (~'onSubscribe [~'this observer#]
                   (~f-name observer#))))
 
+         ; OnSubscribe is just an Action1, so add it to the list of implemented interfaces
+         ; so an action cab be used with Observable/create
+         ~@(if (and (= prefix "rx.util.functions.Action")
+                    (some #{1} arities))
+             `(rx.Observable$OnSubscribe))
+
          ~@(mapcat (clojure.core/fn [n]
                      (let [ifc-sym  (symbol (str prefix n))
                            arg-syms (map #(symbol (str "v" %)) (range n))]
@@ -94,7 +101,8 @@
 
 (defn action*
   "Given function f, returns an object that implements rx.util.functions.Action0-3
-  by delegating to the given function.
+  by delegating to the given function. Also implements rx.Observable$OnSubscribe which
+  is just an Action1.
 
   Example:
 

language-adaptors/rxjava-clojure/src/test/clojure/rx/lang/clojure/interop_test.clj

@@ -70,6 +70,7 @@
   (testing "implements Action0-3"
     (let [calls (atom [])
           a (rx/action* #(swap! calls conj (vec %&)))]
+      (is (instance? rx.Observable$OnSubscribe a))
       (is (instance? rx.util.functions.Action0 a))
       (is (instance? rx.util.functions.Action1 a))
       (is (instance? rx.util.functions.Action2 a))
@@ -114,7 +115,7 @@
 
 (deftest test-basic-usage
 
-  (testing "can create an observable"
+  (testing "can create an observable with old style fn"
     (is (= 99
            (-> (Observable/create (rx/fn [^rx.Observer o]
                                     (.onNext o 99)
@@ -123,6 +124,14 @@
                .toBlockingObservable
                .single))))
 
+  (testing "can create an observable with new-style action"
+    (is (= 99
+           (-> (Observable/create (rx/action [^rx.Subscriber s]
+                                    (when-not (.isUnsubscribed s)
+                                      (.onNext s 99))
+                                    (.onCompleted s)))
+               .toBlockingObservable
+               .single))))
   (testing "can pass rx/fn to map and friends"
     (is (= (+ 1 4 9)
            (-> (Observable/from [1 2 3])