       )
   
        
+(defn send-messages [{:keys [client]} 
+                     conf
+                     msgs]
+  (write! client (fn [^ByteBuf buff] 
+                   (with-size buff write-request conf msgs)
+                   )))
 
-(defn send-msg [topic partition codec msgs]
+  
+(defn producer [host port]
   (try 
-		  (let [
-		        buff (with-size (Unpooled/buffer 1024) write-request codec msgs)
-		        
-		        read-buff (ByteBuffer/allocate 1024)
-		        ch (SocketChannel/open (InetSocketAddress. "localhost" (int 9092)))
-		        write-int (.write ch (.nioBuffer buff))
-		        read-int (try (.read ch read-buff) (catch Exception e (do (prn e) -1)))
-		        ]
-		    (prn "Read int " read-int)
-		    (prn (java.util.Arrays/toString (-> read-buff .array)))
-		    )
-    (catch Exception e (.printStackTrace e))))
-		  
+  (let [c (client host port {:reuse-client true})]
+    (->Producer c host port))
+  (catch Exception e (.printStackTrace e))))
+  

(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 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 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
