From e924e784babad65ab902feec049d7b9bbffd12ff Mon Sep 17 00:00:00 2001
From: zalky <zalan.k@gmail.com>
Date: Thu, 6 Apr 2023 10:17:47 -0400
Subject: [PATCH] Fix #2 Ensure all watches removed after non-blocking alts
 read

Fixes a minor alts performance penalty in cases where tailers are
repeatedly being created and destroyed. Does not affect behaviour of
queues and tailers otherwise.
---
 src/cues/queue.clj | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/src/cues/queue.clj b/src/cues/queue.clj
index 01e81af..1c3ebb1 100644
--- a/src/cues/queue.clj
+++ b/src/cues/queue.clj
@@ -539,12 +539,12 @@
 
 (defn- alts
   [tailers]
-  (reduce
-   (fn [_ t]
+  (some
+   (fn [t]
      (when-let [msg (read t)]
-       (rm-watches t)
-       (reduced [t msg])))
-   nil
+       (doseq [t* tailers]
+         (rm-watches t*))
+       [t msg]))
    tailers))
 
 (defn alts!!
@@ -1651,10 +1651,10 @@
    (send! graph nil msg))
   ([{p      :processors
      config :config} source msg]
-   (let [id        (or source (:source config))
-         {:keys [type impl]
-          :as   s} (get p id)]
-     (if (and s (= type ::source))
+   (let [id           (or source (:source config))
+         {impl :impl
+          type :type} (get p id)]
+     (if (= type ::source)
        (write (:appender @impl) msg)
        (throw (ex-info "Could not find source" {:id id}))))))