Как использовать core.async вместо обратных вызовов?

Question

Я использую core.async на ClojureScript, чтобы избежать использования обратных вызовов node.js.Проблема в том, что я превышаю ограничение на количество ожидающих сообщений в 1024.

Чтобы избежать этого, мне нужно будет отправлять все сообщения на каналы внутри одного блока go.Но это не совсем возможно в core.async, потому что анонимная функция сводит на нет эффект go, поэтому я не могу сделать это:

(go
  (. socket on "data" #(>! chan %)))

Итак, есть ли способ обойти это ограничение?

Answer 1

Чтобы смоделировать ошибку, мы можем «смоделировать» своего рода обратный вызов:

(let [callback (atom nil)
      on-data (fn [fun]
                (reset! callback fun))
      chan (async/chan)]

Если мы попытаемся добавить обратный вызов с помощью (on-data #(async/go (async/put! chan %))), это будет превышать предел.Кроме того, поскольку мы используем async/go, это приведет к тому, что сообщения внутри канала будут не в порядке.

Единственный способ, который я обнаружил, чтобы исправить это, - создать бесконечный список * 1008.* внутри atom и каждый обратный вызов выберет первый элемент, опубликует сообщение и удалит первый для списка.Тогда у нас может быть doseq внутри блока go, который будет публиковать для нас сообщения:

(let [inf-list (atom (map (fn [_] (async/promise-chan)) (range)))]
  ; We need to iterate over inf-list before anything
  (let [lst @inf-list]
    (async/go
     (doseq [c lst]
       (async/>! chan (async/<! c)))))

  (on-data #(do
              (async/put! (first @inf-list) %)
              (swap! inf-list rest))))

Answer 2

Я не уверен, что понимаю, почему он должен быть в одном и том же блоке go.Обычно вы просто делаете:

(. socket on "data" #(go (put! chan %)))

и затем обрабатываете чан в другом блоке go