 (defn function [{:keys [onyx.core/replica
                         onyx.core/peer-replica-view
                         onyx.core/state
-                        onyx.core/id
                         onyx.core/messenger
-                        onyx.core/job-id
                         onyx.core/serialized-task] :as pipeline-data}]
   (->Function replica
               peer-replica-view
               state
-              id
               messenger
-              job-id
               (:egress-ids serialized-task)))

               (let [link (operation/peer-link replica-val state event target)]
                 (onyx.extensions/send-messages messenger event link segs)))))))))

  ([{:keys [onyx.core/id onyx.core/results
            onyx.core/messenger onyx.core/job-id
            onyx.core/state onyx.core/replica
            onyx.core/peer-replica-view onyx.core/serialized-task] :as event}]
   (write-batch event replica peer-replica-view state id messenger job-id (:egress-ids serialized-task))))

(defrecord Function [replica peer-replica-view state id messenger job-id egress-tasks]
  p-ext/Pipeline
  (read-batch
    [_ event]

   {:onyx.core/batch (onyx.extensions/receive-messages messenger event)}))

(defn write-batch
  ([event replica peer-replica-view state messenger egress-tasks]
   (let [segments (:segments (:onyx.core/results event))]
     (when-not (empty? segments)
       (let [replica-val @replica
