tasklet 0.3.1

An asynchronous task scheduling library written in Rust

About

tasklet is a task scheduling library written in Rust. It is built over tokio runtime and utilizes green threads in order to run tasks asynchronously.

Dependencies

How to use this library

In your Cargo.toml add:

[dependencies]
tasklet = "0.3.1"

Upgrading from 0.2.x? See the migration notes below — task steps are now async.

Example

Find more examples in the examples folder.

Task steps are asynchronous: every step is a closure that returns a future, so it can .await real work (I/O, timers, network calls) without blocking the runtime.

use log::info;
use simple_logger::SimpleLogger;
use tasklet::task::TaskStepStatusErr::Error;
use tasklet::task::TaskStepStatusOk::Success;
use tasklet::{TaskBuilder, TaskScheduler};

/// A simple example of a task with two steps,
/// that might work or fail sometimes.
#[tokio::main]
async fn main() {
    // Init the logger.
    SimpleLogger::new().init().unwrap();

    // A variable to be passed in the task.
    let mut exec_count = 0;

    // Task scheduler with 1000ms loop frequency.
    let mut scheduler = TaskScheduler::default(chrono::Local);

    // Create a task with 2 steps and add it to the scheduler.
    // The second step fails every second execution.
    // Append the task to the scheduler.
    let _ = scheduler.add_task(
        TaskBuilder::new(chrono::Local)
            .every("1 * * * * * *")
            .description("A simple task")
            .add_step("Step 1", || async {
                info!("Hello from step 1");
                Ok(Success) // Let the scheduler know this step was a success.
            })
            .add_step("Step 2", move || {
                // Snapshot per-run state, then move it into the async block so the
                // returned future is `'static`.
                let count = exec_count;
                exec_count += 1;
                async move {
                    if count % 2 == 0 {
                        Err(Error) // Indicate that this step was a fail.
                    } else {
                        info!("Hello from step 2");
                        Ok(Success) // Indicate that this step was a success.
                    }
                }
            })
            .build(),
    );

    // Execute the scheduler.
    scheduler.run().await;
}

Graceful shutdown

scheduler.run() normally loops forever. To stop it cleanly, grab a SchedulerHandle with scheduler.handle() before running and call shutdown() from anywhere — the current round finishes, the tasks are drained and run() returns:

# use tasklet::TaskScheduler;
# #[tokio::main]
# async fn main() {
let mut scheduler = TaskScheduler::default(chrono::Utc);
let handle = scheduler.handle();

// Stop on Ctrl-C.
tokio::spawn(async move {
    tokio::signal::ctrl_c().await.ok();
    handle.shutdown();
});

scheduler.run().await; // returns once shutdown is requested
# }

You can also drive the shutdown with any future via scheduler.run_until(future) — for example a timer or an OS signal.

Timeouts, retries and lifecycle callbacks

Tasks can be made resilient with a few optional builder settings (all non-breaking, added in 0.3.1):

use std::time::Duration;
use tasklet::task::TaskStepStatusOk::Success;
use tasklet::{RetryPolicy, TaskBuilder};

let _task = TaskBuilder::new(chrono::Local)
    .every("1 * * * * * *")
    // Cancel any single step attempt that runs longer than this.
    .timeout(Duration::from_secs(5))
    // Retry failing steps with exponential backoff (100ms, 200ms, 400ms), capped at 2s.
    .retry(RetryPolicy::exponential(3, Duration::from_millis(100), 2)
        .with_max_delay(Duration::from_secs(2)))
    // Async lifecycle hooks.
    .on_success(|| async { println!("run succeeded"); })
    .on_failure(|| async { eprintln!("run failed"); })
    .on_finish(|| async { println!("task finished its lifecycle"); })
    .add_step("Step", || async { Ok(Success) })
    .build();

• Timeout (.timeout) bounds each individual step attempt; a step that exceeds it is cancelled and treated as a (retryable) failure.
• Retry (.retry) re-attempts a step that returns TaskStepStatusErr::Error (or times out). Use RetryPolicy::fixed or RetryPolicy::exponential. A step returning TaskStepStatusErr::ErrorDelete bypasses retries and removes the task immediately.
• Callbacks (.on_success / .on_failure / .on_finish) are async hooks; on_finish fires once when the task reaches a terminal state.

See examples/retry_timeout_example.rs for a runnable demo.

Migrating from 0.2.x to 0.3.0

• Steps are now async. A step closure must return a future. Wrap synchronous bodies in an async block:

  // 0.2.x
  .add_step("Step", || Ok(Success))
  // 0.3.0
  .add_step("Step", || async { Ok(Success) })
  

  For state that changes between runs, snapshot it in the (FnMut) closure and move the snapshot into the async move block so the future stays 'static.

• TaskGenerator::new now returns a Result. It no longer panics on an invalid cron expression — call ?/.unwrap() on the result.

• TaskScheduler::run can now stop. It returns when a shutdown is requested through a SchedulerHandle; if you never request one, behaviour is unchanged (runs forever).

Author

Stavros Grigoriou (stav121)