       (let [state    (var-get local-store)
             [new-state rich-events] (pipeline state events)
             txlog     (kv/tx-log new-state)
-            txlog-msg (tx-log->messages txlog)
+            txlog-msg (map tx-log->messages txlog)
             output-events (map out-dispatch events)
             all-output (concat txlog-msg output-events)]
 

   (swap! global-stores
          update
          stream-id
          kv-restore events)
   []))



         (partition-filter-wrapper input-partitions))))


;; TODO: support dispatch-fn and  format fn
(defn topic-process-wrapper [component stream-id pipeline]
  (let [{:keys [local-store state-topic
                output-topic-partition-fn output-topic ]
         } (stream-config component stream-id)
        output-collector (:output-collector component)
        tx-log->messages (fn [event] [state-topic
                                     (:sourceId event)
                                     (to-json event)])
        out-dispatch (fn [event] [output-topic
                                 (output-topic-partition-fn event)
                                 (to-json event)])]
