kafka-clj.tcp — kafka-clj 4.0.3

Simple Direct TCP Client for the producer The producers sit behind an async buffer where data is pushed on by multiple threads, the TCP sending itself does not need yet another layer or indirection which at the current moment under high loads with huge messages can cause out of memory errors

Usage (def client (tcp/tcp-client "localhost" 7002)) (tcp/write! client "one two three" :flush true) (tcp/read-async-loop! client (fn [^bytes bts] (prn (String. bts)))) (tcp/close! client)

Provides a pooled connection via tcp-pool the config options supported are from GenericKeyedObjectPoolConfig

borrow clj

(borrow obj-pool host port)

(borrow obj-pool host port timeout-ms)

check-sasl clj

(check-sasl {:keys [jaas kafka-version] :as conf} fqdn conn)

Returns c assoc :sasl-ctx where sasl-ctx contains or not the connected sasl connection if jaas is specified

close! clj

(close! {:keys [socket input output sasl-ctx]})

close-pool! clj

(close-pool! obj-pool)

closed? clj

(closed? {:keys [socket sasl-ctx]})

driver clj

(driver hosts
        &
        {:keys [routing-conf pool-conf retry-limit]
         :or {retry-limit 2 routing-conf {} pool-conf {}}})

if jaas is specified in pool-conf it must point to a configuration section on the jaas and kerberos files defined as env properties -Djava.security.auth.login.config=/vagrant/vagrant/config/kafka_client_jaas.conf -Djava.security.krb5.conf=/vagrant/vagrant/config/krb5.conf

if jaas is specified and kafka 0.9.0 is used add :kafka-version "0.9.0" to the conf

fqn clj

(fqn socket)

init-tcp-conn clj

(init-tcp-conn conf conn)

open-socket clj

(open-socket host port {:keys [timeout-ms] :or {timeout-ms 10000}})

pool-obj-val clj

(pool-obj-val pooled-obj)

Returns the PoolObj contained value

read-bts clj

(read-bts input timeout cnt)

read-int clj

(read-int input timeout)

read-response clj

release clj

(release obj-pool host port v)

tcp-client clj

(tcp-client host port conf)

Creates a tcp client from host port and conf InputStream is DataInputStream(BufferedInputStream) and output is BufferedOutputStream

if jaas is specified it must point to a configuration section on the jaas and kerberos files defined as env properties -Djava.security.auth.login.config=/vagrant/vagrant/config/kafka_client_jaas.conf -Djava.security.krb5.conf=/vagrant/vagrant/config/krb5.conf

if jaas is specified and kafka 0.9.0 is used add :kafka-version "0.9.0" to the conf

wrap-bts clj

(wrap-bts bts)

Wrap a byte array in a ByteBuf

wrap-exception clj

(wrap-exception f & args)

write! clj