     (cond (= msg-type messages-type-id) 
           (read-messages-buf buf)
           (= msg-type ack-type-id) 
-          (if (<= ack-msg-length (.readableBytes buf))
-            (read-ack-buf buf))
+          (read-ack-buf buf)
           (= msg-type completion-type-id) 
-          (if (<= completion-msg-length (.readableBytes buf))
-            (read-completion-buf buf))
+          (read-completion-buf buf)
           :else (throw (Exception. (str "Invalid message type: " msg-type))))))
+
+

     :ack-val ack-val 
     :message message}))

(defn build-messages-msg-buf [allocator messages] 
  (let [compressed-messages (map (fn [msg]
                                   (update-in msg [:message] compress))
                                 messages)


(def ack-payload-length ^int (+ ack-msg-length type-header-length))

(defn build-ack-msg-buf [^ByteBufAllocator allocator id completion-id ack-val] 
  (let [^ByteBuf buf (.ioBuffer allocator ack-payload-length) #_(byte-buffer ack-payload-length)] 
    (.writeByte buf ack-type-id)
    (write-uuid buf id)
