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

      (recur)))
  (.flush channel))

(defprotocol ConnectionManager
  (connect [_])
  (write [connection buf])
  (reset-connection [_ previous-link-val])
  (enqueue-pending [_ buf])
  (close [_]))

(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 link-val buf]
  (.addListener f (reify GenericFutureListener
                    (operationComplete [_ _]
                      (when-not (.isSuccess f)
                        (timbre/error "Message failed to send: " (.cause f))
                        (reset-connection connection link-val))))))

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

(defrecord Connection [messenger site link]
  ConnectionManager
  (reset-connection [connection previous-link-val]
    (when (compare-and-set! link 
                            previous-link-val 
                            {:state :reset
                             :pending-ch (make-pending-chan messenger)})
      (connect connection)))

  (enqueue-pending [connection buf]
    (let [link-val @link
          channel ^Channel (:channel link-val)] 
      (case (:state link-val)
        :initializing (>!! (:pending-ch link-val) buf)
        :connecting (>!! (:pending-ch link-val) 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 (if (established? channel)
                     (.writeAndFlush channel buf (.voidPromise channel))
                     (reset-connection connection link-val)))))

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

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

  (connect [_]
    ; Ensures only a single thread will connect by only allowing the connection that performed
    ; the CAS to go ahead and connect.
    ; The spawned future will be the one to set the channel in the peer-link. 
    ; All sending threads will continue on adding to the pending-ch until the 
    ; connection has been established
    (let [link-val @link
          state (:state link-val)
          connecting-link-val (assoc link-val :state :connecting)]
      (when (#{:initializing :reset :failed} state)
        (when (compare-and-set! link link-val connecting-link-val)
          (future
            (if-let [channel (create-client (:client-group messenger) (:netty/external-addr site) (:netty/port site))] 
              (let [pending-ch (:pending-ch connecting-link-val)
                    connected-link-val (assoc connecting-link-val 
                                              :channel channel 
                                              :state :connected
                                              :pending-ch nil)
                    was-set? (compare-and-set! link connecting-link-val connected-link-val)] 
                (assert was-set?)
                (flush-pending channel pending-ch))
              (reset! link {:state :failed
                            :pending-ch (make-pending-chan messenger)}))))))))

(defmethod extensions/connect-to-peer NettyTcpSockets
  [messenger event site]
  (doto 
    (->Connection messenger 
                  site 
                  (atom {:state :initializing
                         :pending-ch (make-pending-chan messenger)}))
    connect))

(defmethod extensions/receive-messages NettyTcpSockets

          segments)
        segments))))

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