Struct BroccoliQueue

pub struct BroccoliQueue { /* private fields */ }

Main queue interface for interacting with the message broker.

BroccoliQueue provides methods for publishing and consuming messages, as well as processing messages with custom handlers.

Implementations

impl BroccoliQueue

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

Creates a new BroccoliQueueBuilder with the specified broker URL.

Arguments

• broker_url - The URL of the broker.

Returns

A new BroccoliQueueBuilder instance.

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

Publishes a message to the specified topic.

Arguments

• topic - The name of the topic.
• message - The message to be published.

Returns

A Result indicating success or failure.

Errors

If the message fails to publish, a BroccoliError will be returned.

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

Publishes a batch of messages to the specified topic.

Arguments

• topic - The name of the topic.
• messages - An iterator over the messages to be published.

Returns

A Result indicating success or failure.

Errors

If the messages fail to publish, a BroccoliError will be returned.

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

Consumes a message from the specified topic. This method will block until a message is available. This will not acknowledge the message, use acknowledge to remove the message from the processing queue, or reject to move the message to the failed queue.

Arguments

• topic - The name of the topic.

Returns

A Result containing the consumed message, or a BroccoliError on failure.

Errors

If the message fails to consume, a BroccoliError will be returned.

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

Consumes a batch of messages from the specified topic. This method will block until the specified number of messages are consumed. This will not acknowledge the message, use acknowledge to remove the message from the processing queue, or reject to move the message to the failed queue.

Arguments

• topic - The name of the topic.
• batch_size - The number of messages to consume.
• timeout - The timeout duration for consuming messages.

Returns

A Result containing a vector of consumed messages, or a BroccoliError on failure.

Errors

If the messages fail to consume, a BroccoliError will be returned.

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

Attempts to consume a message from the specified topic. This method will not block, returning immediately if no message is available. This will not acknowledge the message, use acknowledge to remove the message from the processing queue, or reject to move the message to the failed queue.

Arguments

• topic - The name of the topic.

Returns

A Result containing an Option with the consumed message if available, or a BroccoliError on failure.

Errors

If the message fails to consume, a BroccoliError will be returned.

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

Acknowledges the processing of a message, removing it from the processing queue.

Arguments

• topic - The name of the topic.
• message - The message to be acknowledged.

Returns

A Result indicating success or failure.

Errors

If the message fails to acknowledge, a BroccoliError will be returned.

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

Rejects the processing of a message, moving it to the failed queue.

Arguments

• topic - The name of the topic.
• message - The message to be rejected.

Returns

A Result indicating success or failure.

Errors

If the message fails to reject, a BroccoliError will be returned.

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

Cancels the processing of a message, deleting it from the queue.

Arguments

• topic - The name of the topic.
• message_id - The ID of the message to cancel.

Returns

A Result indicating success or failure.

Errors

If the message fails to cancel, a BroccoliError will be returned.

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,

Processes messages from the specified topic with the provided handler function.

Example

use broccoli_queue::queue::BroccoliQueue;
use broccoli_queue::brokers::broker::BrokerMessage;

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct JobPayload {
   id: String,
   task_name: String,
   created_at: chrono::DateTime<chrono::Utc>,
}

#[tokio::main]
async fn main() {
   let queue = BroccoliQueue::builder("redis://localhost:6379")
      .failed_message_retry_strategy(Default::default())
      .pool_connections(5)
      .build()
      .await
      .unwrap();

  queue.process_messages("jobs", None, None, |message: BrokerMessage<JobPayload>| async move {
        println!("Received message: {:?}", message);
        Ok(())
    }).await.unwrap();

}

Arguments

• topic - The name of the topic.
• concurrency - The number of concurrent message handlers.
• handler - The handler function to process messages. This function should return a BroccoliError on failure.

Returns

A Result indicating success or failure.

Errors

If the message fails to process, a BroccoliError will be returned.

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>

where T: DeserializeOwned + Send + Clone + Serialize + 'static, F: Fn(BrokerMessage<T>) -> MessageFut + Send + Sync + Clone + 'static, MessageFut: Future<Output = Result<R, BroccoliError>> + Send + 'static, R: Send + Clone + 'static, S: Fn(BrokerMessage<T>, R) -> SuccessFut + Send + Sync + Clone + 'static, SuccessFut: Future<Output = Result<(), BroccoliError>> + Send + 'static, E: Fn(BrokerMessage<T>, BroccoliError) -> ErrorFut + Send + Sync + Clone + 'static, ErrorFut: Future<Output = Result<(), BroccoliError>> + Send + 'static,

Processes messages from the specified topic with the provided handler functions for message processing, success, and error handling.

Example

use broccoli_queue::queue::BroccoliQueue;
use broccoli_queue::brokers::broker::BrokerMessage;
use broccoli_queue::error::BroccoliError;

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct JobPayload {
    id: String,
    task_name: String,
    created_at: chrono::DateTime<chrono::Utc>,
}

#[tokio::main]
async fn main() {
    let queue = BroccoliQueue::builder("redis://localhost:6379")
        .failed_message_retry_strategy(Default::default())
        .pool_connections(5)
        .build()
        .await
        .unwrap();

    // Define handlers
    async fn process_message(message: BrokerMessage<JobPayload>) -> Result<(), BroccoliError> {
        println!("Processing message: {:?}", message);
        Ok(())
    }

    async fn on_success(message: BrokerMessage<JobPayload>, _result: ()) -> Result<(), BroccoliError> {
        println!("Successfully processed message: {}", message.task_id);
        Ok(())
    }

    async fn on_error(message: BrokerMessage<JobPayload>, error: BroccoliError) -> Result<(), BroccoliError> {
        println!("Failed to process message {}: {:?}", message.task_id, error);
        Ok(())
    }

    // Process messages with 3 concurrent workers
    queue.process_messages_with_handlers(
        "jobs",
        Some(3),
        None,    
        process_message,
        on_success,
        on_error
    ).await.unwrap();
}

Arguments

• topic - The name of the topic.
• concurrency - The number of concurrent message handlers.
• message_handler - The handler function to process messages. This function should return a BroccoliError on failure.
• on_success - The handler function to call on successful message processing. This function should return a BroccoliError on failure.
• on_error - The handler function to call on message processing failure. This function should return a BroccoliError on failure.

Returns

A Result indicating success or failure.

Errors

If the message fails to process, a BroccoliError will be returned.

pub async fn queue_status( &self, queue_name: String, disambiguator: Option<String>, ) -> Result<QueueStatus, BroccoliError>

Retrieves the status of the specified queue.

Arguments

• queue_name - The name of the queue.

Returns

A Result containing the status of the queue, or a BroccoliError on failure.

Errors

If the queue status fails to be retrieved, a BroccoliError will be returned.

Trait Implementations

impl Clone for BroccoliQueue

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.