           bind-addr (bind-addr config)
           external-addr (external-addr config)
           ports (allowable-ports config)]
-      (assoc component 
-             :bind-addr bind-addr 
-             :external-addr external-addr
-             :boss-group boss-group
-             :client-group client-group
-             :worker-group worker-group
-             :ports ports
-             :resources (atom nil)
-             :release-ch release-ch)))
+      (assoc component
+        :bind-addr bind-addr 
+        :external-addr external-addr
+        :boss-group boss-group
+        :client-group client-group
+        :worker-group worker-group
+        :ports ports
+        :resources (atom nil)
+        :release-ch release-ch
+        :decompress-f (or (:onyx.messaging/decompress-fn (:config peer-group)) decompress)
+        :compress-f (or (:onyx.messaging/compress-fn (:config peer-group)) compress))))
 
   (stop [{:keys [resources release-ch] :as component}]
     (taoensso.timbre/info "Stopping Netty TCP Sockets")

        (.channel ^ChannelFuture (.connect b host port))))))

(defn app [daemon parsed-ch inbound-ch release-ch retry-ch]
    (go-loop []
             (try (let [msg (<!! parsed-ch)
                        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 Exception e
                    (taoensso.timbre/error e)
                    (throw e)))
             (recur)))

(defn buf-recv-loop [buf-recv-ch parsed-ch]
  (go-loop []
           (let [buf (<!! buf-recv-ch)]
             (try 
               (let [decompressed (protocol/read-buf buf)]
                 (>!! parsed-ch decompressed)
                 ; undo retain added in server handler
                 (.release ^ByteBuf buf))
               (catch Throwable t
                 (timbre/fatal t "Exception in recv loop.")))
                 (recur))))

(defrecord NettyTcpSockets [peer-group]
  component/Lifecycle

  [config peer-site peer-sites]
  (let [used-ports (->> (vals peer-sites) 
                        (filter 
                         (fn [s]
                           (= (:netty/external-addr peer-site) 
                              (:netty/external-addr s))))
                        (map :netty/port)
                        set)
        port (first (remove used-ports
