将此Clojure调用转换为惰性序列
-
06-07-2019 - |
题
我正在使用消息传递工具包(恰好是 Spread 但我不知道细节问题)。从此工具包接收消息需要一些样板:
- 创建与守护程序的连接。
- 加入小组。
- 接收一条或多条消息。
- 离开小组。
- 与守护程序断开连接。 醇>
根据我见过的一些习惯用法其他地方,我能够做饭使用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
。格尔。此外,我遗漏了一些与结构问题无关的宏。)
这很好用。我可以从结构内部调用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:离开组并从守护进程断开连接。当不再需要序列时,如何将连接和组的状态绑定到序列和运行必要的清理代码?
解决方案
这篇文章描述如何使用clojure-contrib fill-queue做到这一点。关于清理 - 关于填充队列的一个好处就是你可以提供一个阻塞函数,如果出现错误或某些条件,它会自行清理。您还可以保留对资源的引用以在外部控制它。序列将终止。因此,根据您的语义要求,您必须选择适合的策略。
其他提示
试试这个:
(ns your-namespace
(:use clojure.contrib.seq-utils))
(defn messages-from [daemon-spec group-name]
(let [cnx (connect-to-deamon daemon-spec))
group (connect-to-group cnx group-name)]
(fill-queue (fn [fill]
(if done?
(do
(.leave group)
(.disconnect cnx)
(throw (RuntimeException. "Finished messages"))
(fill (.receive cnx))))))
设置完成?要终止列表时为true。此外,(.receive cnx)中抛出的任何异常也将终止列表。
不隶属于 StackOverflow