Observable/toObservable not found as static function?

Question

Using rxjava (http://netflix.github.io/RxJava/javadoc/) under lighttable with clojure 1.5.1, consider the following (using a side-effecting, mutating (horrors!) external collector to extract data from the monad)

(ns expt1.core (:import [rx]))

(def ^:private collector (atom []))
(defn- collect [item]
  (reset! collector (conj @collector item)))

(def string-explode (partial map identity))

(reset! collector [])
(->
 (Observable/toObservable ["one" "two" "three"])
 (.mapMany #(Observable/toObservable (string-explode %)))
 (.subscribe collect)
 )

@collector

which produces:

[\o \n \e \t \w \o \t \h \r \e \e]

I would like to say

(reset! collector [])
(->
 (Observable/toObservable ["one" "two" "three"])
 (.mapMany (comp Observable/toObservable string-explode))
 (.subscribe collect)
 )

but, surprise

clojure.lang.Compiler$CompilerException: java.lang.RuntimeException: Unable to find static field: toObservable in class rx.Observable, compiling:(NO_SOURCE_PATH:93:1)
             Compiler.java:6380 clojure.lang.Compiler.analyze
             Compiler.java:6322 clojure.lang.Compiler.analyze
             Compiler.java:3624 clojure.lang.Compiler$InvokeExpr.parse
             Compiler.java:6562 clojure.lang.Compiler.analyzeSeq
             ...

in fact, just

Observable/toObservable

produces a similar exception. Why can Clojure find Observable/toObservable when it's in an expression like

#(Observable/toObservable (string-explode %))

but not in an expression like

(comp Observable/toObservable string-explode)

?

Answer 1

Observable/toObservable is a Java method, not a Clojure function. You can't treat Java methods as function values, like you can do with Clojure functions. comp composes Clojure functions (objects that implement the IFn interface). Solution: wrap the method call in a Clojure function.

One other thing:

(defn- collect [item]
  (reset! collector (conj @collector item)))

should be:

(defn- collect [item]
  (swap! collector conj item)))

Only use reset! on an atom when you don't want to use the old value. When you do, use swap!, else an atomic update is not guaranteed (another thread could update the atom after you read its value).