kafka-component.mock — kafka-component 0.6.3

->topic-partition clj

(->topic-partition topic partition)

accumulate-messages clj

(accumulate-messages consumer topic options)

Like accumulate-subscribed-messages, but subscribes to the topic first.

accumulate-subscribed-messages clj

(accumulate-subscribed-messages
  consumer
  {:keys [timeout at-least-n format-fn filter-fn]
   :or {at-least-n 1 format-fn identity filter-fn identity}})

Helper to consume messages off any existing subscriptions. Will return as soon as at-least-n messages have been fetched. May return more, since an individual poll can fetch more. If timeout (in ms) expires first, returns any messages fetched so far. Mesages may be reformatted with format-fn then filtered with filter-fn. Messages excluded by filter-fn do not count towards at-least-n.

add-record-in-broker-state clj

(add-record-in-broker-state state producer-record)

add-record-to-topic clj

(add-record-to-topic state producer-record)

assert-producer-not-closed clj

(assert-producer-not-closed producer-state)

Checks conn-open? in producer state

assert-proper-record clj

(assert-proper-record record)

assign-partitions clj

(assign-partitions broker-state
                   consumers
                   participants
                   participants-ch
                   complete-ch)

broker-create-topic clj

(broker-create-topic)

(broker-create-topic num-partitions)

broker-ensure-topic clj

(broker-ensure-topic broker-state topic)

broker-receive-messages clj

(broker-receive-messages state msg-ch)

broker-save-record! clj

(broker-save-record! state record)

broker-state clj

buffer-size clj

close-all-from clj

(close-all-from ch)

close-mock clj

(close-mock state)

committed-offsets clj

committed-record-metadata clj

(committed-record-metadata record)

consumer-backoff clj

consumer-coordinator clj

(consumer-coordinator state broker-state join-ch leave-ch)

consumer-rebalance-timeout clj

consumer-unsubscribe-timeout clj

consumers-with-topic-overlap clj

(consumers-with-topic-overlap consumers topics)

debug clj

debug! clj

(debug! enable)

default-mock-consumer-opts clj

default-num-partitions clj

fixture-restart-broker! clj

(fixture-restart-broker! f)

get-messages clj deprecated

(get-messages consumer timeout)

(get-messages consumer topic timeout)

DEPRECATED: use txfm-messages instead. Poll until consumer receives some messages on topic. If timeout (in ms) expires first, return an empty vector.

get-offset clj

(get-offset broker-state topic partition config)

goe clj/s macro

(goe & body)

goe-loop clj/s macro

(goe-loop & body)

intersects? clj

(intersects? v1 v2)

IRebalance clj protocol

all-topics clj

(all-topics this)

apply-pending-topics clj

(apply-pending-topics this topics)

clean-up-subscriptions clj

(clean-up-subscriptions this)

logger clj

(logger & args)

max-poll-records clj

(max-poll-records config)

mock-consumer clj

(mock-consumer config)

(mock-consumer auto-subscribe-topics config)

mock-producer clj

(mock-producer config)

noop-cb clj

producer-record->consumer-record clj

(producer-record->consumer-record offset record)

read-offsets clj

(read-offsets grouped-messages)

rebalance-consumers clj

(rebalance-consumers relevant-consumers broker-state)

rebalance-participants clj

(rebalance-participants broker-state consumers participants-ch complete-ch)

Try to get all the consumers to participate in the rebalance, but if they don't all check in, continue without some of them.

rebalance-participants-timeout clj

record->clj clj

(record->clj record)

record->topic-partition clj

(record->topic-partition record)

records->clj clj

(records->clj consumer-records)

reset-state! clj

(reset-state!)

send clj

(send producer topic k v)

send-async clj

(send-async producer topic k v)

shutdown! clj

(shutdown!)

standalone-mock-consumer-opts clj

start! clj

(start!)

txfm-messages clj

(txfm-messages consumer topic xf)

(txfm-messages consumer topic xf options)

Like txfm-subscribed-messages, but subscribes to the topic first.

txfm-subscribed-messages clj

(txfm-subscribed-messages consumer xf)

(txfm-subscribed-messages consumer
                          xf
                          {:keys [timeout ixf rf]
                           :or {timeout 1000 ixf cat rf conj}})

Helper to txfm messages sent to the consumer on any existing subscriptions. Messages will be transformed by the transducing function xf. Will return as soon as xf indicates it has enough records, or the timeout (in ms, by default 1000) expires. Thus, to avoid waiting for the timeout, include e.g. (take 3) in xf. If the timeout expires, will return any messages transformed so far.

For example,

(kafka-mock/send producer "topic" "key" "5")
(kafka-mock/send producer "topic" "key" "6")
(kafka-mock/send producer "topic" "key" "7")
(kafka-mock/send producer "topic" "key" "8")
(.subscribe consumer ["topic"])
(kafka-mock/txfm-subscribed-messages consumer (comp (map :value)
                                                    (map #(Integer/parseInt %))
                                                    (filter even?)
                                                    (take 1)))
;; => [6]

By default messages retrieved from (.poll consumer) will be collected as by clojure.core/cat and fed one-by-one through xf. This behvaior can be altered by providing a different ixf. For example, clojure.core/conj will feed each batch through in its entirety. This can be combined with max.poll.records to control batch size.

Also by default, the transformed messages will be accumulated as by clojure.core/conj, and returned. A different reducing function rf can be supplied. For example, if xf produces a stream of numbers, the reducing function clojure.core/+ will return their sum.

with-test-broker clj/s macro

(with-test-broker & body)

with-test-producer-consumer clj/s macro

(with-test-producer-consumer producer-name consumer-name & body)