gerritjvv/tcp-driver

Java/Clojure TCP Connections done right

tcp-driver's Issues

README update with black-list add host remove host to driver

In the readme.md and driver namespace add public methods and document them for blacklist, add and remove hosts to the routing policy using the tcp-driver.routing.policy namespace

Close pool on shutdown handler

This should be configurable on default == false, but if set each connection/pool should register a shutdown hook that closes the pool on JVM shutdown

Test pools and streams

Write tests for the pools and streams namespace

Consistent Hash rounting

Add consistent hash routing.

The default routing is random which generically works best, but for some schemes a consistent hash works better.
This does require the data to be sent via the send-op to contain some kind of format.

Requires: #2

Create generic Data wrapper for send-op

The data we send should support the following protocol:

IData
  -msg-bytes [] : byte-array ;; return the message as bytes
  -msg-key [] : byte-array  ;; get the message key, used for routing
  -clone [key:bytes, msg:bytes] : IData ;; create a new message using the same implementation

Provde default implementations using:

String
Number
byte[]

Implement conn and send-op

Require: #4

Add create conn, send-op and add retry and routing policy mechanisms.

create-conn [ bootstrap-hosts pool-conf routing-obj retry-policy conf]
   { :pool (pool
     :conf conf
     :bootstrap-hosts bootstrap-hosts
     :retry-policy retry-policy
     :routing-policy routing-obj
     :routing-env {:hosts (atom (set boostrap-hosts))}



 add-host! [ ctx host host-conf]
    (assoc-stm! (get-in ctx [:routing-env :hosts]) host)

 remove-host! [ctx host host-conf]
   (dissoc-stm! (get-in ctx [:routing-env :hosts]) host)

 send-op [ctx host i-o-f data timeout-ms]
    try:
       conn = (pool/borrow (:pool ctx) host timeout-ms)
       try:
          return (i-o-f conn data timeout-ms)
       finally:
          (pool/return pool-inst conn)

     catch Exception e
           (pool/invalidate pool-inst conn)
           return e


 send [ ctx i-o-f data timeout-ms]

   hosts = (:bootstrap-hosts ctx)
   routing-policy (:routing-policy ctx)


   loop [retry-state  nil]
        host = (routing/select-host routing-policy (:routing-env ctx) hosts)

        if host == null
           throw NoHostAvailableException

        ret-val =  (send-op ctx host i-o-f data timeout-ms)

        if ret-val == Exception
           [retry-state retry-op] = (io-retry/retry-action! (:retry-policy ctx) (:conf ctx) conn host retry-state)
           if retry-op == :recur
             (recur retry-state)
           else
              (throw ret-val)
        else
           ret-val

    NS: rounting
      Protocol IRounting
        select-host 
        add-host
        remove-host

    NS: io-retry
    multi-method retry-action! identity

    defmulti retry-action! :retry           [_ _ _ _ retry-state] 
                                                [retry-state (if (available-tries? retry-state) :retry nil)]

    defmulti retry-action! :blacklist-retry [ctx conn host e retry-state]
                                                (routing/blacklist! (:routing-policy ctx) (:routing-env ctx) conn host)
                                                [retry-state (if (hosts-left? ... ) :retry :nil)]

    defmulti retry-action! :default [ _ _ _ e _ ] nil

Add blacklisting

Add ability to black list a host as part of the public api