Fixing issue where echo server won't shut down and start up again

1 files changed, 187 insertions, 51 deletions

diff --git a/src/midi.clj b/src/midi.clj
index bf7d03d..57e7705 100644
--- a/src/midi.clj
+++ b/src/midi.clj
@@ -1,7 +1,8 @@
 (ns midi
   (:require [missionary.core :as m])
   (:import [javax.sound.midi MidiSystem Receiver ShortMessage MidiDevice$Info MidiDevice Transmitter MidiMessage]
-           [uk.co.xfactorylibrarians.coremidi4j CoreMidiDeviceProvider]))
+           [uk.co.xfactorylibrarians.coremidi4j CoreMidiDeviceProvider]
+           [missionary Cancelled]))
 
 (defn get-all-midi-device-info []
   (CoreMidiDeviceProvider/getMidiDeviceInfo))
@@ -25,13 +26,22 @@
   [^MidiDevice device tfn]
   (m/sp
    (try
-     (m/? (m/via m/blk  (.open device)))
-     (println "Device opened")
+     (.open device)
+     #_
+     (m/? (m/via m/blk (.open device)))
+     (println "Device opened" device)
      (m/? (tfn device))
-     (catch Exception e (println "E" e) (throw e))
+     (catch Cancelled e (println "with-device received cancellation signal") (throw e))
+     (catch Exception e (println "with-device received exception") (throw e))
      (finally
+       ;; HERE:
+       ;; For some reason, closing device is hanging.
+       (println "Closing device...")
+       (.close ^MidiDevice device)
+       (println "Device closed.")
+       #_
        (m/? (m/via m/blk (.close device)))
-       (println "Device closed")))))
+       (println "Device closed" device)))))
 
 (defn with-tx
   "Feed a transmitter device (e.g. a MIDI keyboard) into a consumer `t`.
@@ -46,30 +56,60 @@
   Returns a task."
   [^MidiDevice device t]
   (m/sp
-   (let [^Transmitter transmitter (m/? (m/via m/blk (.getTransmitter device)))
+   (let [^Transmitter transmitter
+          (.getTransmitter device)
+         #_
+         (m/? (m/via m/blk (.getTransmitter device)))
          _ (println "tx mounted")
-         rv (m/rdv)
+         rv (m/mbx)
          receiver
          (reify Receiver
            (send [_this midi-message _timestamp]
              (println "Sending message")
-             (m/? (rv midi-message))
+             ;; NOTE: RV CANNOT BLOCK!
+             ;; If it does, you can end up with a deadlock during cancellation.
+              (rv midi-message)
              (println "Message sent"))
            (close [_this]))]
-     (println "RX object defined" receiver)
+     (println "RX object defined:" receiver)
      (try
        (println "Setting receiver...")
+       (.setReceiver transmitter receiver)
+       #_
        (m/? (m/via m/blk (.setReceiver transmitter receiver)))
        (println "Receiver set")
-       (m/?
-        (t (m/ap
-            (loop []
-              (m/amb (let [v (m/? rv)]
-                       (println "tx received value" v)
-                       v)
-                     (recur))))))
+       (try
+         (m/?
+          (t (m/ap
+               (m/amb=
+                ;; TODO This might not be necessary
+                (m/? m/never)
+                (loop []
+                  (m/amb
+                   (do
+                     (println "tx awaiting value from device...")
+                     (m/amb))
+
+                   ;; TODO:
+                   ;; As far as I can tell, rx is closing but tx is not.
+                   ;; I think we're hanging on the take here
+                   ;; Probably need to raise an exception somehow
+
+                   ;; Looks like my m/amb= calls are good?
+                   (let [v (m/? rv)]
+                     (println "tx received value" v)
+                     v)
+                   (recur)))))))
+         (catch Cancelled e (println "cancellation signal receved prior to loop exit") (throw e))
+         (catch Exception e (println "exception receved prior to loop exit") (throw e))
+         (finally
+           (println "Loop exited")))
+       (catch Cancelled e (println "cancellation signal receved prior to closing tx") (throw e))
+       (catch Exception e (println "exception receved prior to closing tx") (throw e))
        (finally
          (println "Closing tx")
+         (.close transmitter)
+         #_
          (m/? (m/via m/blk (.close transmitter)))
          (println "Tx closed"))))))
 
@@ -91,20 +131,35 @@
     ;; TODO Try finally
     ;; TODO blocking APIs everywhere
     ;; TODO mailbox -> rdv
-   (let [^Receiver receiver (m/? (m/via m/blk (.getReceiver device)))]
+   (let [^Receiver receiver
+         (.getReceiver device)
+         #_
+         (m/? (m/via m/blk (.getReceiver device)))]
      (println "receiver mounted")
      (try
        (m/?
         (m/reduce {} nil
                   (m/ap
-                   (let [v (m/?< f)]
-                     (println "rx received value" v)
-                     (.send receiver v UNSCHEDULED-EVENT)))))
+                    (try
+                      (let [^MidiMessage v (m/?< f)]
+                        (println "rx received value" v)
+                        (.send receiver v UNSCHEDULED-EVENT)
+                        #_
+                        (m/? (m/via m/blk (.send receiver v UNSCHEDULED-EVENT)))
+                        (println ".send returned"))
+                      (catch Cancelled e (println "cancellation signal receved prior to rx flow exit") (throw e))
+                      (catch Exception e (println "exception receved prior to rx flow exit") (throw e))
+                      (finally (println "RX flow exited"))))))
+       (catch Cancelled e (println "cancellation signal receved prior to rx closure") (throw e))
+       (catch Exception e (println "exception receved prior to rx closure") (throw e))
        (finally
          (println "Closing rx")
+         (.close receiver)
+         #_
          (m/? (m/via m/blk (.close receiver)))
          (println "rx closed"))))))
 
+
 (defn echo [tx-device rx-device]
   (m/sp
    (let [rv (m/rdv)]
@@ -113,48 +168,68 @@
               (with-device tx-device
                 (fn [d]
                   (println "Have tx device")
-                  (with-tx d
-                    (fn [f]
-                      (m/sp
-                       (println "Have transmitter")
-                       (m/?
-                        (m/reduce prn nil
-                                  (m/ap
-                                   (let [^ShortMessage v (m/?< f)
-                                         _ (println v)
-                                         new-msg (ShortMessage. ShortMessage/NOTE_ON 9 (.getData1 v) (.getData2 v))]
-                                     (println "Echo received value from flow:" v)
-                                     (println "Echo sending v to rdv.")
-                                     (m/?
-                                      (rv new-msg))
-                                     (println "Value sent."))))))))))
+                  (m/sp
+                    (try
+                      (m/?
+                       (with-tx d
+                         (fn [f]
+                           (m/sp
+                             (println "Have transmitter")
+                             (m/?
+                              (m/reduce prn nil
+                                        (m/ap
+                                          (let [^MidiMessage v (m/?< f)
+                                                _ (println "Echo received value from flow:" v)
+                                                new-msg (ShortMessage. ShortMessage/NOTE_ON 9 (.getData1 ^ShortMessage v) (.getData2 ^ShortMessage v))]
+                                            (println "sending v to rdv.")
+                                            (m/?
+                                             (rv new-msg))
+                                            (println "Value sent.")))))))))
+                      (catch Cancelled e (println "cancellation signal receved prior transmitter completion") (throw e))
+                      (catch Exception e (println "exception receved prior transmitter completion") (throw e))
+                      (finally
+                        (println "Done with transmitter"))))))
               (with-device rx-device
                 (fn [d]
                   (println "Have rx device")
-                  (with-rx d
-                    (m/ap
-                     (loop []
-                       (println "Echo rx awaiting value...")
-                       (m/amb
-                        (let [v (m/? rv)]
-                          (println "Echo received value:" v)
-                          v)
-                          ;; I think that what is happening here is a deadlock on either give or take of the rdv
-                        (recur))))))))))))
-
+                  (m/sp
+                    (try
+                      (m/?
+                       (with-rx d
+                         (m/ap
+                           (m/amb=
+                            (m/? m/never)
+                            (loop []
+                              (println "Echo rx awaiting value...")
+                              (m/amb
+                               (let [v (m/? rv)]
+                                 (println "Echo received value:" v)
+                                 v)
+                               ;; I think that what is happening here is a deadlock on either give or take of the rdv
+                               (recur)))))))
+                      (catch Cancelled e (println "cancellation signal receved prior receiver completion") (throw e))
+                      (catch Exception e (println "exception receved prior receiver completion") (throw e))
+                      (finally
+                        (println "Done with receiver")))))))))))
 (def run
   (m/sp
-   (let [tx
+   (let [txd
          (first
           (select-devices (get-all-midi-device-info)
                           "CoreMIDI4J - IAC Bus" true false))
-         rx
+
+         rxd
          (first
           (select-devices (get-all-midi-device-info)
                           "CoreMIDI4J - IAC Bus" false true))]
-     (def t tx)
-     (def r rx)
-     (m/?  (echo tx rx)))))
+
+     (def txd txd)
+     (def rxd rxd)
+
+     (m/? (echo txd rxd)))))
+
+;; CoreMidiSource is TX Device
+;; CoreMidiDestination is RX Device
 
 (def close (run prn prn))
 (close)
@@ -195,3 +270,64 @@
 ;; Here, `sigs` is a map of {ch val}, where `ch` is a midi channel number and
 ;; `val` is a signal of sets representing active notes on that channel.
 ;;
+
+(let [_ (println "getting device")
+      txd (first
+              (select-devices (get-all-midi-device-info)
+                              "CoreMIDI4J - IAC Bus" true false))
+      rxd (first
+              (select-devices (get-all-midi-device-info)
+                              "CoreMIDI4J - IAC Bus" false true)) 
+      _ (println "getting transmitter")
+      transmitter (.getTransmitter txd)
+      _ (.setReceiver transmitter
+                      (reify Receiver
+                        (send [_this midi-message _timestamp]
+                          (println "Message received:")
+                          (println midi-message))
+                        (close [_this] (println "Closing!"))))
+      receiver (.getReceiver rxd)]
+  (println "opening txd")
+  (.open txd)
+  (println "Opening rxd")
+  (.open rxd)
+  (println "setting receiver")
+  (def txd txd)
+  (def rxd rxd)
+  (def t transmitter)
+  (def r receiver))
+
+(.send r (ShortMessage. ShortMessage/NOTE_ON 0 0 0) -1)
+
+(.close txd)
+(.close rxd)
+(.close t)
+(.close r)
+
+(def t
+  (m/sp
+    (try
+      (println "A")
+      (m/? (m/sp
+             (try
+               (println "AA")
+               (m/?  (m/sleep 10000000))
+               (println "BB")
+               (catch missionary.Cancelled e
+                 (println "Catch Inner") (throw e))
+               (catch Exception e
+                 (println "E" e)
+                 (println "Exception Inner") (throw e))
+               (finally
+                 (println "Finally Inner")))))
+      (catch missionary.Cancelled e
+        (println "Catch Outer") (throw e))
+      (catch Exception e
+        (println "Exception Outer") (throw e))
+      (finally
+        (println "Finally Outer")))
+    (println "B")))
+
+(def cancel (t {} {}))
+
+(cancel)