 	        )))))
 	    
     
-(defn write-request [^ByteBuf buff codec msgs]
+(defn write-request [^ByteBuf buff {:keys [correlation-id client-id codec acks timeout] :or {correlation-id 1 client-id "1" codec 0 acks 1 timeout 1000}}
+                     msgs]
     (-> buff
-      (.writeShort (short 0))   ;api-key
-      (.writeShort (short 0))   ;version api
-      (.writeInt (int 1))       ;correlation id
-      (put-short-string "cid")  ;short + client-id bytes
-      (.writeShort (short 1))   ;acks
-      (.writeInt (int 1000)))    ;timeout
+      (.writeShort  (short API_KEY_PRODUCE_REQUEST))   ;api-key
+      (.writeShort  (short API_VERSION))   ;version api
+      (.writeInt (int correlation-id))       ;correlation id
+      (put-short-string client-id)  ;short + client-id bytes
+      (.writeShort (short acks))   ;acks
+      (.writeInt (int timeout)))    ;timeout
     
       (let [topic-group (group-by :topic msgs)]
         (.writeInt buff (int (count topic-group))) ;topic count

(ns kafka-clj.produce3
  (:require [kafka-clj.codec :refer [crc32-int get-compress-out compress]])
  (:import [java.nio ByteBuffer]
           [io.netty.buffer ByteBuf Unpooled]
           [java.nio.channels SocketChannel]
           [java.net InetSocketAddress]
           [kafka_clj.util Util]))

(defonce ^:constant compression-code-mask 0x03)

(defn put-short-string [^ByteBuf buff s]
  (-> buff 

      )
  
       
(defn send-messages [{:keys [client]} 
                     conf
                     msgs]
  (write! client (fn [^ByteBuf buff] 
                   (with-size buff write-request conf msgs)
                   )))

  
(defn producer [host port]
  (try 
  (let [c (client host port {:reuse-client true})]
    (->Producer c host port))
  (catch Exception e (.printStackTrace e))))
  
