Struct RedisStorage

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

Represents a Storage that uses Redis for storage.

Implementations

impl<T: Serialize + DeserializeOwned, Conn> 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: Config, ) -> RedisStorage<T, Conn, JsonCodec<Vec<u8>>>

Start a connection with a custom config

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

where K: Codec + 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) -> &Config

Get the config used by the storage

impl<T, Conn, C> RedisStorage<T, Conn, C>

pub fn get_codec(&self) -> &PhantomData<C>

Get the underlying codec details

impl<T, Conn, C> RedisStorage<T, Conn, C>

where Conn: ConnectionLike + Send + Sync + 'static, C: Codec<Compact = Vec<u8>> + Send + 'static,

pub async fn retry( &mut self, worker_id: &WorkerId, task_id: &TaskId, ) -> Result<i32, RedisError>

where T: Send + DeserializeOwned + Serialize + Unpin + Sync + 'static,

Attempt to retry a job

pub async fn kill( &mut self, worker_id: &WorkerId, task_id: &TaskId, error: &BoxDynError, ) -> Result<(), RedisError>

Attempt to kill a job

pub async fn enqueue_scheduled( &mut self, count: usize, ) -> Result<usize, RedisError>

Required to add scheduled jobs to the active set

pub async fn reenqueue_active( &mut self, job_ids: Vec<&TaskId>, ) -> Result<(), RedisError>

Re-enqueue some jobs that might be abandoned.

pub async fn reenqueue_orphaned( &mut self, count: i32, dead_since: DateTime<Utc>, ) -> Result<usize, RedisError>

Re-enqueue some jobs that might be orphaned after a number of seconds

Trait Implementations

impl<T, Conn, C, Res> Ack<T, Res, C> for RedisStorage<T, Conn, C>

where T: Sync + Send + Serialize + DeserializeOwned + Unpin + 'static, Conn: ConnectionLike + Send + Sync + 'static, C: Codec<Compact = Vec<u8>> + Send + 'static, Res: Serialize + Sync + Send + 'static,

type Context = RedisContext

The data to fetch from context to allow acknowledgement

type AckError = RedisError

The error returned by the ack

async fn ack( &mut self, ctx: &Self::Context, res: &Response<Res>, ) -> Result<(), RedisError>

Acknowledges successful processing of the given request

impl<T, Conn, C> Backend<Request<T, RedisContext>> for RedisStorage<T, Conn, C>

where T: Serialize + DeserializeOwned + Sync + Send + Unpin + 'static, Conn: ConnectionLike + Send + Sync + 'static, C: Codec<Compact = Vec<u8>> + Send + 'static,

type Stream = BackendStream<Pin<Box<dyn Stream<Item = Result<Option<Request<T, RedisContext>>, Error>> + Send>>>

The stream to be produced by the backend

type Layer = AckLayer<Sender<(RedisContext, Response<Vec<u8>>)>, T, RedisContext, C>

Returns the final decoration of layers

type Codec = C

Specifies the codec type used by the backend

fn poll(self, worker: &Worker<Context>) -> Poller<Self::Stream, Self::Layer>

Returns a poller that is ready for streaming

impl<T> BackendExpose<T> for RedisStorage<T>

where T: 'static + Serialize + DeserializeOwned + Send + Unpin + Sync,

type Request = Request<T, RedisContext>

The request type being handled by the backend

type Error = RedisError

The error returned during reading jobs and stats

async fn stats(&self) -> Result<Stat, RedisError>

Returns the counts of jobs in different states

async fn list_jobs( &self, status: &State, page: i32, ) -> Result<Vec<Self::Request>, RedisError>

Fetch jobs persisted in a backend

async fn list_workers(&self) -> Result<Vec<Worker<WorkerState>>, RedisError>

List all Workers that are working on a backend

impl<T, Conn: Clone, C> Clone for RedisStorage<T, Conn, C>

fn clone(&self) -> Self

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<T, Conn, C> Debug for RedisStorage<T, Conn, C>

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

Formats the value using the given formatter.

impl<T, Conn, C> Storage for RedisStorage<T, Conn, C>

where T: Serialize + DeserializeOwned + Send + 'static + Unpin + Sync, Conn: ConnectionLike + Send + Sync + 'static, C: Codec<Compact = Vec<u8>> + Send + 'static,

type Job = T

The type of job that can be persisted

type Error = RedisError

The error produced by the storage

type Context = RedisContext

This is the type that storages store as the metadata related to a job

type Compact = Vec<u8>

The format that the storage persists the jobs usually Vec<u8>

async fn push_request( &mut self, req: Request<T, RedisContext>, ) -> Result<Parts<Self::Context>, RedisError>

Pushes a constructed request to a storage

async fn push_raw_request( &mut self, req: Request<Self::Compact, Self::Context>, ) -> Result<Parts<Self::Context>, Self::Error>

Pushes a constructed request to a storage

async fn schedule_request( &mut self, req: Request<Self::Job, RedisContext>, on: i64, ) -> Result<Parts<Self::Context>, RedisError>

Push a request into the scheduled set

async fn len(&mut self) -> Result<i64, RedisError>

Return the number of pending jobs from the queue

async fn fetch_by_id( &mut self, job_id: &TaskId, ) -> Result<Option<Request<Self::Job, RedisContext>>, RedisError>

Fetch a job given an id

async fn update( &mut self, job: Request<T, RedisContext>, ) -> Result<(), RedisError>

Update a job details

async fn reschedule( &mut self, job: Request<T, RedisContext>, wait: Duration, ) -> Result<(), RedisError>

Reschedule a job

async fn is_empty(&mut self) -> Result<bool, RedisError>

Returns true if there is no jobs in the storage

async fn vacuum(&mut self) -> Result<usize, RedisError>

Vacuum the storage, removes done and killed jobs

fn push( &mut self, job: Self::Job, ) -> impl Future<Output = Result<Parts<Self::Context>, Self::Error>> + Send

Pushes a job to a storage

fn schedule( &mut self, job: Self::Job, on: i64, ) -> impl Future<Output = Result<Parts<Self::Context>, Self::Error>> + Send

Push a job with defaults into the scheduled set