timeout-channel.clj

(defn timeout-channel
  "Creates a channel and a go block that takes from it. The go block keeps 
an internal status with two possible values, `:wait` and `:receive`. 
In ':wait' status, execution is blocked until there's a value available in the
channel, it then enters the ':receive' status, until the timeout wins. 
Returns the channel where events need to be pushed."
  [timeout-ms f]
  (let [c (async/chan)]
    (async/go-loop [status   :wait
                    args     nil]
      (condp = status
        :wait
          (recur :receive (async/<! c))
        :receive
          (let [[_ ch] (async/alts! [c (async/timeout timeout-ms)])]
            (if (= ch c)
              (recur :receive args)
              (do
                (async/thread (if (sequential? args) (apply f args) (f args)))
                (recur :wait nil))))))
    c))