 (defn valid-syntax? [message]
   (== (count (ais-ex/extract-envelope-checksum message)) 2))
 
-
 ;;---
 ;; Core
 ;;---

                (if (and (supported-types msg-type) (= frag-num 1))
	          (condp = frag-count
	            1 (async/>!! out-ch [line])
	            ;; We assume group values stream in sequential order hence sequential 
	            ;; reads from the input channel
	            2 (async/>!! out-ch [line (async/<!! in-ch)])
	     	    (.println *err* (str "Unexpected fragment count: " frag-count ". " line)))
	          (.println *err* (str "Dropping [type=" msg-type "] " line)))))

          (async/>!! out-ch acc))))
    out-ch))

(defn- writer [format prefix msgs]
  (let [out-ch (async/chan)
        n (count msgs)
        active-threads (atom n)]
