   (let [^KryoTupleDeserializer deserializer (:deserializer executor-data)
         report-error-and-die (:report-error-and-die executor-data)]
     (fn [[task-id msg] sequence-id end-of-batch?]
+      ;; TODO: optimize by batching emits onto the transfer queue
       (with-error-reaction report-error-and-die
         ;;(log-debug "Processing message " msg)
         (let [^Tuple tuple (if (instance? Tuple msg) msg (.deserialize deserializer msg))]

    (doseq [[task-id task-data] task-datas
            :let [^IBolt bolt-obj (:object task-data)
                  tasks-fn (:tasks-fn task-data)
                  bolt-emit (fn [stream anchors values task]
                              (let [out-tasks (if task
                                                (tasks-fn task stream values)

                                (or out-tasks [])))]]
      (.prepare bolt-obj
                storm-conf
                user-context
                (OutputCollector.
                  (reify IOutputCollector
                     (emit [this stream anchors values]
