ziggurat.messaging.producer — tech.gojek/ziggurat 4.9.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)

(publish exchange
         message-payload
         is-payload-serialized?
         expiration
         retry-counter
         topic-entity)

This is meant for publishing to rabbitmq.

• Supports publishing both serialized and deserialized messages. If is-payload-serialized is true, the message won't be frozen before publishing to rabbitmq, else it would be frozen via nippy.
• Use case(s) of sending serialized-payloads
  1. If a subscriber encounters an exception, while deserializing the message it receives from instant queue, the serialized payload is sent as is to the dead-letter-queue.
  2. If a subscriber encounters any exception while processing the message, the serialized payload is sent as is to the dead-letter-queue
• 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-dead-queue channel
                               message-payload
                               topic-entity
                               is-payload-serialized?)

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-dead-queue message-payload topic-entity is-payload-serialized?)

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)