finding and returning missing tweets

matthiasn · matthiasn · commit 98efd03182e1 · 2014-09-06T00:38:47.000+02:00
diff --git a/Clojure-Websockets/src/clj/birdwatch/channels.clj b/Clojure-Websockets/src/clj/birdwatch/channels.clj
@@ -6,11 +6,15 @@
 
 ;; channels
 (def tweets-chan (chan))
+(def tweet-missing-chan (chan)) ; channel for requesting missing tweet
+(def missing-tweet-found-chan (chan)) ; channel for responding to missing request
 (def persistence-chan (chan))
+(def rt-persistence-chan (chan))
 (def percolation-chan (chan))
 (def percolation-matches-chan (chan))
 
 ;; fanning tweets out to separate channels per task
 (def tweets-chan-mult (mult tweets-chan))
 (tap tweets-chan-mult percolation-chan)
 (tap tweets-chan-mult persistence-chan)
+(tap tweets-chan-mult rt-persistence-chan)
diff --git a/Clojure-Websockets/src/clj/birdwatch/communicator.clj b/Clojure-Websockets/src/clj/birdwatch/communicator.clj
@@ -50,8 +50,8 @@
                ;(doseq [t res] (chsk-send! (:uid params) [:some/tweet (:_source t)]))
                (chsk-send! (:uid params) [:tweet/prev-chunk res])))
 
-           [:cmd/missing tid]
-           (log/info "Missing Tweet requested:" tid)
+           [:cmd/missing params]
+           (put! c/tweet-missing-chan params)
 
            [:chsk/ws-ping params]
            () ; currently just do nothing with ping (no logging either)
@@ -87,3 +87,10 @@
          total-tweet-count (format "%,15d" (:count (p/total-tweet-count)))]
      (doseq [uid uids]
        (chsk-send! uid [:stats/total-tweet-count total-tweet-count])))))
+
+
+;; loop for sending missing tweet back to client
+(go
+ (while true
+   (let [msg (<! c/missing-tweet-found-chan)]
+     (chsk-send! (:uid msg) [:tweet/missing-tweet (:tweet msg)]))))
diff --git a/Clojure-Websockets/src/clj/birdwatch/persistence.clj b/Clojure-Websockets/src/clj/birdwatch/persistence.clj
@@ -26,15 +26,8 @@
   "get total count of indexed tweets from ElasticSearch"
   (esd/count conn (:es-index conf) "tweet"))
 
-;; loop for persisting tweets
-(go
- (while true
-   (let [t (<! c/persistence-chan)]
-     (try
-       (esd/put conn (:es-index conf) "tweet" (:id_str t) t)
-       (catch Exception ex (log/error ex "esd/put error"))))))
-
 (defn strip-tweet [t]
+  "take only actually needed fields from tweet"
   (let [u (:user t)]
     {:id_str (:id_str t)
      :id (:id t)
@@ -57,6 +50,10 @@
       (assoc t :retweeted_status (strip-tweet rt))
       t)))
 
+(defn get-tweet [id]
+  "get Tweet for specified ID"
+  (esd/get conn (:es-index conf) "tweet" id))
+
 (defn get-source [coll]
   "get vector with :_source of each ElasticSearch result"
   (map strip-source coll))
@@ -80,7 +77,7 @@
         n    (esrsp/total-hits res)
         hits (esrsp/hits-from res)
         res (get-source hits)]
-    (log/info "Total hits:" n "Retrieved:" (count hits))
+    ;(log/info "Total hits:" n "Retrieved:" (count hits))
     ;(log/info "top retweets in chunk" (pp/pprint (take 10 (into (priority-map-by >) (d/retweets res {})))))
     res))
 
@@ -93,10 +90,37 @@
     (perc/register-query conn "percolator" sha :query q)
     (log/info "Percolation registered for query" q "with SHA1" sha)))
 
+;; loop for persisting tweets
+(go
+ (while true
+   (let [t (<! c/persistence-chan)]
+     (try
+       (esd/put conn (:es-index conf) "tweet" (:id_str t) t)
+       (catch Exception ex (log/error ex "esd/put error"))))))
+
+;; loop for persisting retweets
+(go
+ (while true
+   (let [t (<! c/rt-persistence-chan)]
+     (when (:retweeted_status t)
+       (try
+         (esd/put conn (:es-index conf) "tweet" (:id_str t) t)
+         (catch Exception ex (log/error ex "esd/put error")))))))
+
 ;; loop for finding percolation matches and delivering those on the appropriate socket
 (go
  (while true
    (let [t (<! c/percolation-chan)
          response (perc/percolate conn "percolator" "tweet" :doc t)
          matches (into #{} (map #(:_id %1) (esrsp/matches-from response)))]
      (put! c/percolation-matches-chan [t matches]))))
+
+;; loop for finding missing tweet
+(go
+ (while true
+   (let [req (<! c/tweet-missing-chan)
+         res (get-tweet (:id_str req))
+         uid (:uid req)]
+     (if res
+       (put! c/missing-tweet-found-chan {:tweet (strip-source res) :uid uid})
+       (log/info "birdwatch.persistence missing" (:id_str req) res)))))
