apalis

Simple, extensible multithreaded background job and messages processing library for Rust

Features

• Simple and predictable task handling model.
• Task handlers are just an async function with a macro free API.
• Familiar dependency injection for task handlers, similar to actix and axum.
• Take full advantage of the [tower] ecosystem of middleware, services, and utilities.
• Easy to scale, backends are distributed by default.
• Runtime agnostic - Use tokio, smol etc.
• Inbuilt concurrency and parallelism.
• Worker monitoring and graceful shutdown.
• Ability to painlessly expose tasks and workers via APIs
• Persisted cron jobs. Pipe your cronjobs to other backends and distribute them.
• Optional Web interface to help you manage your jobs.

apalis job processing is powered by [tower::Service] which means you have access to the [tower] middleware.

apalis has support for:

Source	Crate	Example
Cron Jobs
Redis
Sqlite
Postgres
MySQL
Amqp
From Scratch

Getting Started

To get started, just add to Cargo.toml

[dependencies]
apalis = { version = "0.7", features = "limit" } # Limit for concurrency
apalis-redis = { version = "0.7" } # Use redis for persistence

Usage

use apalis::prelude::*;
use apalis_redis::RedisStorage;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Serialize)]
struct Email {
    to: String,
}

/// A function called for every job
async fn send_email(job: Email, data: Data<usize>) -> Result<(), Error> {
  /// execute job
  Ok(())
}

#[tokio::main]
async fn main() -> {
    std::env::set_var("RUST_LOG", "debug");
    env_logger::init();
    let redis_url = std::env::var("REDIS_URL").expect("Missing env variable REDIS_URL");
    let conn = apalis_redis::connect(redis_url).await.expect("Could not connect");
    let storage = RedisStorage::new(conn);
    WorkerBuilder::new("email-worker")
      .concurrency(2)
      .data(0usize)
      .backend(storage)
      .build_fn(send_email)
      .run()
      .await;
}

Then

//This can be in another part of the program or another application eg a http server
async fn produce_route_jobs(storage: &mut RedisStorage<Email>) -> Result<()> {
    storage
        .push(Email {
            to: "test@example.com".to_string(),
        })
        .await?;
}

Stepped Tasks

Apalis has beta support for stepped tasks. See complete example

#[tokio::main]
async fn main() -> Result<(), std::io::Error> {
    std::env::set_var("RUST_LOG", "debug");
    tracing_subscriber::fmt::init();
    let redis_url = std::env::var("REDIS_URL").expect("Missing env variable REDIS_URL");
    let conn = apalis_redis::connect(redis_url).await.unwrap();
    let config = apalis_redis::Config::default().set_namespace("stepped-email-task");


    let mut storage = RedisStorage::new_with_config(conn, config);
    storage
        .start_stepped(WelcomeEmail { user_id: 1 })
        .await
        .unwrap();

    // Build steps
    let steps = StepBuilder::new()
        .step_fn(welcome) // Steps are tower services
        .step_fn(campaign)
        .step_fn(complete_campaign);

    WorkerBuilder::new("tasty-banana")
        .data(()) // Shared data for all steps
        .enable_tracing()
        .concurrency(2)
        .backend(storage)
        .build_stepped(steps)
        .on_event(|e| info!("{e}"))
        .run()
        .await;
    Ok(())
}

Feature flags

• tracing (enabled by default) — Support Tracing 👀
• sentry — Support for Sentry exception and performance monitoring
• prometheus — Support Prometheus metrics
• retry — Support direct retrying jobs
• timeout — Support timeouts on jobs
• limit — 💪 Support for concurrency and rate-limiting
• filter — Support filtering jobs based on a predicate
• catch-panic - Catch panics that occur during execution

Storage Comparison

Since we provide a few storage solutions, here is a table comparing them:

How apalis works

Here is a basic example of how the core parts integrate

sequenceDiagram
    participant App
    participant Worker
    participant Backend

    App->>+Backend: Add job to queue
    Backend-->>+Worker: Job data
    Worker->>+Backend: Update job status to 'Running'
    Worker->>+App: Started job
    loop job execution
        Worker-->>-App: Report job progress
    end
    Worker->>+Backend: Update job status to 'completed'

External examples

• Shuttle: Using apalis-cron with shuttle.rs
• Actix-Web: Using apalis-redis with actix-web
• Zino-Example: Using zino

Projects using apalis

• Ryot: A self hosted platform for tracking various facets of your life - media, fitness etc.
• Summarizer: Podcast summarizer
• Universal Inbox: Universal Inbox is a solution that centralizes all your notifications and tasks in one place to create a unique inbox.
• Hatsu: Self-hosted and fully-automated ActivityPub bridge for static sites.

Resources

• Background job processing with rust using actix and redis
• Feasibility of implementing cronjob with Rust programming language
• How to schedule and run cron jobs in Rust using apalis

Web UI

If you are running apalis Board, you can easily manage your jobs. See a working rest API example here

Thanks to

• [tower] - Tower is a library of modular and reusable components for building robust networking clients and servers.
• redis-rs - Redis library for rust
• sqlx - The Rust SQL Toolkit
• cron - A cron expression parser and schedule explorer

Contributing

Please read CONTRIBUTING.md for details on our code of conduct, and the process for submitting pull requests to us.

Versioning

We use SemVer for versioning. For the versions available, see the tags on this repository.

Authors

• Njuguna Mureithi - Initial work - Njuguna Mureithi

See also the list of contributors who participated in this project.

License

This project is licensed under the MIT License - see the LICENSE.md file for details