*.clj

(ns aleph_test.core
  (:require [aleph.http :as http]
            [byte-streams :as bs]
            [manifold.deferred :as d]
            [manifold.stream :as s]
            [cheshire.core :as json]
            [clojure.core.async :as async]
            [clojure.tools.logging :as log]))

(defn query-url
  [url]
  (let [options {:timeout 60000
                 :read-timeout 30000
                 :raw-stream? true
                 :throw-exceptions false
                 :ignore-unknown-host? true}
        r (http/get url options)
        status (:status @r)]
      (log/info "Received status: " status)
      (when (= status 200)
        (let [stream (d/chain r
                        :body
                        #(s/map bs/to-string %)
                        #(s/map json/parse-string %)
                        #(s/filter some? %))
              updates (async/chan 5)]
          (s/connect @stream updates)
          updates))))

(defn listen-to-login-events
  []
  (log/info "Sending the request")
  (let [url "http://some-url:8080/events"
        updates (query-url url)]
    (async/go
      (if (some? updates)
        (do
          (log/info "Request is sent, dealing with chunked response now")
          (loop []
            (if-let [data (async/<! updates)]
              (do
                (log/info "Got an update: " data)
                (recur))
              (log/info "Channel was closed, stop receiving updates"))))
        (log/info "Could not establish the request")))))


(defn -main
  []
  ; Block till we get some events
  (log/info "Start listening to the event queue")
  (async/<!! (listen-to-login-events))
  (log/info "Done dealing with the stream, exiting"))