将此Clojure调用转换为惰性序列

Question

我正在使用消息传递工具包(它恰好是Spread,但我不知道细节很重要).从此工具包接收消息需要一些样板:

1. 创建与守护程序的连接.
2. 加入一个小组.
3. 接收一条或多条消息.
4. 离开小组.
5. 断开与守护程序的连接.

根据我在别处看过的一些习惯用法,我能够使用Spread的Java API和Clojure的互操作形式来编写一些工作函数:

(defn connect-to-daemon
  "Open a connection"
  [daemon-spec]
  (let [connection (SpreadConnection.)
        {:keys [host port user]} daemon-spec]
    (doto connection
      (.connect (InetAddress/getByName host) port user false false))))

(defn join-group
  "Join a group on a connection"
  [cxn group-name]
  (doto (SpreadGroup.)
    (.join cxn group-name)))

(defn with-daemon*
  "Execute a function with a connection to the specified daemon"
  [daemon-spec func]
  (let [daemon (merge *spread-daemon* daemon-spec)
        cxn (connect-to-daemon daemon-spec)]
    (try
     (binding [*spread-daemon* (assoc daemon :connection cxn)]
       (func))
     (finally
      (.disconnect cxn)))))

(defn with-group*
  "Execute a function while joined to a group"
  [group-name func]
  (let [cxn (:connection *spread-daemon*)
        grp (join-group cxn group-name)]
    (try
     (binding [*spread-group* grp]
       (func))
     (finally
      (.leave grp)))))

(defn receive-message
  "Receive a single message. If none are available, this will block indefinitely."
  []
  (let [cxn (:connection *spread-daemon*)]
    (.receive cxn)))

(基本上是同一个成语with-open,只是SpreadConnection该类使用disconnect而不是close.Grr.另外,我遗漏了一些与结构问题无关的宏.)

这很好用.我可以从结构内部调用receive-message,如:

(with-daemon {:host "localhost" :port 4803}
  (with-group "aGroup"
    (... looping ...
      (let [msg (receive-message)] 
        ...))))

receive-message如果它是一个产生消息的无限懒惰序列,那么我会发现它会更清晰.所以,如果我想加入一个组并获取消息,那么调用代码应该类似于:

(def message-seq (messages-from {:host "localhost" :port 4803} "aGroup"))
(take 5 message-seq)

我已经看过很多没有清理的懒惰序列的例子,这不是太难.捕获是从上面的步骤#4和5:离开组并从守护进程断开连接.如何在不再需要序列时将连接状态和组绑定到序列中并运行必要的清理代码？

Answer 1

该文章介绍了如何做到这些采用的Clojure-contrib请填充队列.关于清理 - 关于填充队列的一个好处就是你可以提供一个阻塞函数,如果出现错误或某些条件,它会自行清理.您还可以保留对资源的引用以在外部控制它.序列将终止.因此,根据您的语义要求,您必须选择适合的策略.