Struct JsonStorage 

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

Available on crate feature json only.

A backend that persists to a file using json encoding

Warning: This backend is not optimized for high-throughput scenarios and is best suited for development, testing, or low-volume workloads.

Example

Creates a temporary JSON storage backend

let mut backend = JsonStorage::new_temp().unwrap();

Features

Feature	Status	Description
Backend	✅	Basic Backend functionality
TaskSink	✅	Ability to push new tasks
Serialization	✅ ❗	Serialization support for arguments. Only accepts json
Web Interface	⚠️	Expose a web interface for monitoring tasks
FetchById	⚠️	Allow fetching a task by its ID
RegisterWorker	❌	Allow registering a worker with the backend
PipeExt	✅	Allow other backends to pipe to this backend
MakeShared	✅	Share the same JSON storage across multiple workers via SharedJsonStore
Workflow	✅	Flexible enough to support workflows
WaitForCompletion	✅	Wait for tasks to complete without blocking
ResumeById	⚠️	Resume a task by its ID
ResumeAbandoned	⚠️	Resume abandoned tasks
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();
}

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

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> JsonStorage<Args>

pub fn new(path: impl Into<PathBuf>) -> Result<Self>

Creates a new JsonStorage instance using the specified file path.

pub fn new_temp() -> Result<Self, Error>

Creates a new temporary JsonStorage instance.

pub fn remove(&mut self, key: &TaskKey) -> Result<Option<TaskWithMeta>>

Removes a task from the storage.

pub fn reload(&mut self) -> Result<()>

Reload data from disk, useful if the file was modified externally

pub fn clear(&mut self) -> Result<()>

Clear all data from memory and file

pub fn update_status( &mut self, old_key: &TaskKey, new_status: Status, ) -> Result<bool>

Update the status of an existing key

pub fn get(&self, key: &TaskKey) -> Option<TaskWithMeta>

Retrieves a task from the storage.

Trait Implementations

impl<Args> Backend for JsonStorage<Args>

where Args: 'static + Send + Serialize + DeserializeOwned + Unpin,

type Args = Args

The type of arguments the backend handles.

type IdType = RandomId

The type used to uniquely identify tasks.

type Error = SendError

The error type returned by backend operations

type Context = Map<String, Value>

Context associated with each task.

type Stream = Pin<Box<dyn Stream<Item = Result<Option<Task<Args, Map<String, Value>>>, SendError>> + Send>>

A stream of tasks provided by the backend.

type Layer = AcknowledgeLayer<JsonAck<Args>>

The type representing backend middleware layer.

type Beat = Pin<Box<dyn Stream<Item = Result<(), <JsonStorage<Args> 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: 'static + Send + Serialize + DeserializeOwned + Unpin> BackendExt for JsonStorage<Args>

type Codec = JsonCodec<Value>

The codec used for serialization/deserialization of tasks.

type Compact = Value

The compact representation of task arguments.

type CompactStream = Pin<Box<dyn Stream<Item = Result<Option<Task<<JsonStorage<Args> as BackendExt>::Compact, Map<String, Value>>>, SendError>> + 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> Clone for JsonStorage<Args>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<Args> ConfigExt for JsonStorage<Args>

where Args: 'static + Send + Serialize + DeserializeOwned + Unpin,

fn get_queue(&self) -> Queue

Get the queue configuration

impl<Args: Debug> Debug for JsonStorage<Args>

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

Formats the value using the given formatter.

impl<Args: Unpin + Serialize + DeserializeOwned> Sink<Task<Value, Map<String, Value>>> for JsonStorage<Args>

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<Value, Map<String, Value>>, ) -> 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: DeserializeOwned + Unpin> Stream for JsonStorage<Args>

type Item = Task<Args, Map<String, Value>>

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.

impl<Res: 'static + DeserializeOwned + Send, Args> WaitForCompletion<Res> for JsonStorage<Args>

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

Available on crate feature sleep only.

type ResultStream = Pin<Box<dyn Stream<Item = Result<TaskResult<Res>, SendError>> + 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