 (ns eventstore.streams
-  (:require [clojure.core.async :refer [go-loop go <! >! chan buffer
-                                        sliding-buffer mult tap close!]]
+  (:require [clojure.core.async :refer [go-loop go <!! <! >! chan buffer
+                                        sliding-buffer mult tap close!]
+             :as async]
             [clojure.tools.logging :as log]
+            [somnium.congomongo :as m]
+            [clj-time.coerce :as cc]
             [eventstore.riak :as riak]))
 
 (def cold-latency 5000)

        (>! ch event)))
    ch))

(defn cold-stream [a-stream date-string & args]
  (let [m-args (apply hash-map args)
        s (riak/lazy-events (:db a-stream) date-string)]
    (if (true? (:channel m-args))
      (seq->channel s)
      s)))

(defmulti m-stream (fn [params _] (get params "stream-type" "hot")))

(defmethod m-stream "cold" [params a-stream]
  (let [ch (chan (buffer 1))
        full-s (cold-stream a-stream (get params "from" 0))]
    (go
      (loop [e (first full-s) s (rest full-s) closed? false]
        (if (nil? e)
          (do

                (recur (first s) (rest s) closed?)))))))
    ch))

(defmethod stream "hot-cold" [a-stream params]
  (let [ch (chan (buffer 1))
        full-s (data-from a-stream (get params "from" 0))]
    (go
      (loop [e (first full-s)
             s (rest full-s)
             closed? false
             last-t (System/currentTimeMillis)]
        (let [rest-s (rest s)
              new-s (if (empty? rest-s) (data-from a-stream last-t) s)
              last-t (if (empty? rest-s) (System/currentTimeMillis) last-t)]
          (if (nil? e)
            (log/info ":::::::::::::::::::: Stream depleted, switching to hot stream")
            (if closed?
              (log/info ":::::::::::::::::::: Stream closed by peer, switching to hot stream")
              (let [closed? (not (>! ch e))]
                (recur (first new-s) (rest new-s) closed? last-t))))))
      (tap (:mult-channel a-stream) ch))
    ch))

(defmethod stream "hot" [a-stream params]
  (let [ch (chan 1)]
    (tap (:mult-channel a-stream) ch)
    ch))

#_(def test-ch (chan))
#_(def test-ds (->DummyStream (m/make-connection "eventstore") test-ch (mult test-ch)))


#_(def test-s (stream {"stream-type" "cold"}))
