> ## Documentation Index
> Fetch the complete documentation index at: https://densumesh-broccoli-27.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# BroccoliQueue

> API reference for the main queue interface

## BroccoliQueue

The main entry point for interacting with Broccoli message queues.

```rust theme={null}
use broccoli_queue::queue::BroccoliQueue;
```

## Creating a queue

### `builder`

Creates a new `BroccoliQueueBuilder` for configuring the queue.

```rust theme={null}
pub fn builder(broker_url: impl Into<String>) -> BroccoliQueueBuilder
```

**Parameters:**

* `broker_url` - Connection URL for the message broker

**Example:**

```rust theme={null}
let queue = BroccoliQueue::builder("redis://localhost:6379")
    .pool_connections(5)
    .build()
    .await?;
```

### `builder_with` (SurrealDB only)

Creates a builder with an existing SurrealDB connection.

```rust theme={null}
#[cfg(feature = "surrealdb")]
pub fn builder_with(db: Surreal<Any>) -> BroccoliQueueBuilder
```

## Publishing

### `publish`

Publishes a single message to a queue.

```rust theme={null}
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:**

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

### `publish_batch`

Publishes multiple messages to a queue.

```rust theme={null}
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:**

```rust theme={null}
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.

```rust theme={null}
pub async fn consume<T>(
    &self,
    topic: &str,
    options: Option<ConsumeOptions>,
) -> Result<BrokerMessage<T>, BroccoliError>
where
    T: Clone + Serialize + DeserializeOwned
```

**Example:**

```rust theme={null}
let message: BrokerMessage<JobPayload> = queue.consume("jobs", None).await?;
```

### `try_consume`

Attempts to consume a message without blocking.

```rust theme={null}
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:**

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

### `consume_batch`

Consumes multiple messages with a timeout.

```rust theme={null}
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:**

```rust theme={null}
let messages = queue.consume_batch::<JobPayload>(
    "jobs",
    10,
    Duration::seconds(5),
    None
).await?;
```

### `try_consume_batch`

Attempts to consume multiple messages without blocking.

```rust theme={null}
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.

```rust theme={null}
pub async fn acknowledge<T>(
    &self,
    topic: &str,
    message: BrokerMessage<T>,
) -> Result<(), BroccoliError>
where
    T: Clone + Serialize + DeserializeOwned
```

**Example:**

```rust theme={null}
queue.acknowledge("jobs", message).await?;
```

### `reject`

Rejects a message, triggering retry or failure handling.

```rust theme={null}
pub async fn reject<T>(
    &self,
    topic: &str,
    message: BrokerMessage<T>,
) -> Result<(), BroccoliError>
where
    T: Clone + Serialize + DeserializeOwned
```

### `cancel`

Cancels a message by ID.

```rust theme={null}
pub async fn cancel(
    &self,
    topic: &str,
    message_id: String,
) -> Result<(), BroccoliError>
```

## Processing

### `process_messages`

Processes messages in a loop with a handler function.

```rust theme={null}
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:**

```rust theme={null}
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.

```rust theme={null}
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:**

```rust theme={null}
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).

```rust theme={null}
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.

```rust theme={null}
#[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.

```rust theme={null}
pub const fn pool_connections(mut self, connections: u8) -> Self
```

**Default:** 10

### `failed_message_retry_strategy`

Configures retry behavior for failed messages.

```rust theme={null}
pub const fn failed_message_retry_strategy(mut self, strategy: RetryStrategy) -> Self
```

### `enable_scheduling`

Enables message scheduling.

```rust theme={null}
pub const fn enable_scheduling(mut self, enable: bool) -> Self
```

**Default:** false

### `build`

Builds the queue and connects to the broker.

```rust theme={null}
pub async fn build(self) -> Result<BroccoliQueue, BroccoliError>
```