5

我一直在使用 clojure，但我不熟悉 twitter-api ( https://github.com/adamwynne/twitter-api ) 所基于的异步 io。

我想收集与给定关键字集匹配的所有推文。例如所有匹配“玛丽玫瑰”的东西（现在在英国流行的东西）。进行流式调用的文档说要执行以下操作：

(ns mynamespace
  (:use
   [twitter.oauth]
   [twitter.callbacks]
   [twitter.callbacks.handlers]
   [twitter.api.streaming])
  (:require
   [clojure.data.json :as json]
   [http.async.client :as ac]
   [clojure.java.io :as io])
  (:import
   (twitter.callbacks.protocols AsyncStreamingCallback)))

(def my-creds (make-oauth-creds *app-consumer-key*
                            *app-consumer-secret*
                            *user-access-token*
                            *user-access-token-secret*))

; supply a callback that only prints the text of the status
(def ^:dynamic 
     *custom-streaming-callback* 
     (AsyncStreamingCallback. (comp println #(:text %) json/read-json #(str %2)) 
                     (comp println response-return-everything)
              exception-print))

(statuses-filter :params {:track "mary rose"}
     :oauth-creds my-creds
     :callbacks *custom-streaming-callback*)

如果我然后做类似的事情：

(def mary (statuses-filter :params {:track "mary rose"}
     :oauth-creds my-creds
     :callbacks *custom-streaming-callback*))

我得到了 http 响应的地图：

(keys mary)
;; (:id :url :raw-url :status :headers :body :done :error)

我认为正文部分是不断更新的部分：

(class @(:body mary))
;; java.io.ByteArrayOutputStream

并尝试将流保存到文件中：

(with-open [r @(:body (statuses-filter :params {:track "mary rose"}
    :oauth-creds my-creds
    :callbacks *custom-streaming-callback*))
            w (io/writer "mary.txt")]
  (dosync (.write w (str r "\n"))))

这会写入出现在 mary.txt 文件中的第一条推文，但随后会关闭连接 - 大概是因为我在绑定到 r 之前使用了 @（但如果我将 @ 放在 r 的前面，它会窒息改为异步。另外，如果我这样做：

@(dosync (:body (statuses-filter :params {:track "mary rose"}
    :oauth-creds my-creds
    :callbacks *custom-streaming-callback*)))

再次，我只在连接关闭之前收到第一条推文。

如何保持连接打开以无限期地继续接收推文？

4

1 回答 1

2

您应该将“写入”代码放入该回调中：

(let [w (io/writer "mary.txt")
      callback (AsyncStreamingCallback.
                 (fn [_resp payload]
                   (.write w (-> (str payload) json/read-json :text))
                   (.write w "\n"))
                 (fn [_resp]
                   (.close w))
                 (fn [_resp ex]
                   (.close w)
                   (.printStackTrace ex)))]
  (statuses-filter
    :params {:track "mary rose"}
    :oauth-creds my-creds
    :callbacks callback))
于 2013-07-23T10:37:19.260 回答