ziggurat.messaging.consumer — tech.gojek/ziggurat 4.9.0

consumers clj

consumers is a map containing channel and consumer tags for stream and batch routes. One channel is used for all the subscribers and is preserved until shutdown. We have to preserve the channel because for cancelling a subscription, rabbitmq requires the same channel that created the consumer tag for the subscription.

convert-and-ack-message clj

(convert-and-ack-message ch
                         {:keys [delivery-tag]}
                         payload
                         ack?
                         topic-entity
                         ziggurat-channel-key)

De-serializes the message payload (payload) using nippy/thaw and converts it to MessagePayload. Acks the message if ack? is true.

DEFAULT_CHANNEL_PREFETCH_COUNT clj

delete-dead-set-messages clj

(delete-dead-set-messages topic-entity channel count)

This method deletes count number of messages from RabbitMQ dead-letter queue for topic topic-entity and channel channel.

get-dead-set-messages clj

(get-dead-set-messages topic-entity count)

(get-dead-set-messages topic-entity channel count)

This method can be used to read and optionally ack messages in dead-letter queue, based on the value of ack?.

For example, this method can be used to delete messages from dead-letter queue if ack? is set to true.

process-dead-set-messages clj

(process-dead-set-messages topic-entity count processing-fn)

(process-dead-set-messages topic-entity channel count processing-fn)

This method reads and processes count number of messages from RabbitMQ dead-letter queue for topic topic-entity and channel specified by channel. Executes processing-fn for every message read from the queue.

process-message-from-queue clj

(process-message-from-queue ch
                            meta
                            payload
                            topic-entity
                            processing-fn
                            ziggurat-channel-key)

read-message-from-queue clj

(read-message-from-queue ch queue-name topic-entity ack? ziggurat-channel-key)

start-channels-subscriber clj

(start-channels-subscriber channels topic-entity)

start-retry-subscriber* clj

(start-retry-subscriber* handler-fn topic-entity)

start-subscribers clj

(start-subscribers stream-routes batch-routes)

Starts the subscriber to the instant queue of the rabbitmq

stop-subscribers clj

(stop-subscribers subscribers additional-message)

stop-subscribers-for-consumers clj

(stop-subscribers-for-consumers consumers)