BroccoliQueue

The main entry point for interacting with Broccoli message queues.

use broccoli_queue::queue::BroccoliQueue;

Creating a queue

`builder`

Creates a new BroccoliQueueBuilder for configuring the queue.

pub fn builder(broker_url: impl Into<String>) -> BroccoliQueueBuilder

Parameters:

broker_url - Connection URL for the message broker

Example:

let queue = BroccoliQueue::builder("redis://localhost:6379")
    .pool_connections(5)
    .build()
    .await?;

`builder_with` (SurrealDB only)

Creates a builder with an existing SurrealDB connection.

#[cfg(feature = "surrealdb")]
pub fn builder_with(db: Surreal<Any>) -> BroccoliQueueBuilder

Publishing

`publish`

Publishes a single message to a queue.

pub async fn publish<T>(
    &self,
    topic: &str,
    disambiguator: Option<String>,
    message: &T,
    options: Option<PublishOptions>,
) -> Result<BrokerMessage<T>, BroccoliError>
where
    T: Clone + Serialize + DeserializeOwned

Parameters:

topic - Queue name
disambiguator - Optional identifier for fairness queues
message - The message payload
options - Optional publish configuration

Returns: The wrapped BrokerMessage with generated task_id Example:

let message = queue.publish("jobs", None, &job, None).await?;
println!("Published: {}", message.task_id);

`publish_batch`

Publishes multiple messages to a queue.

pub async fn publish_batch<T>(
    &self,
    topic: &str,
    disambiguator: Option<String>,
    messages: impl IntoIterator<Item = T>,
    options: Option<PublishOptions>,
) -> Result<Vec<BrokerMessage<T>>, BroccoliError>
where
    T: Clone + Serialize + DeserializeOwned

Example:

let jobs = vec![job1, job2, job3];
let messages = queue.publish_batch("jobs", None, jobs, None).await?;

Consuming

`consume`

Consumes a message, blocking until one is available.

pub async fn consume<T>(
    &self,
    topic: &str,
    options: Option<ConsumeOptions>,
) -> Result<BrokerMessage<T>, BroccoliError>
where
    T: Clone + Serialize + DeserializeOwned

Example:

let message: BrokerMessage<JobPayload> = queue.consume("jobs", None).await?;

`try_consume`

Attempts to consume a message without blocking.

pub async fn try_consume<T>(
    &self,
    topic: &str,
    options: Option<ConsumeOptions>,
) -> Result<Option<BrokerMessage<T>>, BroccoliError>
where
    T: Clone + Serialize + DeserializeOwned

Returns: Some(message) if available, None otherwise Example:

if let Some(message) = queue.try_consume::<JobPayload>("jobs", None).await? {
    // Process message
}

`consume_batch`

Consumes multiple messages with a timeout.

pub async fn consume_batch<T>(
    &self,
    topic: &str,
    batch_size: usize,
    timeout: Duration,
    options: Option<ConsumeOptions>,
) -> Result<Vec<BrokerMessage<T>>, BroccoliError>
where
    T: Clone + Serialize + DeserializeOwned

Example:

let messages = queue.consume_batch::<JobPayload>(
    "jobs",
    10,
    Duration::seconds(5),
    None
).await?;

`try_consume_batch`

Attempts to consume multiple messages without blocking.

pub async fn try_consume_batch<T>(
    &self,
    topic: &str,
    batch_size: usize,
    options: Option<ConsumeOptions>,
) -> Result<Vec<BrokerMessage<T>>, BroccoliError>

Message handling

`acknowledge`

Acknowledges successful processing of a message.

pub async fn acknowledge<T>(
    &self,
    topic: &str,
    message: BrokerMessage<T>,
) -> Result<(), BroccoliError>
where
    T: Clone + Serialize + DeserializeOwned

Example:

queue.acknowledge("jobs", message).await?;

`reject`

Rejects a message, triggering retry or failure handling.

pub async fn reject<T>(
    &self,
    topic: &str,
    message: BrokerMessage<T>,
) -> Result<(), BroccoliError>
where
    T: Clone + Serialize + DeserializeOwned

`cancel`

Cancels a message by ID.

pub async fn cancel(
    &self,
    topic: &str,
    message_id: String,
) -> Result<(), BroccoliError>

Processing

`process_messages`

Processes messages in a loop with a handler function.

pub async fn process_messages<T, F, Fut>(
    &self,
    topic: &str,
    concurrency: Option<usize>,
    consume_options: Option<ConsumeOptions>,
    handler: F,
) -> Result<(), BroccoliError>
where
    T: DeserializeOwned + Send + Clone + Serialize + 'static,
    F: Fn(BrokerMessage<T>) -> Fut + Send + Sync + Clone + 'static,
    Fut: Future<Output = Result<(), BroccoliError>> + Send + 'static

Parameters:

topic - Queue name
concurrency - Number of concurrent workers (None for single-threaded)
consume_options - Optional consume configuration
handler - Async function to process each message

Example:

queue.process_messages("jobs", Some(4), None, |msg: BrokerMessage<Job>| async move {
    println!("Processing: {:?}", msg.payload);
    Ok(())
}).await?;

`process_messages_with_handlers`

Processes messages with separate success and error handlers.

pub async fn process_messages_with_handlers<T, F, MessageFut, SuccessFut, ErrorFut, S, E, R>(
    &self,
    topic: &str,
    concurrency: Option<usize>,
    consume_options: Option<ConsumeOptions>,
    message_handler: F,
    on_success: S,
    on_error: E,
) -> Result<(), BroccoliError>

Example:

queue.process_messages_with_handlers(
    "jobs",
    Some(4),
    None,
    |msg| async move { process(msg.payload).await },
    |msg, result| async move { log_success(&msg).await; Ok(()) },
    |msg, error| async move { log_error(&msg, &error).await; Ok(()) },
).await?;

Queue information

`size`

Returns the size of queue(s).

pub async fn size(
    &self,
    queue_name: &str,
) -> Result<HashMap<String, u64>, BroccoliError>

Returns: Map of queue names to message counts

`queue_status` (management feature)

Returns detailed queue status.

#[cfg(feature = "management")]
pub async fn queue_status(
    &self,
    queue_name: String,
    disambiguator: Option<String>,
) -> Result<QueueStatus, BroccoliError>

BroccoliQueueBuilder

Builder for configuring BroccoliQueue.

`pool_connections`

Sets the connection pool size.

pub const fn pool_connections(mut self, connections: u8) -> Self

Default: 10

`failed_message_retry_strategy`

Configures retry behavior for failed messages.

pub const fn failed_message_retry_strategy(mut self, strategy: RetryStrategy) -> Self

`enable_scheduling`

Enables message scheduling.

pub const fn enable_scheduling(mut self, enable: bool) -> Self

Default: false

`build`

Builds the queue and connects to the broker.

pub async fn build(self) -> Result<BroccoliQueue, BroccoliError>

​BroccoliQueue

​Creating a queue

​builder

​builder_with (SurrealDB only)

​Publishing

​publish

​publish_batch

​Consuming

​consume

​try_consume

​consume_batch

​try_consume_batch

​Message handling

​acknowledge

​reject

​cancel

​Processing

​process_messages

​process_messages_with_handlers

​Queue information

​size

​queue_status (management feature)

​BroccoliQueueBuilder

​pool_connections

​failed_message_retry_strategy

​enable_scheduling

​build

BroccoliQueue

Creating a queue

`builder`

`builder_with` (SurrealDB only)

Publishing

`publish`

`publish_batch`

Consuming

`consume`

`try_consume`

`consume_batch`

`try_consume_batch`

Message handling

`acknowledge`

`reject`

`cancel`

Processing

`process_messages`

`process_messages_with_handlers`

Queue information

`size`

`queue_status` (management feature)

BroccoliQueueBuilder

`pool_connections`

`failed_message_retry_strategy`

`enable_scheduling`

`build`