             ;; try using multipart messages ... first sort the tuples by the target node (without changing the local ordering)
             
             (fast-list-iter [[task ser-tuple] drainer]
-              ;; TODO: this lookup (with the vector is expensive)
-              ;; TODO: write a batch of tuples here to every target worker  
+              ;; TODO: consider write a batch of tuples here to every target worker  
               ;; group by node+port, do multipart send              
-              (let [socket (node+port->socket (task->node+port task))]
+              (let [socket (get node+port->socket (get task->node+port task))]
                 (msg/send socket task ser-tuple)
                 ))))
         (.clear drainer)))))

        (this (fn [& ignored] (schedule (:timer worker) 0 this))))
      ([callback]
        (let [assignment (.assignment-info storm-cluster-state storm-id callback)
              my-assignment (-> assignment :executor->node+port to-task->node+port (select-keys outbound-tasks))
              ;; we dont need a connection for the local tasks anymore
              needed-connections (->> my-assignment
                                      (filter-key (complement (-> worker :task-ids set)))

      :transfer-fn (mk-transfer-fn <>)
      )))

(defn- endpoint->string [[node port]]
  (str port "/" node))

(defn string->endpoint [^String s]
  (let [[port-str node] (.split s "/" 2)]
    [node (Integer/valueOf port-str)]
    ))

(defn mk-refresh-connections [worker]
  (let [outbound-tasks (worker-outbound-tasks worker)
        conf (:conf worker)
