                (= (m_fc (first frags)) (count frags)))
         (sort-by m_fn frags)))))
 
-(defn complete-multipart? 
+(defn- complete-multipart? 
   "Determines if an input sequence of message fragments is monotonic -- using fragment
   numbers as indicies.  If this is true and the sequence cardinality matches n, it connotes 
   the sequence is a complete multipart message.  Useful if sequence doesn't have group tags.  

                  (do
                    (async/>!! out-ch [line])
                    (recur dropped invalid))
                  (let [groups (preprocess-multipart line include-types unpaired-frags in-ch out-ch)]
                    (recur (+ dropped (count (groups :notag))) invalid)))
                (let [groups (preprocess-multipart line include-types unpaired-frags in-ch out-ch)]
                  (recur (+ dropped (count (groups :notag))) invalid)))
              (recur (inc dropped) invalid))
            (do
              (logging/debug (format "Invalid message syntax: %s" line))
              (recur dropped (inc invalid))))
          (do
            (log-metrics dropped invalid)
            (async/close! out-ch)))))
    out-ch)) 


;;;   Writes decoded messages to disk.
;;;

;; TODO: Optimize decoding when multipart messages aren't in include-list.  
;; Don't pass thru multipart fragments.
(defn- -decode? [include-types line]
  (if (= (m_fn line) 1)
    (contains? include-types (m_type line)) true))

(defn- filter-stream [include-types in-ch]
  (let [out-ch (async/chan buffer-size)
        decode? (partial -decode? include-types)
        unpaired-frags (atom {})]
    (async/thread
      (loop [dropped 0
