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 ----")))

  )