Throttle-first operator which ignores the following events with same key within `duration`

throttle-first.clj

(defn throttle-first [duration keyF]
  (operator*
    (fn [s]
      (let [cache (atom #{})]
        (subscriber
          s
          (fn [s v]
            (let [k (keyF v)]
              (when-not (@cache k)
                (on-next s v)
                (swap! cache conj k)
                (set-timeout #(swap! cache disj k) duration)))))))))