Struct InMemoryWorkflow 

pub struct InMemoryWorkflow<Args> { /* private fields */ }

In-memory queue that is based on channels

Example

let mut backend = InMemoryWorkflow::create();

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
BackendFactory	❌	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> InMemoryWorkflow<Args>

pub fn create() -> Shared<Self>

Create a new in-memory storage

impl<Args> InMemoryWorkflow<Args>

pub fn new_with( sender: MemorySink<Vec<u8>>, receiver: BoxedReceiver<Vec<u8>>, ) -> Shared<Self>

Create a storage given a sender and receiver

Trait Implementations

impl<Args> Backend for InMemoryWorkflow<Args>

type Task = Task<Vec<u8>>

The type of task the backend emits.

type Error = MemoryStorageError

The error type returned by backend operations

fn poll_ready( &mut self, cx: &mut Context<'_>, worker: &WorkerContext, ) -> Poll<Result<(), Self::Error>>

Polls whether the backend is ready for the worker to request more work.

fn poll_next( &mut self, cx: &mut Context<'_>, worker: &WorkerContext, ) -> Poll<Option<Result<Self::Task, Self::Error>>>

Polls the backend for the next available task for this worker.

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

Flushes/releases any resources the backend holds (pending acks, open subscriptions, connections) before the worker fully shuts down.

impl<Args> BackendConfig for InMemoryWorkflow<Args>

type Id = RandomId

The internal type of this Backend’s TaskId

type Args = Args

The type of argument this backend emits

type Kind = Durable

This defines the kind of Backend

type Config = ()

The config for the backend

type Layer = StoreResultsLayer

The type representing backend middleware layer.

fn config(&self) -> &Self::Config

Returns the config associated with the backend.

fn middleware(&mut self, _: &mut WorkerContext) -> Self::Layer

Returns the backend’s middleware layer.

impl<Args> Debug for InMemoryWorkflow<Args>

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

Formats the value using the given formatter.

impl<Args> Sink<Task<Vec<u8>>> for InMemoryWorkflow<Args>

where Args: Unpin,

type Error = MemoryStorageError

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>>, ) -> 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, Output> WaitForCompletion<Output> for InMemoryWorkflow<Args>

where Output: DeserializeOwned + Unpin + Send + 'static,

type ResultStream = WaitForStream<Output>

The result stream type yielding task results

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

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

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

Check current status of tasks without waiting

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

Wait for a single task to complete, yielding its result

impl<Args> WireFormatBackend for InMemoryWorkflow<Args>

type Codec = JsonCodec

The codec used to encode and decode tasks.

type Compact = Vec<u8>

The compact representation of task arguments.

fn codec(&self) -> &Self::Codec

Returns a reference to the codec used by the backend.