 (ns ^:no-doc onyx.messaging.netty-tcp
-    (:require [clojure.core.async :refer [chan >!! >! <!! alts!! timeout close! go-loop dropping-buffer]]
+    (:require [clojure.core.async :refer [chan >!! >! <!! alts!! timeout close! thread go-loop dropping-buffer]]
               [com.stuartsierra.component :as component]
               [taoensso.timbre :as timbre]
               [onyx.messaging.protocol-netty :as protocol]


(defmethod extensions/send-messages NettyTcpSockets
  [messenger event ^Channel peer-link messages]
  (.writeAndFlush peer-link 
                  ^ByteBuf (protocol/build-messages-msg-buf (:compress-f messenger) messages) 
                  (.voidPromise ^Channel peer-link)))

(defmethod extensions/internal-ack-message NettyTcpSockets
  [messenger event ^Channel peer-link message-id completion-id ack-val]
  (.writeAndFlush peer-link 
                  ^ByteBuf (protocol/build-ack-msg-buf message-id completion-id ack-val)
                  (.voidPromise ^Channel peer-link)))

    (handlerAdded [ctx])
    (handlerRemoved [ctx])
    (exceptionCaught [context cause]
      (timbre/error cause "TCP client exception.")
      (.close context))))

(defn client-channel-initializer [handler]
