Когда канал будет отброшен, если поток продолжает его использовать?

Я не знаю, есть ли такая вещь, как лучшая практика для этого конкретного случая. Вот мое решение проблемы. Я думаю, это довольно просто.

(defn alts-while-open [f & chans]
   (let [a-chans (atom (set chans))]
     (go (while (< 0 (count @a-chans))
           (println "iteration started : " (vec @a-chans))
           (let [[v ch] (alts! (vec @a-chans))]
             (if v
               (f v ch)
               (swap! a-chans #(disj % ch))))))))

Функция f выполняется по результату alts! Пока каналы открыты. Держу здесь атом с множеством открытых каналов. Как только я нахожу закрытый канал, я удаляю его из этого набора. Если открытых каналов больше нет, цикл while останавливается. Вы можете запустить его:

(def c1 (chan))
(def c2 (chan))
(def c3 (chan))
(alts-while-open (fn [v ch] (println v)) c1 c2 c3)

Теперь, когда что-то записывается в любой из этих каналов, оно распечатывается. Вы можете видеть, что после этого начинается новая итерация. Как только вы закроете канал, вы увидите, что итерация началась, но вектор каналов уменьшился. Как только все каналы закрыты, цикл while останавливается.

Трудно ответить на вопрос, использовать ли close! функция или какой-либо другой механизм уведомления, чтобы остановить цикл. Я думаю, это зависит от ситуации. Если бы не сложная обработка «конца связи», я бы просто закрыл! канал. Если есть более сложная логика - например. есть успешные варианты «конец связи» и неудачный «конец связи», и я хочу обрабатывать их по-разному, тогда я бы предпочел отправить специальное сообщение. Что-то вроде

[:end-of-communication :success]

Идея с отправкой специального end-of-communication не работает, когда вы не отправляете простые значения, потому что вы не можете гарантировать, что значения будут помещаться и приниматься в том порядке, в котором вы их ожидаете.

Следующая идея заключается в том, что отправитель и получатель заранее знают количество значений для обработки, например. как это:

user> (<!!
        (let [c1 (chan)
              values ["hi" "there"]
              vcount (count values)]
           (doseq [value values]
             (thread
                  (>!! c1 value)))
           (thread
               (loop [recvalue (<!! c1)
                      reccount 1]
                  (println "Read" recvalue)
                  (if (= reccount vcount)
                      (do (close! c1)
                          :done)
                      (recur (<!! c1) (inc reccount)))))))
Read hi
Read there
:done

Это работает в принципе, но имеет очевидный недостаток, заключающийся в том, что вы должны согласовать сумму перед настройкой процессов отправки и получения, и менее очевидный недостаток, заключающийся в том, что в случае, если что-то пойдет не так на стороне отправки, вы в конечном итоге будете бесконечно ждать. снова на принимающей стороне (при условии, что отправляющая сторона делает больше, чем просто помещает значения в канал).

Я пришел к выводу, что единственный способ сделать это надежным — использовать канал timeout, например:

(<!! (let [c1 (chan)
           tchan (timeout 1000) 
           values ["hi" "there"]]
       (doseq [value values]
         (thread
           (>!! c1 value)))
       (thread
          (loop [[recvalue rchan] (alts!! [c1 tchan])
                 timeoutchan tchan]
            (if (= rchan timeoutchan)
                (do (close! c1)
                    :done)
                (do (println "Read" recvalue)
                    (let [newtimeout (timeout 1000)]
                        (recur (alts!! [c1 newtimeout])
                               newtimeout))))))

Read hi
Read there
:done