fast tcp client built on netty
Clojure Java
Latest commit 38da0ad Jun 1, 2016 @gerritjvv update versions
Permalink
Failed to load latest commit information.
clj-tcp update versions Jun 1, 2016
.gitignore Initial commit Dec 19, 2013
LICENSE Initial commit Dec 19, 2013
README.md add connection monitoring, reformat code, and move to netty 4.1 BETA Feb 6, 2015

README.md

clj-tcp

A fast async tcp client library built on netty.

There are many good async http libraries out there, but no lowlevel tcp library that allows you to build your own protocols on top of it. This library aims to provide this, you send, bytes and receives bytes, what the bytes mean is up to the user.

The code uses clojure.core.async heavily to transmit messages for read, write and exceptions.

Contact: feel free to contact me at gerritjvv@gmail.com

Usage

The binaries are published to https://clojars.org/clj-tcp

Clojars Project

(use 'clj-tcp.echo-server :reload)
(def s (start-server 2324))
; this starts a test server, that reads echo and closes the channel

(use 'clj-tcp.client :reload)

(def c (client "localhost" 2324 {:reuse-client false}))
(read-print-in c)
(write! c (.getBytes "hi"))
;; read ...
;; any subsequent write! calls will find a closed channel, the connection will reconnect, and retry the send.

;;close the connection and all its resources
(close-all c)
;;Future this returns a future

(close-and-wait c)
;;calls close-all and waits on the future

Netty and EventLoopGroup

By default all client connections will share a global EventLoopGroup instance with threads=cpus/2 daemon=true.
This lowers resource usage, makes client creation faster and is the recommended way for netty.

See http://normanmaurer.me/presentations/2014-facebook-eng-netty/slides.html#1.0 for a best practices overview.

If you need to override this behaviour and create your own EventLoopGroup for a client,
the client function allows this for both writing and reading.

;;this will create a EventLoopGroup(1) for writing and a EventLoopGroup(2) for reading.
;;on close these EventLoopGroup(s) will be shutdown.
(def c (client "localhost" 2324 {:write-group-threads 1 :read-group-threads 2}))

(close-and-wait c)

Error handling

The connection is 100% asynchronous meaning that errors will also appear and be notified asynchronous. Each connection provides a channel on which exceptions are notified.

Note:

If too many exceptions are thrown, and the exceptions are not taken from the error channel, the connection will eventually block, i.e. in production you must listen for exceptions.

Reading using go loops (preferred)

;reading and printing exceptions
(require '[clojure.core.async :refer [go <! ]])
(go-loop []
  (when-let [[e v] (<! (:error-ch c))]
    (prn "Error " e) (.printStackTrace e)
    (recur)))

Reading using blocking methods

;;inside another thread
(while true
   (if-let [[e v] (read-error c 1000)]
      (prn "Error " e)))

Monitoring

The package clj-tcp.monitor has a global variable (with weak references to avoid a memory leak)
named monitor-ctx.

Functions like report-counts can be used to examine the context and monitor the numbre of open connections

Configuration

The following config options can be passed to the client

(import 'io.netty.channel.ChannelOption)

(def c (client "localhost" 2324 { ;:decoder (fn [read-ch ^ByteBuf in] (>!! read-ch "hi")) decoder allows you to overwrite the default ByteBuf to bytes decoding.
                                  :handlers [default-encoder] ; ChannelHandlers that will be added to the Channel pipeline
                                  channel-options [[ChannelOption/TCP-NODELAY true][ChannelOption/SO_RCVBUF (int 5242880)]] ;io.netty.channel options a sequence of [option val] e.g. [[option val] ... ]
                                  retry-limit 5 ; on write error the client will retry the write this amount of times
                                  write-buff 100 ; writes are async, this is the buffer thats used for the clojure.async.channel
                                  read-buff  100 ; reads are written to the read channel, the buffer is specified here
                  error-buff 100 ; errors are sent to the error-ch, the buffer is specified here
                                  write-group-threads 4 ;threads for writing
                                  read-group-threads 4  ;threads for reading
        ))

Memory

By default and for ease of use the decoder used by the client copies the ByteBuf content to a byte array.
This is not very memory efficient, but for allot of application doesn't matter.
If you want to avoid this copying and directly use the ByteBuf specify the following decoder:

(defn copy-bytebuf [read-ch ^ByteBuf buff]
  (>!! read-ch (.copy buff)))

Contact

Email: gerritjvv@gmail.com

Twitter: @gerrit_jvv

License

Distributed under the Eclipse Public License either version 1.0