I'm using core async to manage a data stream. I need to dispatch incoming datums to one or more "collector" channels. I have a coordinator (a go-loop) that pulls incoming maps off a data channel and determines which collector channels it should to put the datum to. It also needs to manage the creation of those channels as well as creating the collector go-loop that collects the datums into a chunk. Each collector should only live for some configured max collection time. So far, I've implemented it such that the collectors are managing the timeout and putting a value on the main data channel that the coordinator pulls from. See pseudo-ish code below:

(defn create-collector! [collect-ch data-ch]
  (let [timeout-ch (a/timeout max-collection-time)]
    (a/go-loop
     (let [msg (a/! data-ch {:timedout collect-ch})

         :else
         (collect-datum! msg))))))

(defn coordinator [data-ch]
  (a/go-loop [collector-state {}]
    (let [datum (a/! timedout :end)
          (recur (remove-timedout-collector collector-state timedout)))
        (let [{:keys [chs-to-put chs-to-create new-collector-state]} (partition-datum collector-state datum)]
          (doseq [c chs-to-put]
            (a/>! c datum))
          (doseq [c chs-to-create]
            (create-collector! c data-ch)
            (a/>! c datum))
          (recur new-collector-state))))))