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

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

             (timbre/fatal e))))))

(defn established? [channel]
  (and channel (.isActive channel)))

(defn create-client [client-group host port]
  (let [channel (if (epoll?)
