Struct JsonStorage 

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

A backend that persists to a file using json encoding

Features

Feature	Status	Description
TaskSink	✅	Ability to push new tasks
Serialization	✅ ❗	Serialization support for arguments. Only accepts json
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
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

Examples:

TaskSink Example

use apalis_core::backend::TaskSink;
use apalis_core::worker::context::WorkerContext;
use apalis_core::worker::builder::WorkerBuilder;
#[tokio::main]
async fn main() {
    let mut backend = {
    use apalis_core::backend::json::JsonStorage;
    JsonStorage::new_temp().unwrap()
};

    use apalis_core::backend::TaskSink;
    backend.push(42).await.unwrap();

    async fn task(task: u32, worker: WorkerContext) {
        worker.stop().unwrap();
    }
    let worker = WorkerBuilder::new("rango-tango")
        .backend(backend)
        .build(task);
    worker.run().await.unwrap();
}

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<JsonStorage<Args>, 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: 'static + Send + DeserializeOwned + Unpin> Backend<Args> for JsonStorage<Args>

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<Args>>::Error>> + Send>>

A stream representing heartbeat signals.

type Codec = JsonCodec<Value>

The codec used for serialization/deserialization of tasks.

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 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: Debug> Debug for JsonStorage<Args>

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

Formats the value using the given formatter.

impl<Args: Unpin + Serialize> Sink<Task<Args, 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<Args, 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, Compact> WaitForCompletion<Res, Compact> for JsonStorage<Compact>

where Compact: Send + DeserializeOwned + 'static + Unpin + Sync,

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