               (.channel channel)
               (.handler (client-channel-initializer (new-client-handler))))]
       (.channel ^ChannelFuture (.awaitUninterruptibly (.connect b host port))))))
-
-(defn app [daemon messenger buf-recv-ch inbound-ch release-ch retry-ch shutdown-ch]
-  (thread
-    (loop []
-          (let [[buf ch] (alts!! [buf-recv-ch shutdown-ch])]
-            (when buf 
-              (try 
-                (let [msg (protocol/read-buf (:decompress-f messenger) buf)]
-                  ;(taoensso.timbre/info "Message buf: " buf msg)
-                  ;; undo retain added in server handler
-                  (.release ^ByteBuf buf)
-                  (let [t ^byte (:type msg)]
-                    (cond (= t protocol/messages-type-id) 
-                          (doseq [message (:messages msg)]
-                            (>!! inbound-ch message))
-
-                          (= t protocol/ack-type-id)
-                          (acker/ack-message daemon
-                                             (:id msg)
-                                             (:completion-id msg)
-                                             (:ack-val msg))
-
-                          (= t protocol/completion-type-id)
-                          (>!! release-ch (:id msg))
-
-                          (= t protocol/retry-type-id)
-                          (>!! retry-ch (:id msg))
-
-                          :else
-                          (throw (ex-info "Unexpected message received from Netty" {:message msg})))))
-                (catch Throwable e
-                  (taoensso.timbre/error e)
-                  (throw e)))
-              (recur))))))
 
 (defrecord NettyTcpSockets [peer-group]
   component/Lifecycle


(defmethod extensions/open-peer-site NettyTcpSockets
  [messenger assigned]
  ;; TODO: get rid of buf-recv-ch and do the handling inside 
  ;; the netty handler
  (let [buf-recv-ch (chan 10000)
        inbound-ch (:inbound-ch (:messenger-buffer messenger))
        release-ch (:release-ch messenger)
        retry-ch (:retry-ch messenger)
        daemon (:acking-daemon messenger)
        shutdown-fn (start-netty-server (:boss-group messenger)
                                        (:worker-group messenger)
                                        (:bind-addr messenger) 
                                        (:netty/port assigned) 
                                        buf-recv-ch)
        shutdown-ch (chan 1)
        app-loop (app daemon messenger buf-recv-ch inbound-ch release-ch retry-ch shutdown-ch)]
    (reset! (:resources messenger)
            {:shutdown-fn shutdown-fn 
             :shutdown-ch shutdown-ch
             :buf-recv-ch buf-recv-ch
             :app-loop app-loop})))

(defmethod extensions/connect-to-peer NettyTcpSockets
  [messenger event {:keys [netty/external-addr netty/port]}]

    (taoensso.timbre/info "Stopping Netty TCP Sockets")
    (try 
      (when-let [rs @resources]
        (let [{:keys [shutdown-fn]} rs] 
          (shutdown-fn))
        (reset! resources nil)) 
      (catch Throwable e (timbre/fatal e)))
    (assoc component :bind-addr nil :external-addr nil 
