cues.queue — io.zalky/cues 0.2.1

Core Cues API.

add-attempt-hash clj

Only rebind in testing.

all-graph-messages clj

(all-graph-messages {queues :queues})

Eagerly gets all messages in the graph. Could be many!

all-messages clj

(all-messages queue)

Eagerly gets all messages in the cue. Could be many!

alts!! clj

(alts!! tailers)

Completes at most one blocking read from several tailers. Semantics of each read are that or read!!.

appender clj

(appender {q :queue-impl :as queue})

appender? clj

(appender? x)

attempt-index clj

Only rebind for tesitng!

close-and-delete-graph! clj

(close-and-delete-graph! g)

(close-and-delete-graph! g force)

Closes the graph and all queues and deletes all queue data.

close-and-delete-try-queues! clj

(close-and-delete-try-queues! g)

close-graph! clj

(close-graph! graph)

close-queue! clj

(close-queue! {q :queue-impl :as queue})

Closes the given queue.

close-tailer! clj

(close-tailer! tailer)

Closes the given tailer.

collect-graph-queues clj

(collect-graph-queues {:keys [processors queues]})

continue? clj

(continue? {unblock :unblock :as obj})

delete-all-queues! clj

(delete-all-queues!)

(delete-all-queues! queue-path)

Deletes all queue data at either the provided or default path.

Implementation note: must also purge the queue controller or blocking will break the next time the queue is made.

delete-queue! clj

(delete-queue! queue)

(delete-queue! queue force)

Deletes the queue data on disk, prompting by default.

Implementation note: must also purge the queue controller or blocking will break the next time the queue is made.

delete-unused-queue-files! clj

(delete-unused-queue-files! queue)

get-fork-queue clj

(get-fork-queue process)

get-try-queues clj

(get-try-queues process)

graph clj

(graph g)

graph-messages clj

(graph-messages {queues :queues} id)

Eagerly gets all messages in the queue. Could be many!

id->str clj

(id->str id)

index clj

Only rebind for testing!

index* clj

(index* {t :tailer-impl d :dirty :as tailer})

Gets the index at the tailer's current position. ChronicleQueue tailers do not update their current index after changing direction until AFTER the next read. Cues guards against this edge case by throwing an error if you attempt to take the index before the next read.

isomorphic? clj

(isomorphic? g1 g2)

last-index clj

Only rebind for testing!

last-index* clj

(last-index* x)

A last index implementation that works for any kind of queue or appender. Note that appenders and queues will not necessarily return the same result, and appenders will throw an error if they have not appended anything yet.

last-read-index clj

Only rebind for testing!

last-read-index* clj

(last-read-index* tailer)

log-processor-error clj

(log-processor-error {{:keys [id in alts out] :or {in "source" out "sink"}}
                        :config}
                     e)

messages clj

(messages tailer)

Returns a lazy list of all remaining messages.

mono-inc clj

(mono-inc old new)

Monotonically increases counter.

peek clj

(peek tailer)

Like read, but does not advance tailer.

persistent-attempt clj multimethod

Persists message to processor output queues, as well as attempt data to backing queue. Called once during each processor step. Taken together, persistent-snapshot, persistent-attempt, and persistent-recover implement message delivery semantics in Cues processors.

persistent-recover clj multimethod

Recovers processor tailer indices from backing queue once on processor start. Taken together, persistent-snapshot, persistent-attempt, and persistent-recover implement message delivery semantics in Cues processors.

persistent-snapshot clj multimethod

Persists processor tailer indices to a backing queue. Called once before each processor step. Taken together, persistent-snapshot, persistent-attempt, and persistent-recover implement message delivery semantics in Cues processors.

persistent-snapshot-alts clj multimethod

Persists the alts processor tailer id to a backing queue. Called immediately after the processor alts read. Taken together, persistent-snapshot, persistent-attempt, and persistent-recover implement message delivery semantics in Cues processors.

processor clj multimethod

Analogous to Kafka processors. No default method.

processor? clj

(processor? x)

queue clj

(queue id)

(queue id opts*)

Creates a persistent blocking queue.

Options include:

:transient A Boolean. If true, queue backing file will be rolled daily, and deleted after 10 days. You can configure alternative behaviour using the :roll-cycle, :cycle-release-tasks and :cycle-acquire-tasks options described below.

The following options are passed along to the underlying implementation:

:roll-cycle How frequently the queue data file on disk is rolled over. Default is :small-daily. Can be:

      :minutely, :daily, :test4-daily, :test-hourly, :hourly,
      :test-secondly, :huge-daily-xsparse, :test-daily,
      :large-hourly-xsparse, :large-daily, :test2-daily,
      :xlarge-daily, :huge-daily, :large-hourly, :small-daily,
      :large-hourly-sparse

:autoclose-on-jvm-exit? A Boolean. Whether to cleanly close the JVM on exit. Default is true.

:cycle-release-tasks Tasks to run on queue cycle release. For more, see qbits.tape.cycle-listener in the Tape library.

:cycle-acquire-tasks Tasks to run on queue cycle acquisition. For more, see qbits.tape.cycle-listener in the Tape library.

queue-closed? clj

queue-obj clj

Returns the underlying ChronicleQueue object.

queue-path-default clj

queue? clj

(queue? x)

read clj

(read tailer)

Reads message from tailer without blocking. Materializes metadata in message.

read!! clj

(read!! tailer)

Blocking read. The tailer will consume eagerly if it can. If not, it will block until a new message is available. Implementation note: it is critical that the watch be placed before the first read. Note that the tailer index will always be ahead of the last index it read. Read will continue and return nil if at any point an unblock signal is received. This allows blocked threads to be cleaned up on unblock events.

send! clj

(send! graph msg)

(send! {p :processors config :config} source msg)

set-direction clj

(set-direction {t :tailer-impl d :dirty :as tailer} direction)

Sets the direction of the tailer to either :forward or :backward. Note: after changing the direction on the tailer you must do a read before you can measure the index again via cues.queue/index. This is an artifact of ChronicleQueue behaviour.

start-graph! clj

(start-graph! graph)

start-processor! clj

(start-processor! {:keys [id state impl] :as process})

start-processor-impl clj multimethod

stop-graph! clj

(stop-graph! graph)

stop-processor! clj

(stop-processor! {:keys [id state impl] :as process})

tailer clj

(tailer queue)

(tailer queue id)

(tailer queue id unblock)

Creates a tailer.

Providing an id enables tailer position on the queue to persist across restarts. You can also optionally provide an unblock reference that when set to true will prevent the tailer from blocking. This is typically used to unblock and clean up blocked processor threads.

tailer? clj

(tailer? x)

to-end clj

(to-end {t :tailer-impl d :dirty :as tailer})

Moves the tailer to the end of the queue.

to-index clj

Only rebind for testing!

to-index* clj

(to-index* {t :tailer-impl d :dirty :as tailer} i)

to-start clj

(to-start {t :tailer-impl d :dirty :as tailer})

Moves the tailer to the beginning of the queue.

topology clj

(topology graph)

transient-queue-opts clj

Transient queues are purged after 10 days.

unhandled-error clj macro

(unhandled-error msg & body)

unused-queue-files clj

(unused-queue-files queue)

with-tailer clj macro

(with-tailer bindings & body)

write clj

(write appender msg)

The queue controller approximately follows the index of the queue: it can fall behind, but must be eventually consistent.

written-index clj

Returns the index written by the appender. Only rebind for testing!