     (catch Exception e 
     (do
       (println e)
-      (println text)
       {:text ""}))))
 
 (defn parse-filter []
   (comp
     (map safe-parse-json)
-    (filter (fn [t] (-> t :text str/lower-case contains-profanity?)))))
+    (filter 
+      (fn [t] 
+        (-> t
+            ;; Pull the text out, or an empty string. 
+            (:text "")
+            ;; Convert to lower case.
+            str/lower-case 
+            ;; Check for profanity.
+            contains-profanity?)))))
 
 (defn turbine-topology-vector [elastic-url]
   (let [es-conn (esr/connect elastic-url)]

;; Partitioning transducer.
(defn partitioner-xform []
  (comp
    ;; Partition by whether it's a tweet terminator.
    (partition-by #(str/includes? % "\r\n"))
    ;; Merge the terminator with the rest of the tweet.
    (partition-all 2)
    ;; Flatten the nested sequence of chunks.
    (map flatten)
    ;; And concatenate into one string.
    (map #(apply str %))
    ;; And finally filter out empty strings.
    (filter #(not (str/blank? %)))))


;; (def transfer-chan (chan 5))
(def transfer-chan (chan 5))

    ;; Read from the dechunker chan in a go loop and drop into turbine.
    (go-loop []
      (let [tweet-chunk (<! transfer-chan)]
        (turbine-in tweet-chunk))
      (recur))
    
    (statuses-filter :params {:track ["trump" "pence"]}
                     :oauth-creds creds
                     :callbacks *callback*)))
