bolt-emit (fn [stream anchors values task](let [out-tasks (if task(tasks-fn task stream values)(tasks-fn stream values))](fast-list-iter [t out-tasks](let [anchors-to-ids (HashMap.)](fast-list-iter [^TupleImpl a anchors](let [root-ids (-> a .getMessageId .getAnchorsToIds .keySet)](when (pos? (count root-ids))(let [edge-id (MessageId/generateId rand)](.updateAckVal a edge-id)(fast-list-iter [root-id root-ids](put-xor! anchors-to-ids root-id edge-id))))))(transfer-fn t(TupleImpl. worker-contextvaluestask-idstream(MessageId/makeId anchors-to-ids)))))(or out-tasks [])))

:transfer-fn (mk-executor-transfer-fn batch-transfer->worker)

(defn mk-executor-transfer-fn [batch-transfer->worker](fn this([task tuple block? ^List overflow-buffer](if (and overflow-buffer (not (.isEmpty overflow-buffer)))(.add overflow-buffer [task tuple])(try-cause(disruptor/publish batch-transfer->worker [task tuple] block?)(catch InsufficientCapacityException e(if overflow-buffer(.add overflow-buffer [task tuple])(throw e))))))([task tuple overflow-buffer](this task tuple (nil? overflow-buffer) overflow-buffer))([task tuple](this task tuple nil))))

transfer-thread (disruptor/consume-loop* (:transfer-queue worker) transfer-tuples)

(defn mk-transfer-tuples-handler [worker](let [^DisruptorQueue transfer-queue (:transfer-queue worker)drainer (ArrayList.)node+port->socket (:cached-node+port->socket worker)task->node+port (:cached-task->node+port worker)endpoint-socket-lock (:endpoint-socket-lock worker)](disruptor/clojure-handler(fn [packets _ batch-end?](.addAll drainer packets)(when batch-end?(read-locked endpoint-socket-lock(let [node+port->socket @node+port->sockettask->node+port @task->node+port];; consider doing some automatic batching here (would need to not be serialized at this point to remo;; try using multipart messages ... first sort the tuples by the target node (without changing the lo17(fast-list-iter [[task ser-tuple] drainer];; TODO: consider write a batch of tuples here to every target worker;; group by node+port, do multipart send(let [node-port (get task->node+port task)](when node-port (.send ^IConnection (get node+port->socket node-port) task ser-tuple))))))(.clear drainer))))))

storm.messaging.transport: "backtype.storm.messaging.zmq"

refresh-connections (mk-refresh-connections worker)

(swap! (:cached-node+port->socket worker)#(HashMap. (merge (into {} %1) %2))(into {}(dofor [endpoint-str new-connections:let [[node port] (string->endpoint endpoint-str)]][endpoint-str(.connect^IContext (:mq-context worker)storm-id((:node->host assignment) node)port)])))

(^IConnection connect [this ^String storm-id ^String host ^int port](require 'backtype.storm.messaging.zmq)(-> context(mq/socket mq/push)(mq/set-hwm hwm)(mq/set-linger linger-ms)(mq/connect (get-connect-zmq-url local? host port))mk-connection))

  (let [out-tasks (if task            (tasks-fn task stream values)       (tasks-fn stream values))]  (fast-list-iter [t out-tasks]       (let [anchors-to-ids (HashMap.)]            (fast-list-iter [^TupleImpl a anchors]                    (let [root-ids (-> a .getMessageId .getAnchorsToIds .keySet)]                    (when (pos? (count root-ids))                   (let [edge-id (MessageId/generateId rand)]                      (.updateAckVal a edge-id)                     (fast-list-iter [root-id root-ids]                            (put-xor! anchors-to-ids root-id edge-id))                    )))) (transfer-fn t                (TupleImpl. worker-context                        values                        task-id                       stream                        (MessageId/makeId anchors-to-ids)))))

    tasks-fn (:tasks-fn task-data)

fn ([^Integer out-task-id ^String stream ^List values](when debug?(log-message "Emitting direct: " out-task-id "; " component-id " " stream " " values))(let [target-component (.getComponentId worker-context out-task-id)component->grouping (get stream->component->grouper stream)  grouping (get component->grouping target-component)out-task-id (if grouping out-task-id)](when (and (not-nil? grouping) (not= :direct grouping))(throw (IllegalArgumentException. "Cannot emitDirect to a task expecting a regular grouping")))                          (apply-hooks user-context .emit (EmitInfo. values stream task-id [out-task-id]))(when (emit-sampler)(builtin-metrics/emitted-tuple! (:builtin-metrics task-data) executor-stats stream)(stats/emitted-tuple! executor-stats stream)(if out-task-id(stats/transferred-tuples! executor-stats stream 1)(builtin-metrics/transferred-tuple! (:builtin-metrics task-data) executor-stats stream 1)))(if out-task-id [out-task-id])))

:stream->component->grouper (outbound-components worker-context component-id)

(defn outbound-components"Returns map of stream id to component id to grouper"[^WorkerTopologyContext worker-context component-id](->> (.getTargets worker-context component-id)clojurify-structure(map (fn [[stream-id component->grouping]][stream-id(outbound-groupingsworker-contextcomponent-idstream-id(.getComponentOutputFields worker-context component-id stream-id)component->grouping)]))(into {})(HashMap.)))