如何使用core.async代替回调？

Question

我在ClojureScript上使用core.async，以避免使用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时，它会导致通道中的消息顺序混乱。

我发现解决这个问题的唯一方法是在atom中创建一个无限的promise-chan列表，每次回调都会选取第一个元素，发布一条消息，然后删除该列表的第一个元素。然后，我们可以在go块中有一个doseq，它将为我们发布消息：

(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块中处理chan