From 42b1a01f1165ce04cd2addbc44da9e597fcf4ea7 Mon Sep 17 00:00:00 2001
From: Jake Zerrer <him@jakezerrer.com>
Date: Wed, 12 Nov 2025 16:03:18 -0500
Subject: Fix undefined continuous flow

---
 src/unheard/midi.clj | 45 ++++++++++++++++++++++++---------------------
 1 file changed, 24 insertions(+), 21 deletions(-)

diff --git a/src/unheard/midi.clj b/src/unheard/midi.clj
index a44c01b..c3759fe 100644
--- a/src/unheard/midi.clj
+++ b/src/unheard/midi.clj
@@ -175,13 +175,14 @@
   Calls `flow-handler` with a flow of midi messages.
 
   `flow-handler` should return a flow."
-  [>name flow-handler]
+  [name flow-handler]
   (m/ap
    (try
      (let [device
            (first
             (select-devices (get-all-midi-device-info)
-                            (m/?< >name) true false))]
+                            name
+                            true false))]
        (if device
          (m/?
           (with-device device
@@ -283,25 +284,27 @@
 (defn short-messages
   [>device-name]
   (m/stream
-   (m/ap
-    (let [short-messages (atom nil)
-          >short-messages (m/watch short-messages)]
-      (m/amb=
-       (do (reset! short-messages nil)
-           (m/?
-            (<bus >device-name
-                  (fn [v]
-                    (m/ap
-                     (try (let [msg (m/?< v)]
-                            (reset! short-messages msg))
-                          (catch missionary.Cancelled c
-                          ;; When the upstream flow is cancelled, we emit "All notes off" to consumers
-                            (doseq [ch (range 0 16)]
-                              (reset! short-messages (ShortMessage. ShortMessage/CONTROL_CHANGE ch 123 0)))
-                            (throw c))))))))
-       (if-let [m (m/?< >short-messages)]
-         m
-         (m/amb)))))))
+   (m/reductions {} nil
+                 (m/ap
+                  (let [device-name (m/?< >device-name)
+                        short-messages (atom nil)
+                        >short-messages (m/watch short-messages)]
+                    (m/amb=
+                     (do (reset! short-messages nil)
+                         (m/?
+                          (<bus device-name
+                                (fn [v]
+                                  (m/ap
+                                   (try (let [msg (m/?< v)]
+                                          (reset! short-messages msg))
+                                        (catch missionary.Cancelled c
+                                           ;; When the upstream flow is cancelled, we emit "All notes off" to consumers
+                                          (doseq [ch (range 0 16)]
+                                            (reset! short-messages (ShortMessage. ShortMessage/CONTROL_CHANGE ch 123 0)))
+                                          (throw c))))))))
+                     (if-let [m (m/?< >short-messages)]
+                       m
+                       (m/amb))))))))
 
 (defn notes [short-messages]
   (m/signal
-- 
cgit v1.2.3