Struct MemoryStorage 

pub struct MemoryStorage<Args, Ctx = Extensions> { /* private fields */ }

In-memory queue that is based on channels

Example

let mut backend = MemoryStorage::new();

Features

Feature	Status	Description
Backend	✅	Basic Backend functionality
TaskSink	✅	Ability to push new tasks
Serialization	❌	Serialization support for arguments
PipeExt	⚠️	Allow other backends to pipe to this backend
MakeShared	❌	Share the same storage across multiple workers
Update	❌	Allow updating a task
FetchById	❌	Allow fetching a task by its ID
Reschedule	❌	Reschedule a task
ResumeById	❌	Resume a task by its ID
ResumeAbandoned	❌	Resume abandoned tasks
Vacuum	❌	Vacuum the task storage
Workflow	⚠️	Flexible enough to support workflows
WaitForCompletion	⚠️	Wait for tasks to complete without blocking
RegisterWorker	❌	Allow registering a worker with the backend
ListWorkers	❌	List all workers registered with the backend
ListTasks	❌	List all tasks in the backend

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

Tests:

Backend

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    async fn task(task: u32, worker: WorkerContext) {
        // Do some work 
    }
    let worker = WorkerBuilder::new("backend-test")
        .backend(backend)
        .build(task);
    let _ = worker.stream().take(1).collect::<Vec<_>>().await;
}

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();
}

Implementations

impl<Args: Send + 'static> MemoryStorage<Args, Extensions>

pub fn new() -> Self

Create a new in-memory storage

Trait Implementations

impl<Args: 'static + Clone + Send, Ctx: 'static + Default> Backend for MemoryStorage<Args, Ctx>

type Args = Args

The type of arguments the backend handles.

type IdType = RandomId

The type used to uniquely identify tasks.

type Context = Ctx

Context associated with each task.

type Error = SendError

The error type returned by backend operations

type Stream = Pin<Box<dyn Stream<Item = Result<Option<Task<Args, Ctx, RandomId>>, SendError>> + Send>>

A stream of tasks provided by the backend.

type Layer = Identity

The type representing backend middleware layer.

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

A stream representing heartbeat signals.

fn heartbeat(&self, _: &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: Clone + Send + 'static, Ctx: Default + 'static> BackendExt for MemoryStorage<Args, Ctx>

type Codec = IdentityCodec

The codec used for serialization/deserialization of tasks.

type Compact = Args

The compact representation of task arguments.

type CompactStream = Pin<Box<dyn Stream<Item = Result<Option<Task<Args, <MemoryStorage<Args, Ctx> as Backend>::Context, RandomId>>, <MemoryStorage<Args, Ctx> as Backend>::Error>> + 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, Ctx> Debug for MemoryStorage<Args, Ctx>

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

Formats the value using the given formatter.

impl<Args: Send + 'static> Default for MemoryStorage<Args, Extensions>

fn default() -> Self

Returns the “default value” for a type.

impl<Args, Ctx> Sink<Task<Args, Ctx, RandomId>> for MemoryStorage<Args, Ctx>

type Error = SendError

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<Args, Ctx, RandomId>, ) -> 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<Args, Ctx> Stream for MemoryStorage<Args, Ctx>

type Item = Task<Args, Ctx, RandomId>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.