2021-06-24 09:58:02 +00:00
< p align = "center" > < img src = "logo.png" alt = "fang" height = "300px" > < / p >
2021-07-11 10:55:52 +00:00
[![Crates.io][s1]][ci] [![docs page][docs-badge]][docs] ![test][ga-test] ![style][ga-style]
2021-06-24 09:58:02 +00:00
# Fang
2022-07-19 13:49:43 +00:00
Background task processing library for Rust. It uses Postgres DB as a task queue.
2021-07-11 07:26:20 +00:00
2022-08-05 07:06:32 +00:00
## Features
2022-08-04 16:47:17 +00:00
- Asynk feature uses `tokio` . Workers are started in tokio tasks.
- Blocking feature uses `std::thread` . Workers are started in a separated threads.
2021-07-25 12:20:16 +00:00
2021-06-24 09:58:02 +00:00
## Installation
1. Add this to your Cargo.toml
2022-08-04 17:11:58 +00:00
#### Blocking feature
2021-06-24 09:58:02 +00:00
```toml
[dependencies]
2022-08-18 11:45:28 +00:00
fang = { version = "0.8" , features = ["blocking"], default-features = false }
2021-06-24 09:58:02 +00:00
```
2022-08-04 17:11:58 +00:00
#### Asynk feature
2022-08-03 08:37:53 +00:00
```toml
[dependencies]
2022-08-18 11:45:28 +00:00
fang = { version = "0.8" , features = ["asynk"], default-features = false }
2022-08-03 08:37:53 +00:00
```
2022-08-05 07:06:32 +00:00
#### Both features
```toml
2022-08-18 11:45:28 +00:00
fang = { version = "0.8" }
2022-08-05 07:06:32 +00:00
```
2022-07-01 06:03:27 +00:00
*Supports rustc 1.62+*
2021-06-24 10:15:10 +00:00
2. Create `fang_tasks` table in the Postgres database. The migration can be found in [the migrations directory ](https://github.com/ayrat555/fang/blob/master/migrations/2021-06-05-112912_create_fang_tasks/up.sql ).
2021-06-24 09:58:02 +00:00
## Usage
2022-07-19 13:49:43 +00:00
### Defining a task
2021-06-24 09:58:02 +00:00
2022-08-04 17:11:58 +00:00
#### Blocking feature
Every task should implement `fang::Runnable` trait which is used by `fang` to execute it.
2021-06-24 09:58:02 +00:00
```rust
2021-07-11 07:26:20 +00:00
use fang::Error;
use fang::Runnable;
use fang::typetag;
2021-07-31 07:40:11 +00:00
use fang::PgConnection;
2022-08-03 08:37:53 +00:00
use fang::serde::{Deserialize, Serialize};
2021-06-24 09:58:02 +00:00
2021-07-11 07:26:20 +00:00
#[derive(Serialize, Deserialize)]
2022-08-03 08:37:53 +00:00
#[serde(crate = "fang::serde")]
2022-07-19 13:49:43 +00:00
struct MyTask {
2021-07-11 07:26:20 +00:00
pub number: u16,
}
2021-06-24 09:58:02 +00:00
2021-07-11 07:26:20 +00:00
#[typetag::serde]
2022-07-19 13:49:43 +00:00
impl Runnable for MyTask {
2021-07-31 07:40:11 +00:00
fn run(& self, _connection: & PgConnection) -> Result< (), Error> {
2021-07-11 07:26:20 +00:00
println!("the number is {}", self.number);
2021-06-24 09:58:02 +00:00
2021-07-11 07:26:20 +00:00
Ok(())
2021-06-24 09:58:02 +00:00
}
2021-07-11 07:26:20 +00:00
}
2021-06-24 09:58:02 +00:00
```
2022-07-19 13:49:43 +00:00
As you can see from the example above, the trait implementation has `#[typetag::serde]` attribute which is used to deserialize the task.
2021-06-24 09:58:02 +00:00
2022-07-19 13:49:43 +00:00
The second parameter of the `run` function is diesel's PgConnection, You can re-use it to manipulate the task queue, for example, to add a new job during the current job's execution. Or you can just re-use it in your own queries if you're using diesel. If you don't need it, just ignore it.
2021-07-31 07:40:11 +00:00
2022-08-03 08:37:53 +00:00
2022-08-04 17:11:58 +00:00
#### Asynk feature
Every task should implement `fang::AsyncRunnable` trait which is used by `fang` to execute it.
2022-08-03 08:37:53 +00:00
2022-08-05 07:06:32 +00:00
Also be careful to not to call with the same name two impl of AsyncRunnable, because will cause a fail with typetag.
2022-08-03 08:37:53 +00:00
```rust
use fang::AsyncRunnable;
use fang::asynk::async_queue::AsyncQueueable;
use fang::serde::{Deserialize, Serialize};
use fang::async_trait;
#[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(())
}
2022-08-05 07:06:32 +00:00
// this func is optional to impl
2022-08-03 08:37:53 +00:00
// Default task-type it is common
fn task_type(& self) -> String {
"my-task-type".to_string()
}
}
```
2022-07-19 13:49:43 +00:00
### Enqueuing a task
2021-06-24 09:58:02 +00:00
2022-08-04 17:11:58 +00:00
#### Blocking feature
2022-08-04 16:47:17 +00:00
To enqueue a task use `Queue::enqueue_task`
2021-06-24 09:58:02 +00:00
```rust
2021-07-31 07:40:11 +00:00
use fang::Queue;
2021-06-24 09:58:02 +00:00
...
2022-07-19 13:49:43 +00:00
Queue::enqueue_task(& MyTask { number: 10 }).unwrap();
2021-06-24 09:58:02 +00:00
```
2022-07-19 13:49:43 +00:00
The example above creates a new postgres connection on every call. If you want to reuse the same postgres connection to enqueue several tasks use Postgres struct instance:
2021-07-04 06:07:29 +00:00
```rust
2021-07-31 07:40:11 +00:00
let queue = Queue::new();
2021-07-04 06:07:29 +00:00
for id in & unsynced_feed_ids {
2022-07-19 13:49:43 +00:00
queue.push_task(& SyncFeedMyTask { feed_id: *id }).unwrap();
2021-07-04 06:07:29 +00:00
}
```
2021-06-24 09:58:02 +00:00
2021-07-31 07:40:11 +00:00
Or you can use `PgConnection` struct:
```rust
2022-07-19 13:49:43 +00:00
Queue::push_task_query(pg_connection, &new_task).unwrap();
2021-07-31 07:40:11 +00:00
```
2022-08-04 17:11:58 +00:00
#### Asynk feature
To enqueue a task use `AsyncQueueable::insert_task` ,
2022-08-03 08:37:53 +00:00
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 max_pool_size: u32 = 2;
let mut queue = AsyncQueue::builder()
// Postgres database url
.uri("postgres://postgres:postgres@localhost/fang")
2022-08-05 07:06:32 +00:00
// Max number of connections that are allowed
2022-08-03 08:37:53 +00:00
.max_pool_size(max_pool_size)
// false if would like Uniqueness in tasks
.duplicated_tasks(true)
.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 ](https://docs.rs/postgres-openssl/latest/postgres_openssl/ ) and [native-tls ](https://docs.rs/postgres-native-tls/latest/postgres_native_tls/ )
```rust
// AsyncTask from first example
let task = AsyncTask { 8 };
let task_returned = queue
.insert_task(& task as & dyn AsyncRunnable)
.await
.unwrap();
```
2021-06-24 09:58:02 +00:00
### Starting workers
2022-08-04 17:11:58 +00:00
#### Blocking feature
2021-06-24 10:15:10 +00:00
Every worker runs in a separate thread. In case of panic, they are always restarted.
2021-06-24 09:58:02 +00:00
2022-08-04 16:47:17 +00:00
Use `WorkerPool` to start workers. `WorkerPool::new` accepts one parameter - the number of workers.
2021-06-24 09:58:02 +00:00
```rust
use fang::WorkerPool;
2021-07-04 06:07:29 +00:00
WorkerPool::new(10).start();
```
2021-12-05 06:43:17 +00:00
Use `shutdown` to stop worker threads, they will try to finish in-progress tasks.
```rust
use fang::WorkerPool;
worker_pool = WorkerPool::new(10).start().unwrap;
worker_pool.shutdown()
```
2021-12-05 06:19:08 +00:00
Using a library like [signal-hook][signal-hook], it's possible to gracefully shutdown a worker. See the
Simple Worker for an example implementation.
2021-07-31 07:40:11 +00:00
2022-08-04 17:11:58 +00:00
#### Asynk feature
2022-08-04 16:47:17 +00:00
Every worker runs in a separate tokio task. In case of panic, they are always restarted.
Use `AsyncWorkerPool` to start workers.
2022-08-03 08:37:53 +00:00
```rust
use fang::asynk::async_worker_pool::AsyncWorkerPool;
2022-08-04 16:47:17 +00:00
// Need to create a queue
2022-08-03 08:37:53 +00:00
// Also insert some tasks
let mut pool: AsyncWorkerPool< AsyncQueue < NoTls > > = AsyncWorkerPool::builder()
.number_of_workers(max_pool_size)
.queue(queue.clone())
.build();
pool.start().await;
```
2021-08-18 19:09:49 +00:00
Check out:
2021-12-05 06:19:08 +00:00
- [Simple Worker Example ](https://github.com/ayrat555/fang/tree/master/fang_examples/simple_worker ) - simple worker example
2022-08-03 08:37:53 +00:00
- [Simple Async Worker Example ](https://github.com/ayrat555/fang/tree/master/fang_examples/simple_async_worker ) - simple async worker example
2022-08-18 14:59:58 +00:00
- [El Monitorro ](https://github.com/ayrat555/el_monitorro ) - telegram feed reader. It uses Fang blocking module to synchronize feeds and deliver updates to users.
- [weather_bot_rust ](https://github.com/pxp9/weather_bot_rust ) - A bot that provides weather info. It uses Fang asynk module to process updates from Telegram users.
2021-07-31 07:40:11 +00:00
2021-07-04 06:07:29 +00:00
### Configuration
2022-08-04 17:11:58 +00:00
#### Blocking feature
2022-08-03 08:37:53 +00:00
2021-07-04 06:07:29 +00:00
To configure workers, instead of `WorkerPool::new` which uses default values, use `WorkerPool.new_with_params` . It accepts two parameters - the number of workers and `WorkerParams` struct.
2022-08-04 17:11:58 +00:00
#### Asynk feature
2022-08-03 08:37:53 +00:00
Just use `TypeBuilder` done for `AsyncWorkerPool` .
2021-07-04 06:07:29 +00:00
### Configuring the type of workers
2022-08-04 17:11:58 +00:00
#### Blocking feature
2022-08-03 08:37:53 +00:00
2021-07-04 06:07:29 +00:00
You can start workers for a specific types of tasks. These workers will be executing only tasks of the specified type.
Add `task_type` method to the `Runnable` trait implementation:
```rust
...
#[typetag::serde]
2022-07-19 13:49:43 +00:00
impl Runnable for MyTask {
2021-07-04 06:07:29 +00:00
fn run(& self) -> Result< (), Error> {
println!("the number is {}", self.number);
Ok(())
}
fn task_type(& self) -> String {
"number".to_string()
}
}
```
Set `task_type` to the `WorkerParamas` :
2021-07-11 07:26:20 +00:00
```rust
2021-07-04 06:07:29 +00:00
let mut worker_params = WorkerParams::new();
worker_params.set_task_type("number".to_string());
WorkerPool::new_with_params(10, worker_params).start();
```
Without setting `task_type` workers will be executing any type of task.
2022-08-04 17:11:58 +00:00
#### Asynk feature
2022-08-03 08:37:53 +00:00
2022-08-04 17:11:58 +00:00
Same as Blocking feature.
2022-08-03 08:37:53 +00:00
Use `TypeBuilder` for `AsyncWorker` .
2021-07-04 06:07:29 +00:00
### 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 `set_retention_mode` :
2022-08-04 17:11:58 +00:00
#### Blocking feature
2022-08-03 08:37:53 +00:00
2021-07-04 06:07:29 +00:00
```rust
let mut worker_params = WorkerParams::new();
worker_params.set_retention_mode(RetentionMode::RemoveAll);
WorkerPool::new_with_params(10, worker_params).start();
```
2022-08-04 17:11:58 +00:00
#### Asynk feature
2022-08-03 08:37:53 +00:00
Set it in `AsyncWorker` `TypeBuilder` .
2021-07-04 06:07:29 +00:00
### Configuring sleep values
2022-08-04 17:11:58 +00:00
#### Blocking feature
2021-07-04 06:07:29 +00:00
You can use use `SleepParams` to confugure sleep values:
```rust
pub struct SleepParams {
2022-08-18 11:39:10 +00:00
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
2022-08-03 08:37:53 +00:00
}
2021-07-04 06:07:29 +00:00
```
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 {
2022-08-18 11:39:10 +00:00
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),
2021-07-04 06:07:29 +00:00
};
let mut worker_params = WorkerParams::new();
worker_params.set_sleep_params(sleep_params);
WorkerPool::new_with_params(10, worker_params).start();
2021-06-24 09:58:02 +00:00
```
2022-08-04 17:11:58 +00:00
#### Asynk feature
2022-08-03 08:37:53 +00:00
Set it in `AsyncWorker` `TypeBuilder` .
2021-06-24 09:58:02 +00:00
2021-07-24 09:13:59 +00:00
## Periodic Tasks
Fang can add tasks to `fang_tasks` periodically. To use this feature first run [the migration with `fang_periodic_tasks` table ](https://github.com/ayrat555/fang/tree/master/migrations/2021-07-24-050243_create_fang_periodic_tasks/up.sql ).
Usage example:
2022-08-04 17:11:58 +00:00
#### Blocking feature
2022-08-03 08:37:53 +00:00
2021-07-24 09:13:59 +00:00
```rust
use fang::Scheduler;
2021-07-31 07:40:11 +00:00
use fang::Queue;
2021-07-24 09:13:59 +00:00
2021-07-31 07:40:11 +00:00
let queue = Queue::new();
2021-07-24 09:13:59 +00:00
2021-07-31 07:40:11 +00:00
queue
2022-08-18 11:39:10 +00:00
.push_periodic_task(& SyncMyTask::default(), 120000)
2021-07-24 09:13:59 +00:00
.unwrap();
2021-07-31 07:40:11 +00:00
queue
2022-08-18 11:39:10 +00:00
.push_periodic_task(& DeliverMyTask::default(), 60000)
2021-07-24 09:13:59 +00:00
.unwrap();
2022-08-18 11:39:10 +00:00
Scheduler::start(Duration::from_secs(10), Duration::from_secs(5));
2021-07-24 09:13:59 +00:00
```
2022-08-18 11:39:10 +00:00
In the example above, `push_periodic_task` is used to save the specified task to the `fang_periodic_tasks` table which will be enqueued (saved to `fang_tasks` table) every specied number of milliseconds.
2021-07-24 09:13:59 +00:00
2022-08-18 11:39:10 +00:00
`Scheduler::start(Duration::from_secs(10), Duration::from_secs(5))` starts scheduler. It accepts two parameters:
- Db check period
- Acceptable error limit - |current_time - scheduled_time| < error
2021-06-24 09:58:02 +00:00
2022-08-04 17:11:58 +00:00
#### Asynk feature
2022-08-03 08:37:53 +00:00
```rust
use fang::asynk::async_scheduler::Scheduler;
use fang::asynk::async_queue::AsyncQueueable;
use fang::asynk::async_queue::AsyncQueue;
2022-08-18 11:45:28 +00:00
use std::time::Duration;
2022-08-18 11:39:10 +00:00
use chrono::Duration as OtherDuration;
2022-08-03 08:37:53 +00:00
2022-08-05 07:06:32 +00:00
// Build a AsyncQueue as before
2022-08-03 08:37:53 +00:00
let schedule_in_future = Utc::now() + OtherDuration::seconds(5);
let _periodic_task = queue.insert_periodic_task(
& AsyncTask { number: 1 },
schedule_in_future,
2022-08-18 11:39:10 +00:00
10000, // period in milliseconds
2022-08-03 08:37:53 +00:00
)
.await;
let check_period: u64 = 1;
let error_margin_seconds: u64 = 2;
let mut scheduler = Scheduler::builder()
2022-08-18 11:39:10 +00:00
.check_period(Duration::from_secs(check_period))
.error_margin_seconds(Duration::from_secs(error_margin_seconds))
2022-08-03 08:37:53 +00:00
.queue(& mut queue as & mut dyn AsyncQueueable)
.build();
// Add some more task in other thread or before loop
2022-08-05 07:06:32 +00:00
// Scheduler Loop
2022-08-03 08:37:53 +00:00
scheduler.start().await.unwrap();
```
2021-06-24 09:58:02 +00:00
## Contributing
1. [Fork it! ](https://github.com/ayrat555/fang/fork )
2. Create your feature branch (`git checkout -b my-new-feature`)
3. Commit your changes (`git commit -am 'Add some feature'`)
4. Push to the branch (`git push origin my-new-feature`)
5. Create new Pull Request
2021-12-05 06:19:08 +00:00
### Running tests locally
```
cargo install diesel_cli
docker run --rm -d --name postgres -p 5432:5432 \
-e POSTGRES_DB=fang \
-e POSTGRES_USER=postgres \
-e POSTGRES_PASSWORD=postgres \
postgres:latest
DATABASE_URL=postgres://postgres:postgres@localhost/fang diesel migration run
// Run regular tests
cargo test --all-features
// Run dirty/long tests, DB must be recreated afterwards
cargo test --all-features -- --ignored --test-threads=1
docker kill postgres
```
2022-08-03 08:37:53 +00:00
## Authors
2021-06-24 09:58:02 +00:00
2022-08-03 08:37:53 +00:00
- Ayrat Badykov (@ayrat555)
2022-08-05 07:06:32 +00:00
2022-08-03 08:37:53 +00:00
- Pepe Márquez (@pxp9)
2021-07-11 10:55:52 +00:00
[s1]: https://img.shields.io/crates/v/fang.svg
[docs-badge]: https://img.shields.io/badge/docs-website-blue.svg
[ci]: https://crates.io/crates/fang
[docs]: https://docs.rs/fang/
[ga-test]: https://github.com/ayrat555/fang/actions/workflows/rust.yml/badge.svg
[ga-style]: https://github.com/ayrat555/fang/actions/workflows/style.yml/badge.svg
2021-12-05 06:19:08 +00:00
[signal-hook]: https://crates.io/crates/signal-hook