Struct RedisBroker 

pub struct RedisBroker { /* private fields */ }

Redis-based broker implementation

Implementations

impl RedisBroker

pub fn new( redis_url: &str, queue_name: &str, ) -> Result<RedisBroker, CelersError>

Create a new Redis broker with FIFO mode

pub fn with_mode( redis_url: &str, queue_name: &str, mode: QueueMode, ) -> Result<RedisBroker, CelersError>

Create a new Redis broker with specified queue mode

pub fn from_config( config: &RedisConfig, queue_name: &str, mode: QueueMode, ) -> Result<RedisBroker, CelersError>

Create a new Redis broker from a RedisConfig

pub async fn from_sentinel( sentinel: &SentinelClient, queue_name: &str, mode: QueueMode, ) -> Result<RedisBroker, CelersError>

Create a new Redis broker from a SentinelClient (for high availability)

pub fn with_visibility_timeout(self, timeout_secs: u64) -> RedisBroker

Set visibility timeout (default: 300 seconds)

pub fn mode(&self) -> QueueMode

Get the queue mode

pub async fn dlq_size(&self) -> Result<usize, CelersError>

Get the number of tasks in the Dead Letter Queue

pub async fn inspect_dlq( &self, limit: isize, ) -> Result<Vec<SerializedTask>, CelersError>

Inspect tasks in the Dead Letter Queue

pub async fn replay_from_dlq(&self, task_id: &Uuid) -> Result<bool, CelersError>

Replay a task from the Dead Letter Queue back to the main queue

pub async fn clear_dlq(&self) -> Result<usize, CelersError>

Clear all tasks from the Dead Letter Queue

pub fn cancel_channel(&self) -> &str

Get the cancellation channel name (for workers to subscribe)

pub async fn create_pubsub(&self) -> Result<PubSub, CelersError>

Create a PubSub connection for listening to cancellation messages

pub fn health_checker(&self) -> HealthChecker

Create a health checker for monitoring Redis status

pub fn queue_controller(&self) -> QueueController

Create a queue controller for pause/resume operations

pub fn deduplicator(&self) -> Deduplicator

Create a task deduplicator

pub fn script_manager(&self) -> ScriptManager

Create a script manager for Lua script optimization

pub fn partition_manager( &self, num_partitions: usize, strategy: PartitionStrategy, ) -> PartitionManager

Create a partition manager for distributed queues

pub fn batch_operations(&self) -> BatchOperations

Create a batch operations handler for advanced batch processing

pub fn priority_manager(&self) -> PriorityManager

Create a priority manager for dynamic priority adjustments

pub fn task_query(&self) -> TaskQuery

Create a task query interface for inspecting tasks

pub fn backup_manager(&self) -> BackupManager

Create a backup manager for queue backup and restore

pub async fn set_queue_ttl(&self, ttl_secs: u64) -> Result<(), CelersError>

Set TTL (time-to-live) for tasks in all queues

This helps prevent unbounded queue growth by automatically expiring old tasks. TTL is set in seconds.

pub async fn cleanup_dlq( &self, _max_age_secs: u64, ) -> Result<usize, CelersError>

Clean up old tasks from DLQ (older than specified age in seconds)

pub async fn get_queue_stats(&self) -> Result<QueueStats, CelersError>

Get queue statistics (pending, processing, DLQ, delayed counts)

pub async fn check_health(&self) -> RedisHealthStatus

Check Redis health status

pub async fn ping(&self) -> Result<u64, CelersError>

Ping Redis and return latency in milliseconds

pub fn visibility_timeout(&self) -> u64

Get the visibility timeout in seconds

pub fn client(&self) -> &Client

Get the Redis client (for advanced operations)

pub fn delayed_queue_name(&self) -> &str

Get the delayed queue name

pub fn processing_queue_name(&self) -> &str

Get the processing queue name

pub fn dlq_name(&self) -> &str

Get the DLQ name

pub fn queue_name(&self) -> &str

Get the main queue name

pub fn queue_names(&self) -> Vec<String>

Get all queue names managed by this broker

pub async fn queue_exists(&self, queue_name: &str) -> Result<bool, CelersError>

Check if a specific queue exists and has tasks

pub async fn get_queue_size_by_name( &self, queue_name: &str, ) -> Result<usize, CelersError>

Get the size of a specific queue by name

pub async fn delete_queue(&self, queue_name: &str) -> Result<(), CelersError>

Delete a specific queue (use with caution!)

pub async fn purge_all_queues(&self) -> Result<usize, CelersError>

Purge all queues (main, processing, DLQ, delayed)

pub async fn get_all_queue_sizes( &self, ) -> Result<HashMap<String, usize>, CelersError>

Get a summary of all queue sizes

pub async fn recover_processing_tasks(&self) -> Result<usize, CelersError>

Move all tasks from processing queue back to main queue

Useful for recovering tasks that were being processed when a worker crashed. Returns the number of tasks moved.

pub async fn total_task_count(&self) -> Result<usize, CelersError>

Get the total number of tasks across all queues (main, processing, DLQ, delayed)

pub async fn is_idle(&self) -> Result<bool, CelersError>

Check if the broker is idle (all queues empty)

pub async fn estimate_memory_usage(&self) -> Result<usize, CelersError>

Estimate memory usage of tasks in all queues (in bytes)

This is an approximation based on serialized task sizes.

pub async fn bulk_replay_from_dlq( &self, max_count: Option<usize>, ) -> Result<usize, CelersError>

Move tasks from DLQ back to main queue in bulk

Returns the number of tasks moved.

pub async fn peek_next(&self) -> Result<Option<SerializedTask>, CelersError>

Peek at the next task without dequeueing it

Useful for inspecting what will be processed next without removing it from the queue.

pub async fn queue_depth_percentage( &self, max_recommended_size: Option<usize>, ) -> Result<f64, CelersError>

Get queue depth percentage (current size / max recommended size)

Returns a value between 0.0 and 1.0+ where > 0.8 suggests the queue is getting full. max_recommended_size defaults to 10000 if not specified.

pub async fn is_near_capacity( &self, max_recommended_size: Option<usize>, ) -> Result<bool, CelersError>

Check if the queue is approaching capacity (> 80% of recommended max)

Trait Implementations

impl Broker for RedisBroker

fn enqueue<'life0, 'async_trait>( &'life0 self, task: SerializedTask, ) -> Pin<Box<dyn Future<Output = Result<Uuid, CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, RedisBroker: 'async_trait,

Enqueue a task to the broker

fn dequeue<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<BrokerMessage>, CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, RedisBroker: 'async_trait,

Dequeue a task from the broker (blocking/waiting)

fn ack<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, task_id: &'life1 Uuid, receipt_handle: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<(), CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, RedisBroker: 'async_trait,

Acknowledge successful processing of a task

fn reject<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, task_id: &'life1 Uuid, receipt_handle: Option<&'life2 str>, requeue: bool, ) -> Pin<Box<dyn Future<Output = Result<(), CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, RedisBroker: 'async_trait,

Reject a task and potentially requeue it

fn queue_size<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<usize, CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, RedisBroker: 'async_trait,

Get the current queue size

fn cancel<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 Uuid, ) -> Pin<Box<dyn Future<Output = Result<bool, CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, RedisBroker: 'async_trait,

Cancel a pending task

fn enqueue_batch<'life0, 'async_trait>( &'life0 self, tasks: Vec<SerializedTask>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Uuid>, CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, RedisBroker: 'async_trait,

Enqueue multiple tasks in a single operation (batch)

fn dequeue_batch<'life0, 'async_trait>( &'life0 self, count: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<BrokerMessage>, CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, RedisBroker: 'async_trait,

Dequeue multiple tasks in a single operation (batch)

fn ack_batch<'life0, 'life1, 'async_trait>( &'life0 self, tasks: &'life1 [(Uuid, Option<String>)], ) -> Pin<Box<dyn Future<Output = Result<(), CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, RedisBroker: 'async_trait,

Acknowledge multiple tasks in a single operation (batch)

fn enqueue_at<'life0, 'async_trait>( &'life0 self, task: SerializedTask, execute_at: i64, ) -> Pin<Box<dyn Future<Output = Result<Uuid, CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, RedisBroker: 'async_trait,

Schedule a task for execution at a specific Unix timestamp (seconds)

fn enqueue_after<'life0, 'async_trait>( &'life0 self, task: SerializedTask, delay_secs: u64, ) -> Pin<Box<dyn Future<Output = Result<Uuid, CelersError>> + Send + 'async_trait>>

where 'life0: 'async_trait, RedisBroker: 'async_trait,

Schedule a task for execution after a delay (seconds)