   (proxy [SimpleChannelInboundHandler]
     []
     (channelActive [^ChannelHandlerContext ctx]
-      (info "active")
       ;(.writeAndFlush ctx (Unpooled/copiedBuffer "Netty Rocks1" CharsetUtil/UTF_8))
       )
     (channelRead0 [^ChannelHandlerContext ctx ^ByteBuf in]



(defn- do-write [^Client client ^bytes v close-after-write {:keys [error-ch] :as conf}]
  (info "client " client)
  (try 
     (let [ch-f (-> client :channel-f (.channel) (.writeAndFlush v) (.addListener (exception-listener v conf)))]
       (if close-after-write
         (.addListener ch-f (close-listener client conf))))
     
     (catch Exception e (do 
                          (error e e)
                          (>!! error-ch [e v])))))


(defn start-client [host port {:keys [group read-ch error-ch write-ch handlers] :as conf 
                                 :or {group (NioEventLoopGroup.) read-ch (chan 100) error-ch (chan 100) write-ch (chan 100)}}]
  (try
  (let [g (if group group (NioEventLoopGroup.))
        b (Bootstrap.)]
    (-> b (.group g)
      (.channel NioSocketChannel)
      (.remoteAddress (InetSocketAddress. host port))
      (.handler (client-channel-initializer conf)))
    (let [ch-f (.connect b)]
      (.sync ch-f)
      (->Client g ch-f write-ch read-ch error-ch)))
  (catch Exception e (do
                       (error e e)
                       (>!! error-ch [e nil])))))
    
(defn read-print-ch [n ch]
  (go 
    (while true
      (let [c (<! ch)]
         (info n " = " (String. c))))))

(defn read-print-error [{:keys [error-ch]}]
  (read-print-ch "error" error-ch))

(defn read-print-in [{:keys [read-ch]}]
  (read-print-ch "read" read-ch))


(defn client [host port {:keys [handlers
                                  write-buff read-buff error-buff
                                  write-timeout read-timeout] 
                           :or {handlers [byte-encoder]
                                write-buff 100 read-buff 100 error-buff 1000 reuse-client false write-timeout 1500 read-timeout 1500} }]
  (let [ write-ch (chan write-buff) 
         read-ch (chan read-buff)

         g (NioEventLoopGroup.)
         conf {:group g :write-ch write-ch :read-ch read-ch :error-ch error-ch :handlers handlers}
         client (start-client host port conf) ]

    ;async read off error-ch
    (go 
      (loop [local-client client]
        (let [[v o] (<! error-ch)]
          (error "read error from error-ch " v)
          
          (if (instance? Poison v)
            (if local-client (close-client local-client)) ;poinson pill end loop
          (do
            ;on error, pause writing, and close client
	          (>! write-ch (->Pause 1000))
            (close-client local-client)
          
           (let [c 
                 (loop [acc 0] ;reconnect in loop
				            (if (>= acc retry-limit)
				              (do
                        ;if limit reached send poinson to all channels and call close all on client, end loop
				                (error "Retry limit reached, closing all channels and connections")
				                (go (>! write-ch (->Poison) ))
				                (go (>! read-ch (->Poison) ))
				                (go (>! error-ch (->Poison) ))
				                (close-all local-client)
			                  nil
				              )
					            (let [v1 
                             (try 
									              (let [c (start-client host port conf)
									                    reconnected (->Reconnected c v)]
				                          ;if connected, send Reconnected instance to all channels and return c, this c is assigned to the loop using recur
											                (.getAndIncrement ^AtomicInteger (:reconnect-count c))
											                (>! read-ch reconnected)
											                (>! write-ch reconnected)
									                    c)
									              (catch Exception e (do
									                                   (error (str "Error while doing retry " e) e)
                                                     e ;return exception to v, due to a bug in core async http://dev.clojure.org/jira/browse/ASYNC-48, we cannot recur here
				                                             )))]
                            (if (instance? Exception v1) ;if v is an exception recur
                              (recur (inc acc))
                              v1) ;else return the value (this is c, the connection)
                            )))]
                      
                      (if (instance? FailedWrite o) 
                        (do 
                            (info "retry failed write: ")
                            (<! (timeout 500))
                            (>! write-ch (:v o))))
                      
                      (recur c))
           
	          ))))) 
		                
                
              
     ;async read off write-ch     
     (go  
	      (loop [local-client client]
           (let [v (<! write-ch)]
		          (if (instance? Stop v) nil ;if stop exit loop
		             (do 
                   (try 
                      (cond (instance? Reconnected v) (do (recur (:client v))) ;if reconnect recur with the new client
					             (instance? Pause v) (do (<! (timeout (:time v))) (recur local-client)) ;if pause, wait :time millis, then recur loop
					             :else
				                (do ;else write the value to the client channel
                           (if (> (.get ^AtomicInteger (:reconnect-count local-client)) 0) (.set ^AtomicInteger (:reconnect-count local-client) 0))
						               (do-write local-client v false conf)))
							        (catch Exception e (do ;send any exception to the error-ch
								                            (error "!!!!! Error while writing " e)  
								                            (go (>! error-ch [e nil]))
                                    )))
                      (if (not (instance? Stop v))
                         (recur local-client)) ;if not stop recur loop
                      )))))
    
    
		    client))         


       
