Struct Worker 

pub struct Worker<Args, Backend, Svc, Middleware> { /* private fields */ }

A worker polls a backend and processes tasks using a service.

Its the core component responsible for task polling, execution, and lifecycle management.

Example

Basic example:

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let mut storage = MemoryStorage::new();
    for i in 0..5 {
        storage.push(i).await?;
    }

    async fn handler(task: u32) {
        println!("Processing task: {task}");
    }

    let worker = WorkerBuilder::new("worker-1")
        .backend(storage)
        .build(handler);

    worker.run().await?;
    Ok(())
}

See module level documentation for more details.

Implementations

impl<Args, B, Svc, M> Worker<Args, B, Svc, M>

pub fn new(context: WorkerContext, backend: B, service: Svc, layers: M) -> Self

Build a worker that is ready for execution

impl<Args, S, M, FB> Worker<Args, FB, S, M>

where FB: BackendConfig + Backend<Task = Task<FB::Args>> + Send + Unpin, S: Service<Task<FB::Args>> + Send + 'static, FB::Args: Send + 'static, Args: Send + 'static, FB::Error: Into<BoxDynError> + Send + 'static, M: Layer<LifecycleService<S>>, FB::Layer: Layer<<M as Layer<LifecycleService<S>>>::Service>, <FB::Layer as Layer<<M as Layer<LifecycleService<S>>>::Service>>::Service: Service<Task<FB::Args>> + Send + 'static, <<FB::Layer as Layer<<M as Layer<LifecycleService<S>>>::Service>>::Service as Service<Task<<FB as BackendConfig>::Args>>>::Future: Send, <<FB::Layer as Layer<<M as Layer<LifecycleService<S>>>::Service>>::Service as Service<Task<<FB as BackendConfig>::Args>>>::Error: Into<BoxDynError> + Send + Sync + 'static, <<FB::Layer as Layer<<M as Layer<LifecycleService<S>>>::Service>>::Service as Service<Task<<FB as BackendConfig>::Args>>>::Response: Send + Sync + 'static,

pub async fn run(self) -> Result<(), WorkerError>

Run the worker until completion

Example

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let mut storage = MemoryStorage::new();
    for i in 0..5 {
        storage.push(i).await?;
    }

    async fn handler(task: u32) {
        println!("Processing task: {task}");
    }

    let worker = WorkerBuilder::new("worker-1")
        .backend(storage)
        .build(handler);

    worker.run().await?;
    Ok(())
}

pub async fn run_until<Fut, Err>(self, signal: Fut) -> Result<(), WorkerError>

where Fut: Future<Output = Result<(), Err>> + Send + 'static, FB: Send, M: Send, Err: Into<WorkerError> + Send + 'static,

Run the worker until a shutdown signal future is complete.

pub async fn run_with_ctx<F, Fut>(self, fut: F) -> Result<(), WorkerError>

where F: FnMut(WorkerContext) -> Fut + Send + 'static, Fut: Future<Output = Result<(), WorkerError>> + Send, FB: Send, M: Send,

Run the worker until a shutdown signal future is complete.

Note: Using this function requires you to call ctx.stop() in the future to completely stop the worker.

This can also be very powerful with pausing and resuming the worker using the context.

Example

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let mut storage = MemoryStorage::new();
    for i in 0..5 {
        storage.push(i).await?;
    }
    async fn handler(task: u32) {
        println!("Processing task: {task}");
    }
    let worker = WorkerBuilder::new("worker-1")
        .backend(storage)
        .build(handler);
    worker.run_with_ctx(|ctx| async move {
        sleep(Duration::from_secs(1)).await;
        ctx.stop()?;
        Ok(())
    }).await?;
    Ok(())
}

pub fn stream( self, ) -> impl Stream<Item = Result<Event, WorkerError>> + use<Args, S, M, FB>

Returns a stream that will yield events as they occur within the worker’s lifecycle

Example

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let mut storage = MemoryStorage::new();
    for i in 0..5 {
        storage.push(i).await?;
    }
    async fn handler(task: u32) {
        println!("Processing task: {task}");
    }
    let worker = WorkerBuilder::new("worker-1")
        .backend(storage)
        .build(handler);
    let mut stream = worker.stream();
    while let Some(evt) = stream.next().await {
        println!("Event: {:?}", evt);
    }
    Ok(())
}

Trait Implementations

impl<Args, B, Svc, Middleware> Debug for Worker<Args, B, Svc, Middleware>

where Svc: Debug, B: Debug,

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

Formats the value using the given formatter.