dvlopt.kstreams.ctx — dvlopt/kafka 1.3.1

Contexts are needed for low-level processors.

commit clj

(commit ctx)

Manually requests a commit of the current progress.

Commits are handle automatically by the library. This function request the next commit to happen as soon as possible.

forward clj

(forward ctx options)

Forwards a key-value to child nodes. Can be used in the :dvlopt.kstreams/processor.on-record function of a processor or during the execution of a function added with schedule.

A map of options may be given :

:dvlopt.kafka/key Unserialized key.

:dvlopt.kafka/timestamp Chosen timestamp. If missing, selects the timestamp of the record being processed by :dvlopt.kstreams/processor.on-record or the moment the scheduled function executes.

:dvlopt.kafka/value Unserialized value.

:dvlopt.streams/child Name of a specific child node. If missing, forwards to all children.

kv-store clj

(kv-store ctx store-name)

Retrieves a writable key-value store.

Cf. dvlopt.kstreams.store

schedule clj

(schedule ctx interval time-type callback)

Schedules a periodic operation for processors (may be used during :dvlopt.kstreams/processor.init and/or :dvlopt.kstreams/processor.on-record). Can be called multiple times on the same context.

The time interval must be as described in dvlopt.kafka (best effort for millisecond precision).

The time type is either :

:stream-time Time advances following timestamps extracted from records. An operation is skipped if stream time advances more than the interval.

:wall-clock-time Times advances following system time (best effort). An operation is skipped if garbage-collection halts the world for too long or if the current operation takes more time to complete than the interval.

The callback accepts only 1 argument, the timestamp of "when" it is called (depending on time type).

Returns a no-op function for cancelling the operation.

Ex. (schedule ctx [5 :seconds] :stream-time (fn callback [timestamp] (do-stuff-like-forwarding-records ...)))

session-store clj

(session-store ctx store-name)

Retrieves a writable session store.

Cf. dvlopt.kstreams.store

window-store clj

(window-store ctx store-name)

Retrieves a writable window store.

Cf. dvlopt.kstreams.store