carotte - v3.0.0

carotte/queue

Types

DeclaredQueue

</> 
pub type DeclaredQueue {
  DeclaredQueue(
    name: String,
    message_count: Int,
    consumer_count: Int,
  )
}

Constructors 

  • DeclaredQueue(
  name: String,
  message_count: Int,
  consumer_count: Int,
)

Deliver

</> 
pub type Deliver {
  Deliver(
    consumer_tag: String,
    delivery_tag: Int,
    redelivered: Bool,
    exchange: String,
    routing_key: String,
  )
}

Constructors 

  • Deliver(
  consumer_tag: String,
  delivery_tag: Int,
  redelivered: Bool,
  exchange: String,
  routing_key: String,
)

Payload

</> 
pub type Payload {
  Payload(
    payload: String,
    properties: List(publisher.PublishOption),
  )
}

Constructors

Queue

opaque </> 
pub opaque type Queue

QueueOption

</> 
pub type QueueOption {
  AutoAck(Bool)
}

Constructors 

  • AutoAck(Bool)

Values

ack

</> 
pub fn ack(
  channel: channel.Channel,
  delivery_tag: Int,
  multiple: Bool,
) -> Result(Nil, carotte.CarotteError)

Acknowledge a message delivery. Used when manual acknowledgment is enabled (AutoAck(False)).

Parameters

  • channel: The channel to acknowledge on
  • delivery_tag: The delivery tag from the message metadata
  • multiple: If True, acknowledges all messages up to and including this delivery tag

Example

queue.subscribe_with_options(
  channel: ch,
  queue: "my_queue",
  options: [queue.AutoAck(False)],
  callback: fn(msg, meta) {
    // Process message
    let _ = queue.ack(ch, meta.delivery_tag, False)
  },
)

ack_single

</> 
pub fn ack_single(
  channel: channel.Channel,
  delivery_tag: Int,
) -> Result(Nil, carotte.CarotteError)

Acknowledge a message delivery (acknowledges only this message). Convenience function for ack with multiple=False.

as_durable

</> 
pub fn as_durable(queue: Queue) -> Queue

If set, the queue will survive a broker restart

as_exclusive

</> 
pub fn as_exclusive(queue: Queue) -> Queue

If set, only one subscriber can consume from the Queue

as_passive

</> 
pub fn as_passive(queue: Queue) -> Queue

If set, the queue must already exist on the broker

bind

</> 
pub fn bind(
  channel channel: channel.Channel,
  queue queue: String,
  exchange exchange: String,
  routing_key routing_key: String,
) -> Result(Nil, carotte.CarotteError)

Bind a queue to an exchange The routing_key is used to filter messages from the exchange

bind_async

</> 
pub fn bind_async(
  channel channel: channel.Channel,
  queue queue: String,
  exchange exchange: String,
  routing_key routing_key: String,
) -> Result(Nil, carotte.CarotteError)

Bind a queue to an exchange asynchronously. Same semantics as bind

declare

</> 
pub fn declare(
  queue: Queue,
  channel: channel.Channel,
) -> Result(DeclaredQueue, carotte.CarotteError)

Declare a queue on the broker

declare_async

</> 
pub fn declare_async(
  queue: Queue,
  channel: channel.Channel,
) -> Result(Nil, carotte.CarotteError)

Declare a queue on the broker asynchronously

delete

</> 
pub fn delete(
  channel channel: channel.Channel,
  queue queue: String,
  if_unused if_unused: Bool,
  if_empty if_empty: Bool,
) -> Result(Int, carotte.CarotteError)

Delete a queue from the broker If if_unused is set, the queue will only be deleted if it has no subscribers If if_empty is set, the queue will only be deleted if it has no messages

delete_async

</> 
pub fn delete_async(
  channel channel: channel.Channel,
  queue queue: String,
  if_unused if_unused: Bool,
  if_empty if_empty: Bool,
) -> Result(Nil, carotte.CarotteError)

Delete a queue from the broker asynchronously. Same semantics as delete

new

</> 
pub fn new(name: String) -> Queue

Create a new queue with the given name

purge

</> 
pub fn purge(
  channel channel: channel.Channel,
  queue queue: String,
) -> Result(Int, carotte.CarotteError)

Purge a queue of all messages

purge_async

</> 
pub fn purge_async(
  channel channel: channel.Channel,
  queue queue: String,
) -> Result(Nil, carotte.CarotteError)

Purge a queue of all messages asynchronously

status

</> 
pub fn status(
  channel channel: channel.Channel,
  queue queue: String,
) -> Result(DeclaredQueue, carotte.CarotteError)

Get the status of a queue

subscribe

</> 
pub fn subscribe(
  channel channel: channel.Channel,
  queue queue: String,
  callback fun: fn(Payload, Deliver) -> Nil,
) -> Result(String, carotte.CarotteError)

Subscribe to a queue The callback function will be called with each message received, receiving the message Payload and a Deliver struct Returns the consumer tag which can be used to unsubscribe

subscribe_with_options

</> 
pub fn subscribe_with_options(
  channel channel: channel.Channel,
  queue queue: String,
  options options: List(QueueOption),
  callback fun: fn(Payload, Deliver) -> Nil,
) -> Result(String, carotte.CarotteError)

unbind

</> 
pub fn unbind(
  channel channel: channel.Channel,
  queue queue: String,
  exchange exchange: String,
  routing_key routing_key: String,
) -> Result(Nil, carotte.CarotteError)

Unbind a queue from an exchange The routing_key is used to filter messages from the exchange

unsubscribe

</> 
pub fn unsubscribe(
  channel channel: channel.Channel,
  consumer_tag consumer_tag: String,
) -> Result(Nil, carotte.CarotteError)

Unsubscribe a consumer from a queue

unsubscribe_async

</> 
pub fn unsubscribe_async(
  channel channel: channel.Channel,
  consumer_tag consumer_tag: String,
) -> Result(Nil, carotte.CarotteError)

Unsubscribe a consumer from a queue asynchronously

with_auto_delete

</> 
pub fn with_auto_delete(queue: Queue) -> Queue

If set, the queue will be deleted when the last subscriber disconnect

✨ Search Document