       (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
-  ; Ensures only a single thread will connect by CAS.
-  ; 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
-  [messenger _ link {:keys [netty/external-addr netty/port]}]
-  (let [link-val @link
-        pending-ch (:pending-ch link-val)
-        new-link-val (assoc link-val :state :connecting)
-        state (:state link-val)
-        connect-here? (and (#{:initializing :failed} state)
-                           (compare-and-set! link link-val new-link-val))] 
-    (when connect-here?
-      (future
-        (if-let [channel (create-client (:client-group messenger) external-addr port)] 
-          (do (reset! link (assoc new-link-val :channel channel :state :connected))
-              (flush-pending channel pending-ch))
-          (reset! link {:site (:site link-val)}))))))
+  [messenger _ link site]
+  (doto 
+    (->Connection messenger 
+                  site 
+                  (atom {:state :initializing
+                         :pending-ch (make-pending-chan messenger)}))
+    connect))
 
 (defmethod extensions/receive-messages NettyTcpSockets
   [messenger {:keys [onyx.core/task-map] :as event}]

(defn start-netty-server
  [boss-group worker-group host port messenger inbound-ch release-ch retry-ch]
  (let [bootstrap (ServerBootstrap.)
        channel (if (epoll?)
                  EpollServerSocketChannel
                  NioServerSocketChannel)
        channel-group (DefaultChannelGroup. (str "tcp-server " host ":" port)
                        (ImmediateEventExecutor/INSTANCE))
        initializer (channel-initializer-done 
                     (gen-tcp-handler channel-group 
                                      (create-server-handler messenger 
                                                             inbound-ch 
                                                             release-ch 
                                                             retry-ch)))] 
    (doto bootstrap
      (.group boss-group worker-group)
      (.channel channel)

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

(defmethod extensions/internal-ack-message NettyTcpSockets
  [messenger event peer-link message-id completion-id ack-val]
  (write 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 peer-link (protocol/build-completion-msg-buf id)))

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

(defmethod extensions/close-peer-connection NettyTcpSockets
  [messenger event peer-link]
  (close peer-link))
