Use Datomic queries as a source for RxJava pipes.

query.clj

(ns rxjava-datomic.query
  (:require [datomic.api :as d])
  (:use     [clojure.repl :only [pst]])
  (:import  [rx Observable]
            datomic.Peer))

(defn query [qry & inputs]
  (Observable/toObservable
   (Peer/q qry (object-array inputs))))

(comment

  (def uri "datomic:mem://seattle")
  (Peer/createDatabase uri)
  (def conn (Peer/connect uri))
  
  (.transact conn (read-string (slurp "samples/seattle/seattle-schema.dtm")))
  (.transact conn (read-string (slurp "samples/seattle/seattle-data0.dtm")))

  (d/q '[:find ?c :where [?c :community/name]] (d/db conn))

  (def results (d/q '[:find ?c :where [?c :community/name]] (d/db conn)))
  (count results)

  (:community/name (.entity (.db conn) (ffirst results)))

  ;;; Simplest version
  (-> (query '[:find ?c ?n :where [?c :community/name ?n]] (.db conn))
      (.subscribe (fn [[eid n]] (println (str "Hello " n "!")))))

  ;;; With error notification
  (-> (query '[:find ?c ?n :where [?c :community/name ?n]] (d/db conn))
      (.subscribe (fn [[eid n]] (println (str "Hello " n "!")))
                  pst))

  ;;; With error and completion notification
  (-> (query '[:find ?c ?n :where [?c :community/name ?n]] (d/db conn))
      (.subscribe (fn [[eid n]] (println (str "Hello " n "!")))
                  pst
                  #(println "---- Done ----")))

  )

Neat, thanks for sharing! It might be neater if you made a tx-report-queue-backed Observable. It might be neater still if you made Observable foldable such that the resulting "event stream of database values over time" could be manipulated with reducer functions instead of RxJava's type-specific pseudo-seq methods.