returns the timeunit

(q/->timeunit :ms) => java.util.concurrent.TimeUnit/MILLISECONDS

returns all available stacktraces

creates a executor that takes in an atom queue

creates an atom with a vec as queue

processes

(def +state+ (atom []))

(aq:process (fn elems(swap! +state+ conj elems)) (atom 1 2 3 4 5) 3)

@+state+ => [1 2 3 4 5]

submission function for one or multiple entries to aq

allows query inputs to be combined

capture all inputs on the client

map function across a client

creates a transaction within a bulk context

creates and starts a bus

returns all registered ids

(bus:with-temp bus (bus:all-ids bus)) => (contains #{string?})

returns all registered threads

(bus:with-temp bus (= (first (vals (bus:all-threads bus))) (Thread/currentThread))) => true

bus:closes all bus:opened loops

(bus:with-temp bus (let [{:keys stopped id} @(bus:open bus (fn m (update m :value inc)))] (Thread/sleep 10) (bus:close bus id) @stopped (bus:get-count bus))) => (contains-in {:exit :normal, :id string? :unprocessed empty? :start number? :end number?} 1)

stops all thread loops

creates a bus

deregisters from the bus

returns the number of threads registered

gets registered id given thread

(bus:with-temp bus (bus:get-id bus)) => string?

gets the message queue associated with the thread

(bus:with-temp bus (vec (bus:get-queue bus))) => []

gets thread given an id

(bus:with-temp bus (->> (bus:get-id bus) (bus:get-thread bus))) => (cc/thread:current)

checks that the bus has a given id

bus:closes all bus:opened loops

stops all thread loops

bus:opens a new handler loop given function

(bus:with-temp bus (let [{:keys id} @(bus:open bus (fn m (update m :value inc)))] (Thread/sleep 100) @(bus:send bus id {:value 1}))) => (contains {:value 2, :id string?})

registers a thread to the bus

resets the counters for a bus

sends a message to the given thread

bus:sends message to all thread queues

bus:waits on the message queue for message

(bus:with-temp bus (bus:send bus (bus:get-id bus) {:op :hello :message “world”}) (bus:wait bus)) => (contains {:op :hello, :message “world”, :id string?})

(bus:with-temp bus (bus:wait bus {:timeout 100}))

checks if object is instance of Bus

checks if object is instance of Bus

constructs a blocking deque

(q/deque 1 2 3) => java.util.concurrent.LinkedBlockingDeque

checks if object is a BlockingDeque

drains elements to another vector

((juxt #(q/drain % 2) #(into [] %)) (q/queue 1 2 3 4)) => [1 2 3 4]

checks if executor is at capacity

await termination for executor service

returns number of active threads in pool

returns number of completed tasks

returns number of threads in pool

returns number of submitted tasks

gets the queue from the executor

(-> (executor:pool 10 10 1000 (q/queue)) (exec:get-queue))

increases the capacity of the executor

gets and sets the keep alive time

gets and sets the core pool max

gets and sets the core pool size

contructs a raw queue in different ways

(exec:queue)

(exec:queue 1)

(exec:queue {:size 1})

(exec:queue {})

(exec:queue (q/queue))

sets the rejected task handler

shuts down executor

shuts down executor immediately

checks if executor is shutdown

checks if executor is shutdown and all threads have finished

check that executor is terminating

creates an executor

(executor {:type :pool :size 3 :max 3 :keep-alive 1000})

creates a cached executor

(executor:cached)

returns health of the executor

returns executor service info

constructs a pool executor

(executor:pool 10 10 1000 {:size 10})

returns props for getters and setters

constructs a scheduled executor

(executor:scheduled 10)

registers a shared executor

lists all shared executors

(-> (executor:shared) keys sort) => :async :default :pooled

constructs a single executor

;; any sized pool (executor:single)

;; fixed pool (executor:single {:size 10})

returns executor service type

deregisters a shared executor

gets the common stream bus

creates a hub state

adds entries to the hub

(hub:add-entries (hub:new) 1 2 3 4 5) => (contains f/future? 0 5)

creates a hub based executor

(def -exe- (hub:executor nil {:handler (fn & args args) :interval 50 :max-batch 1000}))

(do (def -res- ((:submit -exe-) 1 2 3 4 5 6)) @(first -res-)) => ’(nil (1 2 3 4 5 6))

(hub:wait (:queue -exe-)) => nil

creates a trackable atom queue

like aq:process but with a hub

(def +state+ (atom []))

(hub:process (fn elems(swap! +state+ conj elems)) (hub:new 1 2 3 4 5) 3)

@+state+ => [1 2 3 4 5]

submission function for the hub

waits for the hub executor to be ready

takes element at the front of the queue

(q/peek (q/queue))

peeks from the front of the queue

peeks from the back of the queue

creates and starts the pool

(def -p- (pool {:size 2 :max 10 :keep-alive 10000 :poll 20000 :resource {:create (fn ) :initial 0.3 :thread-local true}}))

acquires a resource from the pool

(pool:acquire |pool|) => (contains string? ’)

runs cleanup on the pool

creates an initial pool

(pool:create {:size 5 :max 8 :keep-alive 10000 :poll 20000 :resource {:create (fn [] ’) :initial 0.3 :thread-local true}})

disposes an idle object

(pool:dispose |pool| (first (keys (pool:resources:idle |pool|))))

disposes if idle and busy are over size limit

marks the current resource for dispose

unmarks the current resource for dispose

releases a resource back to the pool

(let [id _(pool:acquire |pool|)] (pool:release |pool| id)) => string?

returns all the busy resources

returns all the idle resources