kafka-clj.pool.atom — kafka-clj 4.0.3

Object pool using Sempahore and Clojure's atom

Use create-atom-obj-pool to create an atom object pool These functions as made to be used primarily from the kafka-clj.pool.keyed namespace and the kafka-clj.pool.api is used to work with the keyed pool

a-available? clj

(a-available? {:keys [sem]})

Number of objects available in the atom pool

a-destroy clj

(a-destroy {:keys [destroy-f sem ctx]} v)

Run (destroy-f ctx v) and will always release a permit on the semaphore even if destroy-f throws an exception This function does not check if v was attained from the pool, this is the responsibility of the calling program

Note: destroy-f is only called if (pool-obj-val v) is non nil

a-destroy-idle-timeouts! clj

(a-destroy-idle-timeouts! pool timeout-ms)

a-destroy-timeouts! clj

(a-destroy-timeouts! {:keys [queue closed-atom] :as pool}
                     should-destroy?
                     timeout-ms)

Destroy pool objects timeout

a-destroy-ttl-timeouts! clj

(a-destroy-ttl-timeouts! pool timeout-ms)

a-poll clj

(a-poll {:keys [ctx queue create-f sem closed-atom]} timeout-ms)

arg: AtomPool return a PoolObj

a-poll-check-or-create clj

(a-poll-check-or-create ctx x create-f)

If x is nil we call create-f in a delay. returns a delayed object, this function runs inside swap! so needs to be idempotent

a-poll-valid clj

(a-poll-valid pool timeout-ms)

Poll for a valid object, invalid objects are destroyed, this process is repeated till a valid object is found or timeout-ms have expired

Note: validate-f is only called if (pool-obj-val v) is non nil

a-pool clj

(a-pool ctx create-f validate-f destroy-f)

create-f when called should create a valid object called as (create-f ctx), validate-f is called before poll returns, called (validate-f ctx v) destroy-f is called to destroy an object (destroy-f ctx v) sem is a semaphore setting up the limit = (get ctx :pool-limit 20)

a-pool-close-all clj

(a-pool-close-all {:keys [queue closed-atom] :as pool})

a-pool-stats clj

(a-pool-stats pool)

Return a map {:available <available objs> :waiting-threads <threads waiting to acquire an obj>}

a-return clj

(a-return {:keys [queue sem]} v)

We add v back into the pool, and do not change the head v must be an instance of PoolObj

a-validate-safely clj

(a-validate-safely {:keys [ctx validate-f]} v)

Call validate in a try catch, if any exception nil is returned

a-waiting-threads clj

(a-waiting-threads {:keys [sem]})

Returns an estimate of the number of threads waiting to acquire

atom-closed? clj

(atom-closed? closed)

atom-get-and-set! clj

(atom-get-and-set! closed)

create-atom-obj-pool clj

(create-atom-obj-pool ctx create-f validate-f destroy-f)

destroy-all-objs! clj

(destroy-all-objs! pool objs)

Get timedout objecs from any idle objects and call destroy on them, this is a NOOP if the pool is closed