Struct Queue 

pub struct Queue {
    pub connection_pool: Pool<ConnectionManager<PgConnection>>,
}

An async queue that can be used to enqueue tasks. It uses a PostgreSQL storage. It must be connected to perform any operation. To connect a Queue to the PostgreSQL database call the get_connection method. A Queue can be created with the TypedBuilder.

// Set DATABASE_URL enviroment variable if you would like to try this function.
pub fn connection_pool(pool_size: u32) -> r2d2::Pool<r2d2::ConnectionManager<PgConnection>> {
  let database_url = env::var("DATABASE_URL").expect("DATABASE_URL must be set");
  let manager = r2d2::ConnectionManager::<PgConnection>::new(database_url);
  r2d2::Pool::builder()
    .max_size(pool_size)
    .build(manager)
    .unwrap()
}

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

Fields

connection_pool: Pool<ConnectionManager<PgConnection>>

Implementations

impl Queue

pub fn builder() -> QueueBuilder<((),)>

Create a builder for building Queue. On the builder, call .connection_pool(...) to set the values of the fields. Finally, call .build() to create the instance of Queue.

impl Queue

pub fn get_connection(&self) -> Result<PoolConnection, QueueError>

Connect to the db if not connected

pub fn schedule_task_query( connection: &mut PgConnection, params: &dyn Runnable, ) -> Result<Task, QueueError>

pub fn insert_query( connection: &mut PgConnection, params: &dyn Runnable, scheduled_at: DateTime<Utc>, ) -> Result<Task, QueueError>

pub fn fetch_task_query( connection: &mut PgConnection, task_type: String, ) -> Option<Task>

pub fn fetch_and_touch_query( connection: &mut PgConnection, task_type: String, ) -> Result<Option<Task>, QueueError>

pub fn find_task_by_id_query( connection: &mut PgConnection, id: &Uuid, ) -> Option<Task>

pub fn remove_all_tasks_query( connection: &mut PgConnection, ) -> Result<usize, QueueError>

pub fn remove_all_scheduled_tasks_query( connection: &mut PgConnection, ) -> Result<usize, QueueError>

pub fn remove_tasks_of_type_query( connection: &mut PgConnection, task_type: &str, ) -> Result<usize, QueueError>

pub fn remove_task_by_metadata_query( connection: &mut PgConnection, task: &dyn Runnable, ) -> Result<usize, QueueError>

pub fn remove_task_query( connection: &mut PgConnection, id: &Uuid, ) -> Result<usize, QueueError>

pub fn update_task_state_query( connection: &mut PgConnection, task: &Task, state: FangTaskState, ) -> Result<Task, QueueError>

pub fn fail_task_query( connection: &mut PgConnection, task: &Task, error: &str, ) -> Result<Task, QueueError>

pub fn schedule_retry_query( connection: &mut PgConnection, task: &Task, backoff_seconds: u32, error: &str, ) -> Result<Task, QueueError>

Trait Implementations

impl Clone for Queue

fn clone(&self) -> Queue

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Queueable for Queue

fn remove_task_by_metadata( &self, task: &dyn Runnable, ) -> Result<usize, QueueError>

To use this function task has to be uniq. uniq() has to return true. If task is not uniq this function will not do anything.

fn fetch_and_touch_task( &self, task_type: String, ) -> Result<Option<Task>, QueueError>

This method should retrieve one task of the task_type type. After fetching it should update the state of the task to FangTaskState::InProgress.

fn insert_task(&self, params: &dyn Runnable) -> Result<Task, QueueError>

Enqueue a task to the queue, The task will be executed as soon as possible by the worker of the same type created by an WorkerPool.

fn schedule_task(&self, params: &dyn Runnable) -> Result<Task, QueueError>

Schedule a task.

fn remove_all_scheduled_tasks(&self) -> Result<usize, QueueError>

Remove all tasks that are scheduled in the future.

fn remove_all_tasks(&self) -> Result<usize, QueueError>

The method will remove all tasks from the queue

fn remove_tasks_of_type(&self, task_type: &str) -> Result<usize, QueueError>

Removes all tasks that have the specified task_type.

fn remove_task(&self, id: &Uuid) -> Result<usize, QueueError>

Remove a task by its id.

fn update_task_state( &self, task: &Task, state: FangTaskState, ) -> Result<Task, QueueError>

Update the state field of the specified task See the FangTaskState enum for possible states.

fn fail_task(&self, task: &Task, error: &str) -> Result<Task, QueueError>

Update the state of a task to FangTaskState::Failed and set an error_message.

fn find_task_by_id(&self, id: &Uuid) -> Option<Task>

fn schedule_retry( &self, task: &Task, backoff_seconds: u32, error: &str, ) -> Result<Task, QueueError>