title: How do I save streaming tweets from twitter-api in Clojure?
body: <p>I have been using clojure for a while but I am unfamiliar with async io that the twitter-api (<a href="https://github.com/adamwynne/twitter-api" rel="nofollow">https://github.com/adamwynne/twitter-api</a>) is based on.</p>
  
  <p>I want to collect all tweets matching a given set of keywords.  For example everything matching "Mary rose" (something trending in the UK right now).  The docs for making streaming calls say to do something like:</p>
  
  <pre><code>(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*)
  </code></pre>
  
  <p>If I then do something like:</p>
  
  <pre><code>(def mary (statuses-filter :params {:track "mary rose"}
       :oauth-creds my-creds
       :callbacks *custom-streaming-callback*))
  </code></pre>
  
  <p>I get a map of the http response:</p>
  
  <pre><code>(keys mary)
  ;; (:id :url :raw-url :status :headers :body :done :error)
  </code></pre>
  
  <p>I think the body part is the bit that is continually updated:</p>
  
  <pre><code>(class @(:body mary))
  ;; java.io.ByteArrayOutputStream
  </code></pre>
  
  <p>and have tried this to save the stream to a file:</p>
  
  <pre><code>(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")))) 
  </code></pre>
  
  <p>This writes the first tweet that comes up to the mary.txt file but then closes the connection - presumably because I am using the @ in front of the binding to r (but it chokes if I put the @ in front of the r in the desync instead.  Also, if I do:</p>
  
  <pre><code>@(dosync (:body (statuses-filter :params {:track "mary rose"}
      :oauth-creds my-creds
      :callbacks *custom-streaming-callback*)))
  </code></pre>
  
  <p>again I only get the first tweet before the connection is closed.</p>
  
  <p>How can I keep the connection open to continue receiving tweets indefinitely?</p>