         (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))))

  (:require [clojure.stacktrace :as strace])
  (:gen-class))

;;;                      ____                     _           
;;;                     |  _ \  ___  ___ ___   __| | ___ _ __ 
;;;                     | | | |/ _ \/ __/ _ \ / _` |/ _ \ '__|
;;;                     | |_| |  __/ (_| (_) | (_| |  __/ |   
;;;                     |____/ \___|\___\___/ \__,_|\___|_|   
;;;
;;; The following application is a robust implemenation of ais-lib's core decoding functionality.  In practice, multipart 
;;; messages tend to stream in correct monotonically increasing order however we must account for all streaming configurations
;;; and not naively consume input streams.
;;; 
;;; Consider two sets or multipart messages, [a, a'] and [b, b']. x+ denotes 1 more messages.
;;;
;;; We must consider the following senarios when parsing multipart fragments from input streams:
;;;
;;; 1. a  a'
;;; 2. a' a
;;; 3. a  x+ a'
;;; 4. a' x+ a
;;; 5. a  b' b a'
;;;
;;;
;;; Entry point parameters:
;;;  * output-prefix: string
;;;    - Output filename prefix
;;;  * supported-types: [1-28]
;;;    - Messages types to decode.  All other messages are dropped
;;;  * n-threads: [1..)
;;;    - Number of processing threads.  Also sets the number of output files (where output files have the suffix
;;;      part-n).
;;;  * format: [csv | json]
;;;    - Output file data format
;;;


(def buffer-size 1000) 

(def ais-msg-types (into {} (for [k (range 1 28)] [k false])))

