   (merge event {:onyx.core/acked (count batch)}))
 
 (defn apply-fn-shim
-  [{:keys [onyx.core/decompressed onyx.core/task
-           onyx.core/catalog onyx.core/params] :as event}]
-  (let [task (first (filter (fn [entry] (= (:onyx/name entry) task)) catalog))
-        results (flatten (map (partial apply-fn task params) decompressed))]
+  [{:keys [onyx.core/decompressed onyx.transform/fn onyx.core/params] :as event}]
+  (let [results (flatten (map (partial apply-fn fn params) decompressed))]
     (merge event {:onyx.core/results results})))
 
 (defn compress-batch-shim [{:keys [onyx.core/results] :as event}]


(defn seal-resource-shim [{:keys [onyx.core/queue onyx.core/egress-queues] :as event}]
  (merge event (cap-queue queue egress-queues)))

(defmethod l-ext/read-batch :default
  [event] (read-batch-shim event))

(defn decompress-segment [queue message]
  (extensions/read-message queue message))

(defn apply-fn [f params segment]
  ((reduce #(partial %1 %2) f params) segment))

(defn compress-segment [segment]
  {:compressed (.array (fressian/write segment))
