Struct RedisStorage 

pub struct RedisStorage<Args, Conn = ConnectionManager, C = JsonCodec<Vec<u8>>> { /* private fields */ }

Represents a Backend that uses Redis for storage.

Feature Support

Features

Feature	Status	Description
TaskSink	✅	Ability to push new tasks
MakeShared	✅	Share the same connection across multiple workers
Workflow	✅	Supports workflows and orchestration
Web Interface	✅	Supports apalis-board for monitoring and managing tasks
WaitForCompletion	✅	Wait for tasks to complete without blocking
Serialization	✅	Supports multiple serialization formats such as JSON and MessagePack
RegisterWorker	✅	Allow registering a worker with the backend
ResumeAbandoned	✅	Resume abandoned tasks

Key: ✅ : Supported | ⚠️ : Not implemented | ❌ : Not Supported | ❗ Limited support

Tests:

TaskSink

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    backend.push(42).await.unwrap();

    async fn task(task: u32, worker: WorkerContext) {
        worker.stop().unwrap();
    }
    let worker = WorkerBuilder::new("task-sink-test")
        .backend(backend)
        .build(task);
    worker.run().await.unwrap();
}

Workflow

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    backend.push_start(42).await.unwrap();

    async fn task1(task: u32, worker: WorkerContext) -> u32 {
        task + 99 
    }
    async fn task2(task: u32, worker: WorkerContext) -> u32 {
        task + 1 
    }
    async fn task3(task: u32, worker: WorkerContext) {
        assert_eq!(task, 142);
        worker.stop().unwrap();
    }
    let workflow = Workflow::new("test-workflow")
       .and_then(task1)
       .and_then(task2)
       .and_then(task3);
    let worker = WorkerBuilder::new("workflow-test")
        .backend(backend)
        .build(workflow);
    worker.run().await.unwrap();
}

WebUI

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    fn assert_web_ui<B: Expose<u32>>(backend: B) {};
    assert_web_ui(backend);
}

WaitForCompletion

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    fn assert_wait_for_completion<B: WaitForCompletion<(), Args = u32>>(backend: B) {};
    assert_wait_for_completion(backend);
}

Implementations

impl<Args, Conn, C> RedisStorage<Args, Conn, C>

where Args: Unpin + Send + Sync + 'static, Conn: ConnectionLike + Send + Sync + 'static, C: Codec<Args, Compact = Vec<u8>>, C::Error: Into<BoxDynError>,

pub async fn fetch_next( worker: &WorkerContext, config: &RedisConfig, conn: &mut Conn, ) -> Result<Vec<Task<Vec<u8>, RedisContext, Ulid>>, RedisError>

Fetches the next batch of tasks for the given worker.

impl<T, Conn: Clone> RedisStorage<T, Conn, JsonCodec<Vec<u8>>>

pub fn new(conn: Conn) -> RedisStorage<T, Conn, JsonCodec<Vec<u8>>>

Start a new connection

pub fn new_with_config( conn: Conn, config: RedisConfig, ) -> RedisStorage<T, Conn, JsonCodec<Vec<u8>>>

Start a connection with a custom config

pub fn new_with_codec<K>( conn: Conn, config: RedisConfig, ) -> RedisStorage<T, Conn, K>

where K: Sync + Send + 'static,

Start a new connection providing custom config and a codec

pub fn get_connection(&self) -> &Conn

Get current connection

pub fn get_config(&self) -> &RedisConfig

Get the config used by the storage

Trait Implementations

impl<Args, Conn, C> Backend for RedisStorage<Args, Conn, C>

where Args: Unpin + Send + Sync + 'static, Conn: Clone + ConnectionLike + Send + Sync + 'static, C: Codec<Args, Compact = Vec<u8>> + Unpin + Send + 'static, C::Error: Into<BoxDynError>,

type Args = Args

The type of arguments the backend handles.

type Stream = Pin<Box<dyn Stream<Item = Result<Option<Task<Args, RedisContext, Ulid>>, RedisError>> + Send>>

A stream of tasks provided by the backend.

type IdType = Ulid

The type used to uniquely identify tasks.

type Error = RedisError

The error type returned by backend operations

type Layer = AcknowledgeLayer<RedisAck<Conn, C>>

The type representing backend middleware layer.

type Context = RedisContext

Context associated with each task.

type Beat = Pin<Box<dyn Stream<Item = Result<(), <RedisStorage<Args, Conn, C> as Backend>::Error>> + Send>>

A stream representing heartbeat signals.

fn heartbeat(&self, worker: &WorkerContext) -> Self::Beat

Returns a heartbeat stream for the given worker.

fn middleware(&self) -> Self::Layer

Returns the backend’s middleware layer.

fn poll(self, worker: &WorkerContext) -> Self::Stream

Polls the backend for tasks for the given worker.

impl<Args, Conn, C> BackendExt for RedisStorage<Args, Conn, C>

where Args: Unpin + Send + Sync + 'static, Conn: Clone + ConnectionLike + Send + Sync + 'static, C: Codec<Args, Compact = Vec<u8>> + Unpin + Send + 'static, C::Error: Into<BoxDynError>,

type Compact = Vec<u8>

The compact representation of task arguments.

type Codec = C

The codec used for serialization/deserialization of tasks.

type CompactStream = Pin<Box<dyn Stream<Item = Result<Option<Task<<RedisStorage<Args, Conn, C> as BackendExt>::Compact, RedisContext, Ulid>>, RedisError>> + Send>>

A stream of encoded tasks provided by the backend.

fn poll_compact(self, worker: &WorkerContext) -> Self::CompactStream

Polls the backend for encoded tasks for the given worker.

impl<Args, Conn: Clone, Cdc: Clone> Clone for RedisStorage<Args, Conn, Cdc>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<Args: Sync, Conn, C> ConfigExt for RedisStorage<Args, Conn, C>

where RedisStorage<Args, Conn, C>: BackendExt<Context = RedisContext, Compact = Vec<u8>, IdType = Ulid, Error = RedisError>,

fn get_queue(&self) -> Queue

Get the queue configuration

impl<Args: Debug, Conn: Debug, C: Debug> Debug for RedisStorage<Args, Conn, C>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<Args, Conn, C> FetchById<Args> for RedisStorage<Args, Conn, C>

where RedisStorage<Args, Conn, C>: BackendExt<Context = RedisContext, Compact = Vec<u8>, IdType = Ulid, Error = RedisError>, C: Codec<Args, Compact = Vec<u8>> + Send, C::Error: Error + Send + Sync + 'static, Args: 'static + Send, Conn: ConnectionLike + Send,

async fn fetch_by_id( &mut self, task_id: &TaskId<Self::IdType>, ) -> Result<Option<RedisTask<Args>>, Self::Error>

Fetch a task by its unique identifier

impl<Args, Conn, C> ListAllTasks for RedisStorage<Args, Conn, C>

where RedisStorage<Args, Conn, C>: BackendExt<Context = RedisContext, Compact = Vec<u8>, IdType = Ulid, Error = RedisError>, C: Codec<Args, Compact = Vec<u8>> + Send + Sync, C::Error: Error + Send + Sync + 'static, Args: 'static + Send + Sync, Conn: ConnectionLike + Send + Sync + Clone,

async fn list_all_tasks( &self, filter: &Filter, ) -> Result<Vec<RedisTask<Vec<u8>>>, Self::Error>

List tasks matching the given filter in all queues

impl<Args, Conn, C> ListQueues for RedisStorage<Args, Conn, C>

where RedisStorage<Args, Conn, C>: BackendExt<Context = RedisContext, Compact = Vec<u8>, IdType = Ulid, Error = RedisError>, C: Codec<Args, Compact = Vec<u8>> + Send, C::Error: Error + Send + Sync + 'static, Args: 'static + Send, Conn: ConnectionLike + Send + Clone,

fn list_queues( &self, ) -> impl Future<Output = Result<Vec<QueueInfo>, Self::Error>> + Send

List all available queues in the backend

impl<Args, Conn, C> ListTasks<Args> for RedisStorage<Args, Conn, C>

where RedisStorage<Args, Conn, C>: BackendExt<Context = RedisContext, Compact = Vec<u8>, IdType = Ulid, Error = RedisError>, C: Codec<Args, Compact = Vec<u8>> + Send + Sync, C::Error: Error + Send + Sync + 'static, Args: 'static + Send + Sync, Conn: ConnectionLike + Send + Clone + Sync,

async fn list_tasks( &self, queue: &str, filter: &Filter, ) -> Result<Vec<RedisTask<Args>>, Self::Error>

List tasks matching the given filter in the current queue

impl<Args, Conn, C> ListWorkers for RedisStorage<Args, Conn, C>

where RedisStorage<Args, Conn, C>: BackendExt<Context = RedisContext, Compact = Vec<u8>, IdType = Ulid, Error = RedisError>, C: Codec<Args, Compact = Vec<u8>> + Send, C::Error: Error + Send + Sync + 'static, Args: 'static + Send + Sync, Conn: ConnectionLike + Send + Clone,

fn list_workers( &self, queue: &str, ) -> impl Future<Output = Result<Vec<RunningWorker>, Self::Error>> + Send

List all registered workers in the current queue

fn list_all_workers( &self, ) -> impl Future<Output = Result<Vec<RunningWorker>, Self::Error>> + Send

List all registered workers in all queues

impl<Args, Conn, C> Metrics for RedisStorage<Args, Conn, C>

where RedisStorage<Args, Conn, C>: BackendExt<Context = RedisContext, Compact = Vec<u8>, IdType = Ulid, Error = RedisError>, C: Codec<Args, Compact = Vec<u8>> + Send + Sync, C::Error: Error + Send + Sync + 'static, Args: 'static + Send + Sync, Conn: ConnectionLike + Send + Clone + Sync,

fn global( &self, ) -> impl Future<Output = Result<Vec<Statistic>, Self::Error>> + Send

Collects and returns global statistics from the backend

fn fetch_by_queue( &self, queue_id: &str, ) -> impl Future<Output = Result<Vec<Statistic>, Self::Error>> + Send

Collects and returns statistics for a specific queue

impl<Args, Cdc, Conn> Sink<Task<Vec<u8>, RedisContext, Ulid>> for RedisStorage<Args, Conn, Cdc>

where Args: Unpin, Conn: ConnectionLike + Unpin + Send + Clone + 'static, Cdc: Unpin,

type Error = RedisError

The type of value produced by the sink when an error occurs.

fn poll_ready( self: Pin<&mut Self>, _cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Attempts to prepare the Sink to receive a value.

fn start_send( self: Pin<&mut Self>, item: Task<Vec<u8>, RedisContext, Ulid>, ) -> Result<(), Self::Error>

Begin the process of sending a value to the sink. Each call to this function must be preceded by a successful call to poll_ready which returned Poll::Ready(Ok(())).

fn poll_flush( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Flush any remaining output from this sink.

fn poll_close( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Flush any remaining output and close this sink, if necessary.

impl<Res, Args, Conn, Decode, Err> WaitForCompletion<Res> for RedisStorage<Args, Conn, Decode>

where Args: Unpin + Send + Sync + 'static, Conn: Clone + ConnectionLike + Send + Sync + 'static, Decode: Codec<Args, Compact = Vec<u8>, Error = Err> + Codec<Result<Res, String>, Compact = Vec<u8>, Error = Err> + Send + Sync + Unpin + 'static + Clone, Err: Into<BoxDynError> + Send + 'static, Res: Send + 'static,

type ResultStream = Pin<Box<dyn Stream<Item = Result<TaskResult<Res>, <RedisStorage<Args, Conn, Decode> as Backend>::Error>> + Send>>

The result stream type yielding task results

fn wait_for( &self, task_ids: impl IntoIterator<Item = TaskId<Self::IdType>>, ) -> Self::ResultStream

Wait for multiple tasks to complete, yielding results as they become available

async fn check_status( &self, task_ids: impl IntoIterator<Item = TaskId<Self::IdType>> + Send, ) -> Result<Vec<TaskResult<Res>>, Self::Error>

Check current status of tasks without waiting

fn wait_for_single(&self, task_id: TaskId<Self::IdType>) -> Self::ResultStream

Wait for a single task to complete, yielding its result