Playing with Rx + Clojure

sys-action.clj

(defmacro sys-action
   "Translates to a gen-delegate for a System.Action<,...> call"
   [typesyms & body ]
   (let [types (map (fn [tsym] (clojure.lang.CljCompiler.Ast.HostExpr/MaybeType tsym false)) typesyms)
         join  ; clojure.string not yet loaded
           (fn [coll] 
            (loop [sb (StringBuilder. (str (first coll)))
                 more (next coll)]
            (if more
              (recur (-> sb (.Append ",") (.Append (str (first more))))
                   (next more))
              (str sb))))
    ftype (symbol (str "System.Action`" (count types) "[" (join types) "]"))]
    `(gen-delegate ~ftype ~@body)))

test-rxlinq.clj

user=> (System.Reflection.Assembly/Load "System.CoreEx, Version=1.0.2838.104, Culture=neutral, PublicKeyToken=31bf3856ad364e35")
#<RuntimeAssembly System.CoreEx, Version=1.0.2838.104, Culture=neutral, PublicKeyToken=31bf3856ad364e35>
user=> (System.Reflection.Assembly/Load "System.Reactive, Version=1.0.2838.104, Culture=neutral, PublicKeyToken=31bf3856ad364e35")
#<RuntimeAssembly System.Reactive, Version=1.0.2838.104, Culture=neutral, PublicKeyToken=31bf3856ad364e35>
user=> (import '(System.Linq Observable Observer))
System.Linq.Observer
user=> (import '(System ObservableExtensions))
System.ObservableExtensions
user=> (def o1 (Observable/Range 1 10))
#'user/o1
user=> (def o2 (. Observable (generic Where Int32) o1 (sys-func [Int32 Boolean] [x] (even? x))))
#'user/o2
user=> (def on-next (sys-action [Int32] [x] (printf "Next %d\n" x )))
#'user/on-next
user => (def observer (. Observer (generic Create Int32) on-next))
#'user/observer
user=> (def sub (.Subscribe o2 observer))
Next 2
Next 4
Next 6
Next 8
Next 10
#'user/sub
user=> (def sub1 (. ObservableExtensions (generic Subscribe Int32) o2 on-next))
Next 2
Next 4
Next 6
Next 8
Next 10
#'user/sub1