aleph-with-async.clj

(require '[manifold.deferred :as d])
(require '[manifold.stream :as s])
(require '[aleph.http :as http])
(require '[clojure.core.async :as a])
(require '[byte-streams :as bs])

(def oauth-token "asdfasdfsadfasdfasdfsadfsad")

(defn get-stream []
  (http/request
    {:method      :get
     :url         "https://nakadi.somewhere/event-types/an_event/events"
     ;:content-type :json
     ;:as           :json
     :raw-stream? true
     :headers     {"Authorization" (str "Bearer " oauth-token)}}))

(defn do-call
  "Processes incoming buffers from Netty in a clojure.async/go.
   s/connect is used to convert between from manifold stream to async's channel."
  []
  (let [a @(d/chain
             (http/get "https://zalando.de"
               {:raw-stream? true
                :headers     {"Authorization" (str "Bearer " oauth-token)}})
             :body
             #(s/map (comp bs/to-string
                       bs/to-byte-array) %))
        c (a/chan)]
    (s/connect a c)
    (a/go-loop [v (a/<! c)
                i 4]
      (println v)
      (if (and v (< 1 i))
        (recur (a/<! c) (dec i))
        (s/close! a)))))

line-splitter.clj

(require '[clojure.string :as cs])
(let [b "asdfasdfasdf\nfhgdfhgfdhgdfhgdf\n\n"]
  (loop [p 0]
    (if-let [nlp (cs/index-of b \newline p)]
      (let [l (subs b p nlp)]
        (println "LN" (count l) ":" (pr-str l))
        (recur (inc nlp))))))