         (if-let [line (async/<!! in-ch)]
           (do 
             (when (valid-syntax? line)
-              (println (format "filter-stream: %s" line))
               (let [d (details line)]
-                ;; Fragment 2 of multipart messages should be automatically passed thru. 
                 (if (passthru? supported-types d)
 	          (condp = (d "frag-count")
-	            1 (do
-                        (println "Consume 1")
-			(println line)
-                        (async/>!! out-ch [line]))
-	            2 (do
-                        (println (format "Consume 2: %s" (group-key line)))
-			(println line)
-                        (consume-multipart d (async/<!! in-ch) out-ch unpaired-frags))
-	     	    (.println *err* (str "Unexpected fragment count: " (d "frag-count") ". " line)))
-	          (.println *err* (str "Dropping [type=" (d "type") "] " line)))))
+	            1 (async/>!! out-ch [line])
+	            2 (consume-multipart line (async/<!! in-ch) unpaired-frags out-ch)
+	     	    (.println *err* (str "Unexpected fragment count: " (d "frag-count") ". " line))) 
+	          (.println *err* (str "Dropping [type=" (d "type") "] " line))))) 
             (recur))
           (async/close! out-ch))))
     out-ch))

;; Core
;;---


(defn details [line]
  (hash-map
    "line"       line
    "frag-count" (ais-ex/parse "frag-count" line)
    "frag-num"   (ais-ex/parse "frag-num" line)
    "type"       (extract-type (ais-ex/parse "payload" line))))

(defn group-key [line]
  (if-let  [prefix (ais-ex/parse "g" line)]
    (let [key (format "%s-%s" prefix (ais-ex/parse "radio-ch" line))]
      key)))

(defn find-fragment-match! [out-ch group-key msg unpaired-frags]
  (if-let [frag-match (@unpaired-frags group-key)]
    (async/>!! out-ch (sort-by (partial ais-ex/parse "frag-num") [msg frag-match]))
    (swap! unpaired-frags assoc group-key msg)))

(defn consume-multipart [a-msg b-msg unpaired-frags out-ch]
  (if (nil? b-msg)
    ;; Clearly msg b is not a match, let's look up a possible match in unpaired fragments
    (find-fragment-match! out-ch (group-key a-msg) a-msg unpaired-frags)
    (let [a-key (group-key a-msg)
          b-key (group-key b-msg)]
      (if (nil? b-key)

        ;; Msg b is not a multipart fragment.  Send to out channel for processing
        ;; and find possible match for msg a in unpaired fragments.  If a match is
        ;; not found, add to unpaired fragments
        (do
          (async/>!! out-ch [b-msg])
          (find-fragment-match! out-ch a-key a-msg unpaired-frags))

        ;; Msg b is a multipart fragment; if grouping key matches msg a, send pair
        ;; to out channel otherwise find possible matches for both msgs in unparied
        ;; fragments.  Add msgs to unpaired fragments if matches not found.
        (if (= a-key b-key)
          (async/>!! out-ch (sort-by (partial ais-ex/parse "frag-num") [a-msg b-msg]))
          (do
            (find-fragment-match! out-ch b-key b-msg unpaired-frags)
            (find-fragment-match! out-ch a-key a-msg unpaired-frags)))))))

(defn passthru? [supported-types d]
  (or 
