Version of map handling deferreds/errors

gistfile1.clj

(ns clj-amp.core
  (:require [manifold.deferred :as d]
            [manifold.stream :as s]))
(defn- map'
  "Like manifold.stream/map, except handles deferreds and closes the stream if an error occurs."
  ([f s]
   (let [s' (s/stream)]
     (s/connect-via
      s
      (fn [msg]
        (-> msg
            (d/chain
             f
             #(s/put! s' %))
            (d/catch
              (fn [e]
                (s/close! s')
                e))))
      s'
      {:description {:op "map"}})
     (s/source-only s')))
  ([f s & rest]
   (map' #(apply f %)
         (apply s/zip s rest))))

Pasted an old buggy version of the code, updated with the proper error-close logic.