clojure - 如何在 Clojure 中保存来自 twitter-api 的流式推文？

Question

我已经使用 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)

我认为 body 部分是不断更新的部分:

(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 的绑定(bind)前面使用了 @(但如果我将 @ 放在 r 前面，它就会窒息) r 改为 desync。另外，如果我这样做:

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

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

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

Answer 1

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

(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))