Asynchronously handle tweet queue for twitter-streaming-client

gistfile1.clj

;; create the client with a twitter.api streaming method
(def stream (twitter-client/create-twitter-stream twitter.api.streaming/statuses-filter
                                          :oauth-creds twitter-creds
                                          :params {:follow user-id}))

(defn create-tweet-entities [twitter-tweet]
  ; the entirety of this method should be wrapped in a database transaction
  (insert stream-tweets
    (values (twitter-tweet-to-stream-tweet twitter-tweet))))

(defn commit-tweet-queue-to-database [queues]
  "Takes a map of queues and creates the tweet database objects for each tweet
  in the queue"
  (doseq [tweet (:tweet queues)] (create-tweet-entities tweet)))

(defn do-on-queues-changed [k, stream, os, nst]
  "do something when an object is added to queues"
  (let [buffered-tweets (:tweet (k nst))]
    (if (> (count buffered-tweets) 0)
      ; at least one tweet is in the queue
      (twitter-client/empty-queues stream commit-tweet-queue-to-database))))

(twitter-client/start-twitter-stream stream)
(add-watch stream :queues do-on-queues-changed))