ended up with this, probably awful for some reason 😛

(defn
  <throttle-by-topic
  [<in topic-fn >out]
  (let [topic->c (atom {})
        ensure-sub (fn
                     [topic]
                     (or
                      (@topic->c topic)
                      (let [c1 (a/chan (a/sliding-buffer 1))
                            c (<throttle c1 2000)]
                        (a/pipe c >out)
                        (get
                         (swap! topic->c assoc topic c1)
                         topic))))]
    (a/go-loop
        [m (a/<! <in)]
        (when-let
            [topic (and m (topic-fn m))]
            (a/>! (ensure-sub topic) m)
            (recur (a/<! <in))))
    >out))

i started thinking about your problem, then i started reading this post https://github.com/ptaoussanis/sente/issues/124

i belive zach built a lib around the idea of windowing and backpressure bc he couldn't make core async work the way he needed.

https://github.com/clj-commons/manifold/blob/master/doc/stream.md talks about throttling but i didnt look at the implementation at all.

gonna check 👀