 (ns ^:no-doc onyx.messaging.protocol-netty
-    (:require [taoensso.timbre :as timbre])
+    (:require [taoensso.timbre :as timbre]
+              [onyx.types :refer [->Leaf]])
     (: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)
  ; this could probably be a short
  (.writeInt buf (alength ^bytes message)) 
  (.writeBytes buf ^bytes message))

(defn read-message-buf [decompress-f ^ByteBuf buf]
  (let [id (take-uuid buf)
        acker-id (take-uuid buf)
        completion-id (take-uuid buf)
        ack-val (.readLong buf)
        message-size (.readInt buf)
        arr (byte-array message-size)
        _ (.readBytes buf arr)
        message (decompress-f arr)]
    {:id id 
     :acker-id acker-id 
     :completion-id completion-id 
     :ack-val ack-val 
     :message message}))

(defn build-messages-msg-buf [compress-f messages] 
  (let [compressed-messages (map (fn [msg]
                                   (update-in msg [:message] compress-f))
                                 messages)
        buf-size (+ type-header-length
                    (+ 4 ; message count int
                       (* (count messages) message-base-length)
                       ; nippy compressed segments
                       (apply + (map (fn [m]
                                       (alength ^bytes (:message m)))  
                                     compressed-messages))))
        buf ^ByteBuf (byte-buffer buf-size)] 
    (.writeByte buf messages-type-id) ; message type header
    (.writeInt buf (int (count compressed-messages))) ; number of messages
    (doseq [msg compressed-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)]
               (doall (repeatedly message-count #(read-message-buf decompress-f buf))))})

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

(def ^:const ack-base-length (int 40))

; message length without nippy segments
; 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))
