stream.clj

(defn paged-http-get
  ([url]
   (paged-http-get url 1))
  ([url page]
   (let [sink (s/stream)]
     (d/loop [page page]
       (d/chain (http-get url {:query-params {"page" page}})
                (fn [response]
                  (let [body (:body response)]
                    (if (seq body)
                      (s/put-all! sink body)
                      (s/close! sink))))
                (fn [puts]
                  (when puts
                    (d/recur (inc page))))))
     sink)))

(defn stream-take-count [limit s]
  (let [s' (stream/stream)
        count (atom 0)]
    (stream/connect-via s
                        (fn [msg]
                          (if (>= @count limit)
                            (do
                              (stream/close! s')
                              (d/success-deferred false))
                            (do
                              (swap! count inc)
                              (stream/put! s' msg))))
                        s')
    (stream/source-only s')))

(let [s (paged-http-get "https://api.github.com/orgs/facebook/repos")]
  (->> s
    (stream/map :name)
    (stream-take-count 10)
    (stream/stream->seq)))