 
 (defn- m_fn [line]
   (ais-ex/parse "frag-num" line))  
-
 
 
 ;;---

         result []]
    (if-let [[grp-k frags] (first pairs)]
      (if-let [unpaired-ret (@unpaired-frags grp-k)]
        (let [n (ais-ex/parse "frag-count" (first unpaired-ret))]
          ;; |frags| + |unpaired-ret| == n --> complete fragment set
          (if (= (+ (count unpaired-ret) (count frags)) n)
            (do
              (swap! unpaired-frags dissoc grp-k) ; fragment set complete, remove key
              (recur (rest pairs) (conj result (sort-by (partial ais-ex/parse "frag-num") (concat frags unpaired-ret)))))
            (do
              (swap! unpaired-frags update grp-k concat frags)
              (recur (rest pairs) result))))

(defn- preprocess-multipart [line include-types unpaired-frags in-ch out-ch]
  (let [remaining (filter (partial _decode? include-types) (consume (- (m_fc line) 1) in-ch))
        lines (conj remaining line)
        groups (group-fragments lines)
        notag (groups :notag)]
    (doseq [msg (groups :single)] ; frag reordering can mix single msgs in multipart sequence
      (async/>!! out-ch [msg]))
    (if-let [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.
      (if (and (> (count notag) 0) (complete-multipart? (m_fc frag) notag) (contains? include-types (m_type frag)))
        (do
          (async/>!! out-ch notag)
          {:notag []})
        {:notag notag})
      (do
        (doseq [msg (parse-multipart (dissoc groups :notag :single) unpaired-frags)]
          (if (contains? include-types (m_type (first msg)))
            (async/>!! out-ch msg)))
        {:notag notag}))))

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