 (def ^:const ack-base-length (int 40))
 
 ; message length without nippy segments
-; id (uuid), acker-id (uuid), completion-id (uuid), ack-val (long), nippy-byte-count (int)
-(def ^:const message-base-length (int 60))
+; id (uuid), acker-id (uuid), completion-id (uuid), ack-val (long)
+(def ^:const message-base-length (int 56))
 
 ; messages with 0 messages in it
 (def ^:const messages-base-length (int 4))

(ns ^:no-doc onyx.messaging.protocol-netty
    (:require [taoensso.timbre :as timbre])
    (:import [java.util UUID]
             [io.netty.buffer ByteBuf Unpooled UnpooledByteBufAllocator
              PooledByteBufAllocator ByteBufAllocator]))

  (write-uuid buf id)
  (write-uuid buf acker-id)
  (write-uuid buf completion-id)
  (.writeLong buf ack-val))

(defn read-message-buf [decompress-f ^ByteBuf buf message]
  (let [id (take-uuid buf)
        acker-id (take-uuid buf)
        completion-id (take-uuid buf)
        ack-val (.readLong buf)]
    ;;; Maybe should have a new record type for the read messages
    (->Leaf message id acker-id completion-id ack-val nil nil nil nil)))

(defn build-messages-msg-buf [compress-f messages] 
  (let [message-bytes ^bytes (compress-f (map :message messages))
        message-count (int (count messages))
        buf-size (+ type-header-length
                    (+ 4 ; message count int
                       (alength message-bytes)
                       (* message-base-length message-count)))
        buf ^ByteBuf (byte-buffer buf-size)] 
    (.writeByte buf messages-type-id) ; message type header
    (.writeInt buf message-count) ; number of messages
    (.writeInt buf (alength message-bytes)) ; nippy compressed data size
    (.writeBytes buf message-bytes) ; nippy compressed messages
    (doseq [msg messages]
      (write-message-msg buf msg))
    buf))

(defn read-messages-buf [decompress-f ^ByteBuf buf]
  {:type messages-type-id 
   :messages (let [message-count (.readInt buf)
                   messages-payload-size (.readInt buf)
                   arr (byte-array messages-payload-size)
                   _ (.readBytes buf arr)
                   messages (decompress-f arr)]
               (doall (map (fn [msg] 
                             (read-message-buf decompress-f buf msg)) 
                           messages)))})

(defn read-buf [decompress-f ^ByteBuf buf]
  (let [msg-type ^byte (.readByte buf)] 
