     (reset! (:resources messenger)
             {:shutdown-fn shutdown-fn})))
 
-(defmethod extensions/initialize-peer-link NettyTcpSockets
-  [messenger _ site]
-  {:state :initializing
-   :site site 
-   :pending-ch (chan (:pending-buffer-size messenger))})
-
 (defn flush-pending 
-  "Flush all pending bufs. Useful when the channel is established"
+  "Flush all pending bufs. Run after when the channel is established."
   [^Channel channel pending-ch]
   (close! pending-ch)
   (loop []

          segments)
        segments))))

(defn enqueue-pending [messenger event peer-link buf]
  (let [link-val @peer-link
        channel ^Channel (:channel link-val)] 
    (case (:state link-val)
      :initializing (>!! (:pending-ch link-val) buf)
      :connecting (>!! (:pending-ch link-val) buf)
      ;; may have finished connecting in the time between
      ;; so we should check if it's established now before trying to reconnect
      :connected (if (established? channel)
                   (.writeAndFlush channel buf (.voidPromise channel))
                   (let [site (:site link-val)
                         peer-link-new (extensions/initialize-peer-link messenger event site)
                         link-val (reset! peer-link peer-link-new)] 
                     (>!! (:pending-ch link-val) buf)
                     (extensions/connect-to-peer messenger event peer-link site))))))

(defn add-failed-check [f messenger event peer-link buf]
  (.addListener f (reify GenericFutureListener
                    (operationComplete [_ _]
                      (when-not (.isSuccess f)
                        (timbre/error "Message failed to send: " (.cause f))
                        ; Not sure if re-queing here is a great idea
                        (enqueue-pending messenger event peer-link buf))))))
              
(defn write [messenger event peer-link ^ByteBuf buf]
  (let [channel ^Channel (:channel @peer-link)] 
    (if (established? channel)
      (let [fut (.writeAndFlush channel buf)] 
        (add-failed-check fut messenger event peer-link buf)) 
      (enqueue-pending messenger event peer-link buf))))

(defmethod extensions/send-messages NettyTcpSockets
  [messenger event peer-link messages]
  (write messenger event peer-link (protocol/build-messages-msg-buf (:compress-f messenger) messages)))

(defmethod extensions/internal-ack-message NettyTcpSockets
  [messenger event peer-link message-id completion-id ack-val]
  (write messenger event peer-link (protocol/build-ack-msg-buf message-id completion-id ack-val)))

(defmethod extensions/internal-complete-message NettyTcpSockets
  [messenger event id peer-link]
  (write messenger event peer-link (protocol/build-completion-msg-buf id)))

(defmethod extensions/internal-retry-message NettyTcpSockets
  [messenger event id ^Channel peer-link]
  (write messenger event peer-link (protocol/build-retry-msg-buf id)))

(defmethod extensions/close-peer-connection NettyTcpSockets
  [messenger event ^Channel peer-link]
  (some-> @peer-link :channel .close))

          assigned-port (.. ch localAddress getPort)]
      (timbre/info "Netty server" host assigned-port "online")
      (fn killer []
        (.close channel-group)
        (timbre/info "TCP server" host port "shut down")))))

(defn new-client-handler []
