         ch (.channel ch-fut)]
     (if (and (.isSuccess ch-fut) 
              (established? ch))
-      ;; should set to connected now and then flush pending messages
       ch
       (do (.close ch)
           nil))))

          segments)
        segments))))

; (defn reset-connection-enqueue [messenger event peer-link buf]
;   (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)))

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

      (recur)))
  (.flush channel))

(defprotocol IConnectionManager
  (connect [_])
  (write [connection buf])
  (reset-connection [_])
  (enqueue-pending [_ buf])
  (close [_]))

(defprotocol IConnectionState 
  (initializing [this])
  (reset [this])
  (connecting [this])
  (failed [this])
  (connected [this]))

(defn add-failed-check 
  "Check if the future failed. link-val is used as a staleness check
  for the connection as it should not be reset if it has already been reset."
  [f connection buf]
  (.addListener f (reify GenericFutureListener
                    (operationComplete [_ _]
                      (when-not (.isSuccess f)
                        (timbre/error "Message failed to send: " (.cause f))
                        (reset-connection connection))))))


(defn make-pending-chan [messenger]
  (chan (sliding-buffer (:pending-buffer-size messenger))))

(defrecord ConnectionState [state]
  IConnectionState
  (initializing [_]
    (compare-and-set! state :initializing :connecting))
  (reset [_] 
    (compare-and-set! state :connected :reset))
  (connecting [_]
    (or (compare-and-set! state :reset :connecting)
        (compare-and-set! state :initializing :connecting)
        (compare-and-set! state :failed :connecting)))
  (failed [_]
    (compare-and-set! state :connecting :failed))
  (connected [_]
    (compare-and-set! state :connecting :connected)))

(defrecord ConnectionManager [messenger site state pending-ch channel]
  IConnectionManager
  (reset-connection [connection]
    (when (reset state)
      (reset! pending-ch (make-pending-chan messenger))
      (reset! channel nil)
      (connect connection)))

  (enqueue-pending [connection buf]
    (case @(:state state)
      :initializing (>!! @pending-ch buf)
      :connecting (>!! @pending-ch buf)
      :reset (connect connection)
      :failed (connect connection)
      ;; may have finished connecting in the time between
      ;; so we should check if it's established now before trying to reconnect
      :connected (let [channel-val @channel] 
                   (if (established? channel-val)
                     (.writeAndFlush channel-val buf (.voidPromise channel-val))
                     (reset-connection connection)))))

  (write [connection buf]
    (let [channel-val ^Channel @channel] 
      (if (and channel-val (.isActive channel-val))
        (let [fut (.writeAndFlush channel-val ^ByteBuf buf)] 
          (add-failed-check fut connection ^ByteBuf buf)) 
        (enqueue-pending connection buf))))

  (close [_] 
    (some-> @channel .close)
    (some-> @pending-ch close!))

  (connect [_]
    ; The state machine decides who gets to connect to ensure only one thread
    ; will connect, and the remaining will write out to the pending channel
    (when (connecting state)
      (future
        (if-let [opened-channel (create-client (:client-group messenger) 
                                               (:netty/external-addr site) 
                                               (:netty/port site))] 
          (let [connected? (connected state)] 
            (assert connected?)
            (flush-pending opened-channel @pending-ch)
            (reset! channel opened-channel))
          (do
            (failed state)
            (reset! channel nil)
            (reset! pending-ch (make-pending-chan messenger))))))))

(defmethod extensions/connect-to-peer NettyTcpSockets
  [messenger event site]
  (doto 
    (->ConnectionManager messenger 
                         site
                         (->ConnectionState (atom :initializing)) 
                         (atom (make-pending-chan messenger))
                         (atom nil))
    connect))

(defmethod extensions/receive-messages NettyTcpSockets
