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

(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)]
    (doseq [msg (groups :single)]
      ;; Send along any single messages that are interspersed in multipart fragments --
      ;; possible via fragment reordering.
      (async/>!! out-ch [msg]))
    (let [notag (groups :notag)]
      (if-let [frag (first notag)]
        ;; Assumption: multipart fragments are either tagged or untagged and not both.
        (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)]
            ;; We can't verify multipart message types until entire message is constructed.
            ;; Only propagate multiparts in include list
            (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))

         result []]
    (if-let [[grp-k frags] (first pairs)]
      (if-let [unpaired-ret (@unpaired-frags grp-k)]
        (let [n (m_fc (first unpaired-ret))]
          (if (= (+ (count unpaired-ret) (count frags)) n) ; all fragments present => match
            (do
              (swap! unpaired-frags dissoc grp-k) ; fragment set complete, remove key
              (recur (rest pairs) (conj result (sort-by m_fn (concat frags unpaired-ret)))))
            (do
              (swap! unpaired-frags update grp-k concat frags)
              (recur (rest pairs) result))))
