       :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)

            ;; 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  
              ;; group by node+port, do multipart send              
              (let [socket (node+port->socket (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)
                                (#(map-val endpoint->string %)))
              ;; we dont need a connection for the local tasks anymore
              needed-connections (->> my-assignment
                                      (filter-key (complement (-> worker :task-ids set)))
