ziggurat.messaging.producer — tech.gojek/ziggurat 4.8.0

borrow-from-pool clj

(borrow-from-pool pool)

delay-queue-name clj

(delay-queue-name topic-entity queue-name)

get-channel-delay-exchange-name clj

(get-channel-delay-exchange-name topic-entity channel message-payload)

This function return delay exchange name for retry when using channel flow. It will return exchange name with retry count as suffix if exponential backoff enabled.

get-channel-queue-timeout-ms clj

(get-channel-queue-timeout-ms topic-entity channel message-payload)

Calculate queue timeout for channel delay queue. Uses the value from [[get-exponential-backoff-timeout-ms]] if exponential backoff enabled.

get-channel-retry-count clj

(get-channel-retry-count topic-entity channel)

get-delay-exchange-name clj

(get-delay-exchange-name topic-entity message-payload)

This function return delay exchange name for retry when using flow without channel. It will return exchange name with retry count as suffix if exponential backoff enabled.

get-queue-timeout-ms clj

(get-queue-timeout-ms message-payload)

Calculate queue timeout for delay queue. Uses the value from [[get-exponential-backoff-timeout-ms]] if exponential backoff enabled.

make-queues clj

(make-queues routes)

MAX_EXPONENTIAL_RETRIES clj

publish clj

(publish exchange message-payload)

(publish exchange message-payload expiration)

(publish exchange message-payload expiration retry-counter)

This is meant for publishing to rabbitmq.

• Checks if the pool is alive - We do this so that publish does not happen after the channel pool state is stopped.
• publish-internal returns multiple states
  • :success - Message has been successfully produced to rabbitmq
  • :retry - A retryable exception was encountered and message will be retried until it is successfully published.
  • :retry-with-counter - A non recoverable exception is encountered, but the message will be retried for a few times. defined by the counter { :rabbit-mq-connection { :publish-retry { :non-recoverable-exception {:count}}}}}

publish-to-channel-dead-queue clj

(publish-to-channel-dead-queue channel message-payload)

publish-to-channel-delay-queue clj

(publish-to-channel-delay-queue channel message-payload)

publish-to-channel-instant-queue clj

(publish-to-channel-instant-queue channel message-payload)

publish-to-dead-queue clj

(publish-to-dead-queue message-payload)

publish-to-delay-queue clj

(publish-to-delay-queue message-payload)

publish-to-instant-queue clj

(publish-to-instant-queue message-payload)

retry clj

(retry {:keys [retry-count] :as message-payload})

retry-for-channel clj

(retry-for-channel {:keys [retry-count topic-entity] :as message-payload}
                   channel)

return-to-pool clj

(return-to-pool pool ch)