                   (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)))
+                  (let [ret (preprocess-multipart line include-types unpaired-frags in-ch out-ch)]
+                    (recur (+ dropped (count (ret :drop))) invalid)))
+                (let [ret (preprocess-multipart line include-types unpaired-frags in-ch out-ch)]
+                  (recur (+ dropped (count (ret :drop))) invalid)))
               (recur (inc dropped) invalid))
             (do
               (logging/debug (format "Invalid message syntax: %s" line))
               (recur dropped (inc invalid))))
           (do
-            (log-metrics dropped invalid)
+            (log-metrics (+ dropped (count @unpaired-frags)) invalid)
             (async/close! out-ch)))))
     out-ch)) 
 

               (= (m_fc (first frags)) (count frags)))
        (sort-by m_fn frags)))))

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

    (for [match (find-matches! frag-groups unpaired-frags)]
      match)))

(defn- -forward [source include-types out-ch]
  (loop [msgs source
         acc []]
    (if-let [msg-frags (first msgs)]
      (if (contains? include-types (m_type (first msg-frags))) 
        (do
          (async/>!! out-ch msg-frags)
          (recur (rest msgs) acc))
        (recur (rest msgs) (concat acc msg-frags)))
      acc)))

(defn- -forward-multipart [include-types unpaired-frags groups out-ch]
  (let [notag (groups :notag)]
    (if-let [lead-frag (first notag)]
      ;; ASSUMPTION: multipart fragments are either tagged or untagged (and not both) in an input stream.
      ;; In either case, determine if input fragment sequence forms a complete message.  In tagged
      ;; case, we can search for unmatched fragments in cache to form complete message.  In untagged
      ;; case, no searches can be done; As such, if current fragment sequence doesn't form a complete
      ;; message they cannot be used as possible matches for subsequent fragment sequences due to ordering 
      ;; ambiguity 
      (if (and (complete-multipart? (m_fc lead-frag) notag) (contains? include-types (m_type lead-frag)))
        (do
          (async/>!! out-ch notag)
          [])
        notag)
      (-forward (parse-multipart (dissoc groups :notag :single) unpaired-frags) include-types out-ch))))

(defn- preprocess-multipart [line include-types unpaired-frags in-ch out-ch]
  (let [remaining (consume (- (m_fc line) 1) in-ch)
        groups (group-fragments (conj remaining line))]
    {:drop (concat (-forward (map vector (groups :single)) include-types out-ch)
                   (-forward-multipart include-types unpaired-frags groups out-ch))}))

(defn- log-metrics [dropped invalid]
  (logging/info (format "count.invalid.total=%d" invalid))
