transducers for processing stream and logging processed count

matthiasn · matthiasn · commit 2a946745ab73 · 2014-09-27T18:12:13.000+02:00
diff --git a/Clojure-Websockets/src/clj/birdwatch/twitter_client.clj b/Clojure-Websockets/src/clj/birdwatch/twitter_client.clj
@@ -10,23 +10,16 @@
    [twitter.api.streaming :as tas]
    [twitter.oauth :as oauth]
    [twitter.callbacks.handlers :as tch]
-   [clojure.core.async :as async :refer [<! <!! >! >!! chan put! take! alts! timeout go go-loop pipe]]
+   [clojure.core.async :as async :refer [<! <!! >! >!! close! chan put! take! alts! timeout go go-loop pipe]]
    [com.stuartsierra.component :as component])
   (:import (twitter.callbacks.protocols AsyncStreamingCallback)))
 
 (defn- creds [config] (oauth/make-oauth-creds (:consumer-key config) (:consumer-secret config)
                                               (:user-access-token config) (:user-access-token-secret config)))
 
-(def cat (fn [step] (fn [r x] (reduce step r x))))
-
-(defn- streaming-buffer []
+(defn flattening []
   (fn [step]
-    (let [buff (volatile! "")]
-      (fn [r x]
-        (let [json-lines (str/split-lines (str/replace (str @buff x) #"\}\{" "}\r\n{"))
-              to-process (butlast json-lines)]
-          (vreset! buff (last json-lines))
-          (if to-process (step r to-process) r))))))
+    (fn [r x] (reduce step r x))))
 
 (defn mapping [f]
   (fn [step]
@@ -36,18 +29,41 @@
   (fn [step]
     (fn [r x] (if (pred x) (step r x) r))))
 
-(defn tweet? [json]
-  (let [text (:text json)]
-    (when-not text (log/error "error-msg" json))
+(defn- log-count [last-received]
+  (fn [step]
+    (let [cnt (volatile! 0)]
+      (fn [r x]
+        (when (== (mod @cnt 1000) 0) (log/info "processed" @cnt "since startup"))
+        (vswap! cnt inc)
+        (reset! last-received (t/now))
+        (step r x)))))
+
+(defn- streaming-buffer []
+  (fn [step]
+    (let [buff (volatile! "")]
+      (fn [r x]
+        (let [json-lines (str/split-lines (str/replace (str @buff x) #"\}\{" "}\r\n{"))
+              to-process (butlast json-lines)]
+          (vreset! buff (last json-lines))
+          (if to-process (step r to-process) r))))))
+
+(defn- tweet? [data]
+  (let [text (:text data)]
+    (when-not text (log/error "error-msg" data))
     text))
 
 (defn- parse-json [jstr]
   "parse JSON string"
-  (try (json/read-json jstr)
-    (catch Exception ex (log/error ex "JSON parsing" jstr) {:ex ex})))
+  (try (json/read-str jstr :key-fn clojure.core/keyword)
+    (catch Exception e {:exception e :jstr jstr})))
 
-(defn- process-chunk []
-  (comp (streaming-buffer) cat (mapping parse-json) (filtering tweet?)))
+(defn- process-chunk [last-received]
+  (comp
+   (streaming-buffer)
+   (flattening)
+   (mapping parse-json)
+   (filtering tweet?)
+   (log-count last-received)))
 
 (defn- tweet-chunk-callback [chunk-chan]
   (tas/AsyncStreamingCallback. #(>!! chunk-chan (str %2))
@@ -61,35 +77,40 @@
                 :oauth-creds (creds conf)
                 :callbacks (tweet-chunk-callback chunk-chan))))
 
-(defn- stop-twitter-conn! [conn]
+(defn stop-twitter-conn! [conn]
   (let [m (meta @conn)]
     (when m (log/info "Stopping Twitter client.")
       ((:cancel m)))))
 
-(defn run-watch-loop [conf conn chunk-chan last-received watch-active]
+(defn- run-watch-loop [conf conn chunk-chan last-received watch-active]
   "run loop watching the twitter client and restarting it if necessary"
   (reset! watch-active true)
-  (go-loop [] (<! (timeout 10000)) ;; check connection every 10 seconds
-           (let [since-last-sec (t/in-seconds (t/interval @last-received (t/now)))]
-             (when @watch-active
+  (go-loop [] (<! (timeout (* (:tw-check-interval-sec conf) 1000)))
+           (let [since-last-sec (t/in-seconds (t/interval @last-received (t/now)))
+                 active @watch-active]
+             (when active
                (when (> since-last-sec (:tw-check-interval-sec conf))
                  (log/error since-last-sec "seconds since last tweet received")
                  (stop-twitter-conn! conn)
-                 (<! (timeout (* (:tw-check-interval-sec conf) 1000)))
+                 (<! (timeout (* (:tw-restart-wait conf) 1000)))
                  (start-twitter-conn! conf conn chunk-chan))
                (recur)))))
 
-(defrecord Twitterclient [conf channels conn watch-active]
+(defrecord Twitterclient [conf channels conn chunk-chan watch-active]
   component/Lifecycle
   (start [component] (log/info "Starting Twitterclient Component")
-         (let [chunk-chan (chan 1 (process-chunk))
-               counter (atom 0)
+         (let [last-received (atom (t/epoch))
+               chunk-chan (chan 1 (process-chunk last-received))
                conn (atom {})
-               last-received (atom (t/epoch))
                watch-active (atom false)]
+
+           (def connection conn)
+           (def config conf)
+           (def channel chunk-chan)
+
            (start-twitter-conn! conf conn chunk-chan)
            (pipe chunk-chan (:tweets channels) false)
-           ; (run-watch-loop conf conn chunk-chan last-received watch-active)
+           (run-watch-loop conf conn chunk-chan last-received watch-active)
            (assoc component :conn conn :chunk-chan chunk-chan :watch-active watch-active)))
   (stop [component] (log/info "Stopping Twitterclient Component")
         (reset! watch-active false)
diff --git a/Clojure-Websockets/twitterconf-tpl.edn b/Clojure-Websockets/twitterconf-tpl.edn
@@ -13,6 +13,7 @@
   :es-cluster-name          "<YOUR CLUSTER NAME HERE>"
   :es-index                 "birdwatch"
   :track                    "clojure,love"
-  :tw-check-interval-sec    120
+  :tw-check-interval-sec    10
+  :tw-restart-wait          60
   :port                     8888
   :pidfile-name             "birdwatch.pid"}
