Queue

use broccoli_queue::queue::BroccoliQueue;

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

use broccoli_queue::queue::BroccoliQueue;

let job = JobPayload { id: "1".into(), task: "process".into() };

// Basic publish
queue.publish("jobs", None, &job, None).await?;

// With disambiguator for fairness
queue.publish("jobs", Some("tenant-123".into()), &job, None).await?;

let jobs = vec![
    JobPayload { id: "1".into(), task: "task-a".into() },
    JobPayload { id: "2".into(), task: "task-b".into() },
];

queue.publish_batch("jobs", None, jobs, None).await?;

let message: BrokerMessage<JobPayload> = queue.consume("jobs", None).await?;
println!("Received: {:?}", message.payload);

// Process the message...

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

if let Some(message) = queue.try_consume::<JobPayload>("jobs", None).await? {
    println!("Got message: {:?}", message.payload);
    queue.acknowledge("jobs", message).await?;
} else {
    println!("No messages available");
}

use time::Duration;

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

for message in messages {
    // Process and acknowledge each message
    queue.acknowledge("jobs", message).await?;
}

queue.process_messages(
    "jobs",
    Some(4),     // 4 concurrent workers
    None,        // default consume options
    |message: BrokerMessage<JobPayload>| async move {
        println!("Processing: {:?}", message.payload);
        // Return Ok(()) on success, Err on failure
        Ok(())
    },
).await?;

queue.process_messages_with_handlers(
    "jobs",
    Some(4),
    None,
    // Message handler
    |msg| async move {
        process_job(msg.payload).await
    },
    // Success handler
    |msg, result| async move {
        println!("Job {} completed", msg.task_id);
        Ok(())
    },
    // Error handler
    |msg, error| async move {
        eprintln!("Job {} failed: {:?}", msg.task_id, error);
        Ok(())
    },
).await?;

let sizes = queue.size("jobs").await?;
for (queue_name, size) in sizes {
    println!("{}: {} messages", queue_name, size);
}