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

【问题标题】：How do I save streaming tweets from twitter-api in Clojure?如何在 Clojure 中保存来自 twitter-api 的流式推文？
【发布时间】：2013-07-23 18:37:19
【问题描述】：

我使用 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 在 desync 中。另外，如果我这样做：

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

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

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

【问题讨论】：

标签： clojure twitter

【解决方案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))

【讨论】：