Erstellen Sie Kanäle, die mehrmals geschlossen werden müssen

Question

Meine Situation ist, dass der gleiche Kanal zwischen verschiedenen Funktionen geteilt wird und ich den Kanal schließen muss, wenn alle diese Funktionen signalisieren, dass sie damit erledigt sind. Das ist, was ich mir ausgedacht habe, aber es ist nicht ideal, weil ich Schleppkännchen behandeln muss und ein spezielles Schlüsselwort senden muss, anstatt das Schließen zu verwenden! Funktion. Gibt es einen anderen, besseren Weg, dies zu tun? Gibt es dafür eine Funktion von core.async?

(defn shared-chan [n] 
    (let [in (chan) 
     out (chan)] 
    (go-loop [n n] 
     (if (= n 0) 
     (do 
      (async/close! in) 
      (async/close! out)) 
     (let [in-msg (<! in)] 
      (if (not= :close in-msg) 
      (do 
       (>! out in-msg) 
       (recur n)) 
      (recur (dec n)))))) 
    [in out])) 

Answer 1

merge kann Ihnen helfen. Gemäß der doc, merge

nimmt eine Sammlung von Quellenkanälen und gibt einen Kanal, der alle Werte von ihnen genommen enthält. ... Der Kanal wird geschlossen, nachdem alle Quellkanäle geschlossen wurden.

Also, im Grunde müssen Sie einen Kanal für jede Ihrer Funktionen erstellen und dann merge sie in eins.

Answer 2

Sie können diesen geteilten Kanal mit take Transducer erzeugen. Zum Beispiel, wenn Sie einen Auto-verschließbaren Kanal für 3 Einzelteile wollen:

user> (require '[clojure.core.async :as a]) 
nil 

user> (def shared-ch (a/chan 3 (take 3))) 
#'user/shared-ch 

user> (a/go-loop [] 
     (when-let [val (a/<! shared-ch)] 
      (println :reading val) 
      (recur))) 

#object[clojure.core.async.impl.channels.ManyToManyChannel 0x6bb1dee5 "clojure.core.async.impl.channels.ManyToManyChannel@6bb1dee5"] 

user> (a/>!! shared-ch 1) 
true:reading 1 

user> (a/>!! shared-ch 2) 
true:reading 2 

user> (a/>!! shared-ch 3) 
true:reading 3 

user> (a/>!! shared-ch 4) 
false 

user> (a/>!! shared-ch 5) 
false 

Sie sehen können, dass der Kanal so schnell geschlossen wurde als der Wandler erschöpft ist (wie false nach dem Schreibversuch zurückgeführt wird)