db.clj

(ns timeline.db
  (:import org.bson.types.ObjectId)
  (:require [ monger.core                    :as mg ]
            [ monger.collection              :as mc ]
            [ com.palletops.leaven           :as leaven ]
            [ taoensso.timbre                :as timbre ]
            [ clojure.core.async             :refer (go-loop go chan alts! put! <! <!! >! close! dropping-buffer) ]
            [ com.palletops.leaven.protocols :refer (Startable Stoppable) ]
            [ timeline.macros                :refer (alt-over) ]))

(timbre/refer-timbre)

(defn oid
  ([ id ] (ObjectId. id))
  ([] (ObjectId.)))

(defmulti db-action (fn [ action & _ ] action))

(defmethod db-action :find-one [ _ db [ collection query fields ] ]
  (mc/find-one-as-map db collection (or query {}) (or fields {})))

(defmethod db-action :find-all [ _ db [ collection query fields ] ]
  (mc/find-maps db collection (or query {}) (or fields {})))

(defmethod db-action :insert [ _ db [ collection document ] ]
  (mc/insert-and-return db collection
   (if-not (:_id document)
     (assoc document :_id (oid))
     document)))

(defrecord Db [ req-ch db conn ]
  Startable
  (start [ self ]
    (let [ { :keys [ db conn ] } (mg/connect-via-uri "mongodb://localhost/timeline")
           ctrl-ch (chan)
           stop-fn #(close! ctrl-ch)]
      (alt-over [ ctrl-ch req-ch ] [ [ action & params ] respond? :as msg ]
        (try
          (let [ result (db-action action db params)]
            (if respond?
              (put! respond? result)))
          (catch Exception e
            (error e))))

      (assoc self
        :db      db
        :conn    conn
        :stop-fn stop-fn)))

  Stoppable
  (stop  [ self ]
    (prn "Closing connection to database.")
    ((:stop-fn self))
    (mg/disconnect (:conn self))
    (assoc self
      :req-ch nil
      :ctrl-ch nil)))


(defn make-db [ req-ch ]
  (map->Db { :req-ch req-ch }))

(def ch (chan))
(def db (leaven/start (make-db ch)))
(def out-ch (chan))
; (leaven/stop db)

(defn putback! [ out-ch value ]
  (let [ ret-ch (chan (dropping-buffer 1)) ]
    (put! out-ch [ value ret-ch ])
    ret-ch))


(go 
  (prn (<! (putback! ch [ :insert :messages { :text "Hello" } ]))))

(go
  (prn (<! (putback! ch [ :find-all :messages ]))))