Fang

Background task processing library for Rust. It uses Postgres DB as a task queue.

Features

Asynk feature uses tokio. Workers are started in tokio tasks.
Blocking feature uses std::thread. Workers are started in a separated threads.

Installation

Add this to your Cargo.toml

Blocking feature

toml [dependencies] fang = { version = "0.9" , features = ["blocking"], default-features = false }

Asynk feature

toml [dependencies] fang = { version = "0.9" , features = ["asynk"], default-features = false }

Both features

toml fang = { version = "0.9" }

Supports rustc 1.62+

Create fang_tasks table in the Postgres database. The migration can be found in the migrations directory.

Usage

Defining a task

Blocking feature

Every task should implement fang::Runnable trait which is used by fang to execute it.

```rust use fang::Error; use fang::Runnable; use fang::typetag; use fang::PgConnection; use fang::serde::{Deserialize, Serialize};

[derive(Serialize, Deserialize)]

[serde(crate = "fang::serde")]

struct MyTask { pub number: u16, }

[typetag::serde]

impl Runnable for MyTask { fn run(&self, _queue: &dyn Queueable) -> Result<(), Error> { println!("the number is {}", self.number);

    Ok(())
}

// If you want to make the tasks of this type uniq.
fn uniq(&self) -> bool {
  true
}

// This will be useful if you want to filter tasks.
// default value: "common".to_string()
fn task_type(&self) -> String {
  "my_task".to_string()
}

// This will be useful if you would like to schedule tasks.
// default value: None (task is not schedule just executes when it is fetched)
fn cron(&self) -> Option<Scheduled> {
    //               sec  min   hour   day of month   month   day of week   year
    //               be careful works only with UTC hour.
    //               https://www.timeanddate.com/worldclock/timezone/utc
    let expression = "0/20 * * * Aug-Sep * 2022/1";
    Some(Scheduled::CronPattern(expression.to_string()))
}

} ```

As you can see from the example above, the trait implementation has #[typetag::serde] attribute which is used to deserialize the task.

The second parameter of the run function is a is an struct that implements fang::Queueable (fang::Queue for example), You can re-use it to manipulate the task queue, for example, to add a new job during the current job's execution. If you don't need it, just ignore it.

Asynk feature

Every task should implement fang::AsyncRunnable trait which is used by fang to execute it.

Also be careful to not to call with the same name two impl of AsyncRunnable, because will cause a fail with typetag. ```rust use fang::AsyncRunnable; use fang::asynk::asyncqueue::AsyncQueueable; use fang::serde::{Deserialize, Serialize}; use fang::asynctrait;

[derive(Serialize, Deserialize)]

[serde(crate = "fang::serde")]

struct AsyncTask { pub number: u16, }

[typetag::serde]

[async_trait]

impl AsyncRunnable for AsyncTask { async fn run(&self, queueable: &mut dyn AsyncQueueable) -> Result<(), Error> { Ok(()) } // this func is optional to impl // Default task-type it is common fn tasktype(&self) -> String { "my-task-type".to_string() }

// If you want to make the tasks of this type uniq.
fn uniq(&self) -> bool {
  true
}

// This will be useful if you would like to schedule tasks.
// default value: None (task is not schedule just executes when it is fetched)
fn cron(&self) -> Option<Scheduled> {
    //               sec  min   hour   day of month   month   day of week   year
    //               be careful works only with UTC hour.
    //               https://www.timeanddate.com/worldclock/timezone/utc
    let expression = "0/20 * * * Aug-Sep * 2022/1";
    Some(Scheduled::CronPattern(expression.to_string()))
}

} ```

In both modules, tasks can be schedule to be execute once. Use Scheduled::ScheduleOnce enum variant to schedule in specific datetime.

Datetimes and cron pattern are interpreted in UTC timezone. So you should introduce an offset to schedule in the desire hour.

Example:

If your hour is UTC + 2 and you would like to schedule at 11:00 all days, your expression will be this one.

rust let expression = "0 0 9 * * * *";

Enqueuing a task

Blocking feature

To enqueue a task use Queue::enqueue_task

For Postgres Backend. ```rust use fang::Queue;

// create a r2d2 pool

// create a fang queue

let queue = Queue::builder().connection_pool(pool).build();

let taskinserted = queue.inserttask(&MyTask::new(1)).unwrap();

```

Queue::insert_task method will insert a task with uniqueness or not it depends on uniq method defined in a task. If uniq is set to true and the task is already in storage this will return the task in the storage.

Asynk feature

To enqueue a task use AsyncQueueable::insert_task, depending of the backend that you prefer you will need to do it with a specific queue.

For Postgres backend. ```rust use fang::asynk::async_queue::AsyncQueue; use fang::NoTls; use fang::AsyncRunnable;

// Create a AsyncQueue let maxpoolsize: u32 = 2;

let mut queue = AsyncQueue::builder() // Postgres database url .uri("postgres://postgres:postgres@localhost/fang") // Max number of connections that are allowed .maxpoolsize(maxpoolsize) .build();

// Always connect first in order to perform any operation queue.connect(NoTls).await.unwrap();

``` For easy example we are using NoTls type, if for some reason you would like to encrypt postgres traffic.

You can implement a Tls type.

It is well documented for openssl and native-tls

rust // AsyncTask from first example let task = AsyncTask { 8 }; let task_returned = queue .insert_task(&task as &dyn AsyncRunnable) .await .unwrap();

Starting workers

Blocking feature

Every worker runs in a separate thread. In case of panic, they are always restarted.

Use WorkerPool to start workers. Use WorkerPool::builder to create your worker pool and run tasks.

```rust use fang::WorkerPool; use fang::Queue;

// create a Queue

let mut workerpool = WorkerPool::::builder() .queue(queue) .numberofworkers(3u32) // if you want to run tasks of the specific kind .tasktype("mytask_type") .build();

worker_pool.start(); ```

Asynk feature

Every worker runs in a separate tokio task. In case of panic, they are always restarted. Use AsyncWorkerPool to start workers.

```rust use fang::asynk::asyncworkerpool::AsyncWorkerPool;

// Need to create a queue // Also insert some tasks

let mut pool: AsyncWorkerPool> = AsyncWorkerPool::builder() .numberofworkers(maxpoolsize) .queue(queue.clone()) // if you want to run tasks of the specific kind .tasktype("mytask_type") .build();

pool.start().await; ```

Check out:

Simple Worker Example - simple worker example
Simple Cron Worker Example - simple worker example
Simple Async Worker Example - simple async worker example
Simple Cron Async Worker Example - simple async worker example
El Monitorro - telegram feed reader. It uses Fang blocking module to synchronize feeds and deliver updates to users.
weatherbotrust - A bot that provides weather info. It uses Fang asynk module to process updates from Telegram users and schedule weather info.

Configuration

Blocking feature

Just use TypeBuilder done for WorkerPool.

Asynk feature

Just use TypeBuilder done for AsyncWorkerPool.

Configuring the type of workers

Configuring retention mode

By default, all successfully finished tasks are removed from the DB, failed tasks aren't.

There are three retention modes you can use:

rust pub enum RetentionMode { KeepAll, // doesn't remove tasks RemoveAll, // removes all tasks RemoveFinished, // default value }

Set retention mode with worker pools TypeBuilder in both modules.

Configuring sleep values

Blocking feature

You can use use SleepParams to confugure sleep values:

rust pub struct SleepParams { pub sleep_period: Duration, // default value is 5 seconds pub max_sleep_period: Duration, // default value is 15 seconds pub min_sleep_period: Duration, // default value is 5 seconds pub sleep_step: Duration, // default value is 5 seconds }

If there are no tasks in the DB, a worker sleeps for sleep_period and each time this value increases by sleep_step until it reaches max_sleep_period. min_sleep_period is the initial value for sleep_period. All values are in seconds.

Use set_sleep_params to set it: rust let sleep_params = SleepParams { sleep_period: Duration::from_secs(2), max_sleep_period: Duration::from_secs(6), min_sleep_period: Duration::from_secs(2), sleep_step: Duration::from_secs(1), };

Set sleep params with worker pools TypeBuilder in both modules.

Contributing

Fork it!
Create your feature branch (git checkout -b my-new-feature)
Commit your changes (git commit -am 'Add some feature')
Push to the branch (git push origin my-new-feature)
Create new Pull Request

Running tests locally

Install diesel_cli. cargo install diesel_cli
Install docker in your machine.
Run a Postgres docker container. (See in Makefile.) make db
Run the migrations make diesel
Run tests make tests
Run dirty//long tests, DB must be recreated afterwards. make ignored
Kill docker container make stop

Authors

Ayrat Badykov (@ayrat555)
Pepe Márquez (@pxp9)