topology and codec

Two public modules in queuey-rabbitmq hold the naming and encoding rules. Everything in them is pure: no connection, no I/O, no async. They exist so tooling can compute the same names and read the same headers the backend does.

Reach them through the facade as queuey::rabbitmq::topology and queuey::rabbitmq::codec.

topology

Constants

pub const DEFAULT_DEAD_SUFFIX: &str = ".dead";
pub const DEFAULT_DEFERRED_SUFFIX: &str = ".deferred";

pub const HEADER_DEATH_REASON: &str = "x-death-reason";
pub const HEADER_ORIGINAL_QUEUE: &str = "x-original-queue";
pub const HEADER_ATTEMPTS: &str = "x-attempts";
pub const HEADER_ATTEMPT: &str = "x-attempt";
pub const HEADER_DEFERRALS: &str = "x-deferrals";

pub const ARG_DEAD_LETTER_EXCHANGE: &str = "x-dead-letter-exchange";
pub const ARG_DEAD_LETTER_ROUTING_KEY: &str = "x-dead-letter-routing-key";
pub const ARG_MESSAGE_TTL: &str = "x-message-ttl";
pub const ARG_MAX_PRIORITY: &str = "x-max-priority";
pub const ARG_EXPIRES: &str = "x-expires";

pub const MAX_TTL_MS: u32 = u32::MAX;             // about 49.7 days
pub const MAX_DEFERRAL_MS: u32 = MAX_TTL_MS / 2;  // about 24.8 days

x-attempts is on dead letters; x-attempt is on live messages.

Functions

pub fn dead_queue_name(queue: &str, suffix: &str) -> String;
pub fn deferred_queue_name(queue: &str, suffix: &str, ttl_ms: u32) -> String;
pub fn deferred_ttl_ms(delay: Duration, granularity: Duration) -> Option<u32>;
pub fn queue_args(config: &QueueConfig) -> FieldTable;
pub fn deferred_queue_args(config: &QueueConfig, ttl_ms: u32) -> FieldTable;
pub fn dead_queue_args(_config: &QueueConfig) -> FieldTable;

Function	Behaviour
`dead_queue_name`	`"{queue}{suffix}"`
`deferred_queue_name`	`"{queue}{suffix}.{ttl_ms}"`
`deferred_ttl_ms`	Rounds `delay` up to a whole multiple of `granularity`, at least one step. `None` when the result exceeds `MAX_DEFERRAL_MS`. A granularity below 1 ms is clamped to 1 ms
`queue_args`	`x-message-ttl` when `message_ttl` is set, `x-max-priority` when `max_priority` is `Some`
`deferred_queue_args`	`x-message-ttl = ttl_ms`, `x-dead-letter-exchange = ""`, `x-dead-letter-routing-key` the work queue, `x-expires = 2 * ttl_ms`
`dead_queue_args`	Always empty

deferred_ttl_ms returning None is what produces the “longer than a hold queue can wait” error. Rounding up means nothing is released early.

codec

pub const CONTENT_TYPE_JSON: &str = "application/json";
pub const DELIVERY_MODE_PERSISTENT: u8 = 2;
pub const REASON_MALFORMED: &str = "malformed envelope";

pub fn base_headers(envelope: &Envelope) -> FieldTable;
pub fn props_for(envelope: &Envelope) -> BasicProperties;
pub fn dead_letter_headers(envelope: &Envelope, reason: &str) -> FieldTable;
pub fn dead_letter_props(envelope: &Envelope, reason: &str) -> BasicProperties;
pub fn malformed_props(original_queue: &str, reason: &str) -> BasicProperties;

props_for sets content-type: application/json, delivery-mode: 2, message-id from the job id, type from the job type, priority from the envelope (always), and the x-attempt and x-deferrals headers.

expiration is never set on a message. Per-message TTL would reintroduce head-of-line blocking between delays, which is the problem per-delay hold queues exist to avoid. See Broker topology.

An over-long message-id or type is truncated at a UTF-8 character boundary rather than panicking, since AMQP short strings cap at 255 bytes.

Reading a dead-letter queue

These are the pieces a drain tool needs, and none of them requires the worker runtime:

use queuey::rabbitmq::topology::{HEADER_DEATH_REASON, HEADER_ORIGINAL_QUEUE};
use queuey::Envelope;

// `delivery` came from your own lapin consumer on `myapp.emails.dead`.
let envelope = Envelope::from_bytes(&delivery.data)?;
let reason = delivery
    .properties
    .headers()
    .as_ref()
    .and_then(|h| h.inner().get(HEADER_DEATH_REASON))
    .map(|v| v.to_string());

tracing::warn!(job_id = %envelope.job_id, attempt = envelope.attempt, ?reason, "dead letter");

if envelope.job_type == SendEmail::NAME {
    let job: SendEmail = envelope.decode()?;
    // Decide: fix and re-enqueue as a new job, or discard.
}

Check job_type before decoding: Envelope::decode does not verify it.

Also public

pub use lapin;                    // the lapin version this was built against
pub struct RabbitMqDelivery;      // the Delivery implementation

Re-exporting lapin means tooling and connection_properties tweaks cannot end up linking a different version than the backend uses.

RabbitMqError is not public. Its variants reach you boxed inside queuey::Error::Backend; see Errors.