Struct apalis_sql::postgres::PostgresStorage

pub struct PostgresStorage<T, C = JsonCodec<Value>>
where
    C: Codec,
{ /* private fields */ }

Available on crate feature postgres only.

Represents a Storage that persists to Postgres

Implementations

impl PostgresStorage<()>

pub fn migrations() -> Migrator

Get postgres migrations without running them

pub async fn setup(pool: &Pool<Postgres>) -> Result<(), Error>

Do migrations for Postgres

impl<T> PostgresStorage<T>

pub fn new(pool: PgPool) -> Self

New Storage from PgPool

pub fn new_with_config(pool: PgPool, config: Config) -> Self

New Storage from PgPool and custom config

pub fn pool(&self) -> &Pool<Postgres>

Expose the pool for other functionality, eg custom migrations

pub fn config(&self) -> &Config

Expose the config

impl<T, C: Codec> PostgresStorage<T, C>

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

Expose the codec

impl<T, C: Codec> PostgresStorage<T, C>

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

Kill a job

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

Puts the job instantly back into the queue Another Worker may consume

pub async fn reenqueue_orphaned(&mut self, count: i32) -> Result<(), Error>

Reenqueue jobs that have been abandoned by their workers

Trait Implementations

impl<T, Res, C> Ack<T, Res> for PostgresStorage<T, C>

where T: Sync + Send, Res: Serialize + Sync, C: Codec<Compact = Value> + Send,

type Context = SqlContext

The data to fetch from context to allow acknowledgement

type AckError = Error

The error returned by the ack

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

Acknowledges successful processing of the given request

impl<T, C, Res> Backend<Request<T>, Res> for PostgresStorage<T, C>

where T: Serialize + DeserializeOwned + Sync + Send + Unpin + 'static, C: Codec<Compact = Value> + Send + 'static,

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

The stream to be produced by the backend

type Layer = AckLayer<PostgresStorage<T, C>, T, Res>

Returns the final decoration of layers

fn poll<Svc>(self, worker: WorkerId) -> Poller<Self::Stream, Self::Layer>

Returns a poller that is ready for streaming

impl<T, C: Codec> Clone for PostgresStorage<T, 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, C: Codec> Debug for PostgresStorage<T, C>

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

Formats the value using the given formatter.

impl<T, C> Storage for PostgresStorage<T, C>

where T: Serialize + DeserializeOwned + Send + 'static + Unpin + Sync, C: Codec<Compact = Value> + Send + 'static,

async fn push(&mut self, job: Self::Job) -> Result<TaskId, Error>

Push a job to Postgres Storage

SQL Example

Select apalis.push_job(job_type::text, job::json);

type Job = T

The type of job that can be persisted

type Error = Error

The error produced by the storage

type Identifier = TaskId

Jobs must have Ids.

async fn schedule(&mut self, job: Self::Job, on: i64) -> Result<TaskId, Error>

Push a job into the scheduled set

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

Fetch a job given an id

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

Return the number of pending jobs from the queue

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

Reschedule a job

async fn update(&mut self, job: Request<Self::Job>) -> Result<(), Error>

Update a job details

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

Returns true if there is no jobs in the storage

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

Vacuum the storage, removes done and killed jobs