To just write the stream to file, a simple approach is using clojure.java.io/copy (which takes an input-stream such as that returned by (:body (client/get some-url {:as :stream})) and an output stream and copies from one to the other). Something like
(ns http-stream
(:require [clj-http.client :as client]
[clojure.java.io :as io]))
(with-open [in-stream (:body (client/get "http://hopey.netfonds.no/tradedump.php?date=20150508&paper=AAPL.O&csv_format=txt" {:as :stream}))
out-stream (->> "streamoutput.txt"
io/as-file
io/output-stream)]
(io/copy in-stream out-stream))
That gave me several thousand lines of tab separated values over a couple seconds. Now, to process them with core.async at the level of lines we probably want to process the stream a bit more using a reader and a line-seq:
(ns http-stream
(:require [clj-http.client :as client]
[clojure.core.async :as async]
[clojure.java.io :as io]
[clojure.string :as str]))
(defn trades-chan
"Open the URL as a stream of trades information. Return a channel of the trades, represented as strings."
[dump-url]
(let[lines (-> dump-url
(client/get {:as :stream})
:body
io/reader
line-seq) ];;A lazy seq of each line in the stream.
(async/to-chan lines))) ;;Return a channel which outputs the lines
;;Example: Print the first 250 lines.
(let [a (trades-chan "http://hopey.netfonds.no/tradedump.php?date=20150508&paper=AAPL.O&csv_format=txt")]
(async/go-loop [takes 250]
(when (< 0 takes)
(println (async/<! a))
(recur (dec takes)))))
Now, with this you are largely started up, but I notice that the stream always starts with a description of what the columns are
time price quantity board source buyer seller initiator
and you can use that as a chance to improve a little bit. In particular, that's enough information to build a transducer for the trades-chan that can turn the trades into a more convenient format to work with, like a map. Also, we likely want a way to stop taking elements and close the connection sometime. I'm not that familiar with core.async myself but this seems to work:
(defn trades-chan
"Open the URL as a tab-separated values stream of trades.
Returns a core.async channel of the trades, represented as maps.
Closes the HTTP stream on channel close!"
[dump-url]
(let[stream (-> dump-url
(client/get {:as :stream})
:body)
lines (-> stream
io/reader
line-seq) ;;A lazy seq of each line in the stream.
fields (map keyword (str/split (first lines) #"\t")) ;; (:time :price :quantity ...
transducer (map (comp #(zipmap fields %) #(str/split % #"\t"))) ;;A transducer that splits strings on tab and makes them into maps with keys from fields
output-chan (async/chan 50 transducer)]
(async/go-loop [my-lines (drop 1 lines)]
(if (async/>! output-chan (first my-lines)) ;;If we managed to put
(recur (rest my-lines)) ;;then the chan is not closed. Recur with the rest of the lines.
(.close stream))) ;;else close the HTTP stream.
output-chan))
