Struct FileStorage 

pub struct FileStorage<Args, A: Adapter> { /* private fields */ }

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<u32> = 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
BackendFactory	✅	Share the same 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(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 = SteppedFlow::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<u32> + Backend>(backend: B) {};
    assert_wait_for_completion(backend);
}

Implementations

impl<A: Adapter, Args> FileStorage<Args, A>

pub fn new(path: impl AsRef<Path>) -> Result<Self, FileStorageError<A>>

where A: Default,

Creates a new FileStorage instance using the specified file path.

pub fn new_temp() -> Result<Self, FileStorageError<A>>

where A: Default,

Creates a new temporary FileStorage instance.

pub fn set_result( &mut self, line_id: usize, result: Value, ) -> Result<(), FileStorageError<A>>

Attach a result to the entry identified by line_id.

pub fn flush(&mut self) -> Result<(), FileStorageError<A>>

Persist all pending changes under an exclusive file lock.

pub fn tick(&mut self) -> Result<(), FileStorageError<A>>

Drive periodic syncing. Call this from your timer loop.

Trait Implementations

impl<Args, Res, A> Acknowledge<Res> for FileStorage<Args, A>

where Args: Send + 'static + Debug, Res: Serialize, A: Adapter + Clone,

type Error = FileStorageError<A>

The error type returned by the acknowledgment process

type Future = Ready<Result<(), <FileStorage<Args, A> as Acknowledge<Res>>::Error>>

The future returned by the ack method

fn ack( &mut self, res: &Result<Res, BoxDynError>, ctx: &ExecutionContext, ) -> Self::Future

Acknowledge the result of a task processing

impl<Args, A> Backend for FileStorage<Args, A>

where Args: 'static + Send + Serialize + for<'de> Deserialize<'de> + Unpin, A: Adapter + Unpin, A::Error: Error + Send + Sync + 'static,

type Task = Task<Value>

The type of task the backend emits.

type Error = FileStorageError<A>

The error type returned by backend operations

fn poll_ready( &mut self, cx: &mut Context<'_>, _: &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<'_>, _: &WorkerContext, ) -> Poll<Option<Result<Self::Task, Self::Error>>>

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

fn poll_close( &mut self, _: &mut Context<'_>, _: &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, A> BackendConfig for FileStorage<Args, A>

where Args: 'static + Send + Serialize + for<'de> Deserialize<'de> + Unpin, A: Adapter + Unpin + Clone, A::Error: Error + Send + Sync + 'static,

type Args = Args

The type of argument this backend emits

type Id = RandomId

The internal type of this Backend’s TaskId

type Kind = Durable

This defines the kind of Backend

type Config = PathBuf

The config for the backend

type Layer = AcknowledgeLayer<FileStorage<Args, A>>

The type representing backend middleware layer.

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

Returns the backend’s middleware layer.

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

Returns the config associated with the backend.

impl<Args, A> Clone for FileStorage<Args, A>

where A: Adapter + Clone,

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, A: Debug + Adapter> Debug for FileStorage<Args, A>

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

Formats the value using the given formatter.

impl<A: Adapter + Unpin, Args: Unpin> Sink<Task<Value>> for FileStorage<Args, A>

type Error = FileStorageError<A>

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>, ) -> 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>, _: &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<A: Adapter + Unpin, Args: Unpin> Stream for FileStorage<Args, A>

type Item = Result<(usize, RawTask), FileStorageError<A>>

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<Args, A> TryNewBackend for FileStorage<Args, A>

where A: Adapter + Default + Unpin + Sync + Clone, A::Error: Error + Send + Sync + 'static, Args: Serialize + DeserializeOwned + 'static + Send + Sync + Unpin,

type Backend = FileStorage<Args, A>

The final backend after decoration

fn try_new(config: Self::Config) -> Result<Self::Backend, Self::Error>

Attempt to construct a new backend instance from its configuration.

impl<Args, A, O> WaitForCompletion<O> for FileStorage<Args, A>

where O: Send + DeserializeOwned + 'static + Unpin + Serialize, A: Adapter + Unpin + Sync + Clone, A::Error: Error + Send + Sync + 'static, Args: Serialize + DeserializeOwned + 'static + Send + Sync + Unpin,

type ResultStream = Pin<Box<dyn Stream<Item = Result<TaskResult<O>, FileStorageError<A>>> + Send>>

The result stream type yielding task results

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

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

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

Check current status of tasks without waiting

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

Wait for a single task to complete, yielding its result

impl<Args, A> WireFormatBackend for FileStorage<Args, A>

where Args: 'static + Send + Serialize + for<'de> Deserialize<'de> + Unpin, A: Adapter + Unpin, A::Error: Error + Send + Sync + 'static,

type Codec = JsonCodec<Value>

The codec used to encode and decode tasks.

type Compact = Value

The compact representation of task arguments.

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

Returns a reference to the codec used by the backend.