2021-03-28 01:57:17 +00:00
|
|
|
use std::borrow::Cow;
|
|
|
|
use std::fmt::Debug;
|
|
|
|
use std::time::Duration;
|
|
|
|
|
|
|
|
use serde::Serialize;
|
|
|
|
use sqlx::Postgres;
|
|
|
|
use uuid::Uuid;
|
|
|
|
|
2021-03-29 20:39:07 +00:00
|
|
|
/// Type for building a job to send.
|
2021-03-28 01:57:17 +00:00
|
|
|
#[derive(Debug, Clone)]
|
2021-03-29 20:39:07 +00:00
|
|
|
pub struct JobBuilder<'a> {
|
2021-03-28 01:57:17 +00:00
|
|
|
id: Uuid,
|
|
|
|
delay: Duration,
|
|
|
|
channel_name: &'a str,
|
|
|
|
channel_args: &'a str,
|
2021-03-29 02:05:20 +00:00
|
|
|
retries: u32,
|
2021-03-28 01:57:17 +00:00
|
|
|
retry_backoff: Duration,
|
|
|
|
commit_interval: Option<Duration>,
|
|
|
|
ordered: bool,
|
|
|
|
name: &'a str,
|
|
|
|
payload_json: Option<Cow<'a, str>>,
|
|
|
|
payload_bytes: Option<&'a [u8]>,
|
|
|
|
}
|
|
|
|
|
2021-03-29 20:39:07 +00:00
|
|
|
impl<'a> JobBuilder<'a> {
|
|
|
|
/// Prepare to send a job with the specified name.
|
2021-03-28 01:57:17 +00:00
|
|
|
pub fn new(name: &'a str) -> Self {
|
|
|
|
Self::new_with_id(Uuid::new_v4(), name)
|
|
|
|
}
|
2021-03-29 20:39:07 +00:00
|
|
|
/// Prepare to send a job with the specified name and ID.
|
2021-03-28 01:57:17 +00:00
|
|
|
pub fn new_with_id(id: Uuid, name: &'a str) -> Self {
|
|
|
|
Self {
|
|
|
|
id,
|
|
|
|
delay: Duration::from_secs(0),
|
|
|
|
channel_name: "",
|
|
|
|
channel_args: "",
|
|
|
|
retries: 4,
|
|
|
|
retry_backoff: Duration::from_secs(1),
|
|
|
|
commit_interval: None,
|
|
|
|
ordered: false,
|
|
|
|
name,
|
|
|
|
payload_json: None,
|
|
|
|
payload_bytes: None,
|
|
|
|
}
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
/// Use the provided function to set any number of configuration
|
|
|
|
/// options at once.
|
|
|
|
pub fn set_proto<'b>(
|
|
|
|
&'b mut self,
|
|
|
|
proto: impl FnOnce(&'b mut Self) -> &'b mut Self,
|
|
|
|
) -> &'b mut Self {
|
|
|
|
proto(self)
|
|
|
|
}
|
|
|
|
/// Set the channel name (default "").
|
2021-03-28 01:57:17 +00:00
|
|
|
pub fn set_channel_name(&mut self, channel_name: &'a str) -> &mut Self {
|
|
|
|
self.channel_name = channel_name;
|
|
|
|
self
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
/// Set the channel arguments (default "").
|
2021-03-28 01:57:17 +00:00
|
|
|
pub fn set_channel_args(&mut self, channel_args: &'a str) -> &mut Self {
|
|
|
|
self.channel_args = channel_args;
|
|
|
|
self
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
/// Set the number of retries after the initial attempt (default 4).
|
|
|
|
pub fn set_retries(&mut self, retries: u32) -> &mut Self {
|
2021-03-28 01:57:17 +00:00
|
|
|
self.retries = retries;
|
|
|
|
self
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
/// Set the initial backoff for retries (default 1s).
|
2021-03-28 01:57:17 +00:00
|
|
|
pub fn set_retry_backoff(&mut self, retry_backoff: Duration) -> &mut Self {
|
|
|
|
self.retry_backoff = retry_backoff;
|
|
|
|
self
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
/// Set the commit interval for two-phase commit (default disabled).
|
2021-03-28 01:57:17 +00:00
|
|
|
pub fn set_commit_interval(&mut self, commit_interval: Option<Duration>) -> &mut Self {
|
|
|
|
self.commit_interval = commit_interval;
|
|
|
|
self
|
|
|
|
}
|
2021-03-29 20:39:07 +00:00
|
|
|
/// Set whether this job is strictly ordered with respect to other ordered
|
|
|
|
/// job in the same channel (default false).
|
2021-03-28 01:57:17 +00:00
|
|
|
pub fn set_ordered(&mut self, ordered: bool) -> &mut Self {
|
|
|
|
self.ordered = ordered;
|
|
|
|
self
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
|
2021-03-29 20:39:07 +00:00
|
|
|
/// Set a delay before this job is executed (default none).
|
2021-03-28 01:57:17 +00:00
|
|
|
pub fn set_delay(&mut self, delay: Duration) -> &mut Self {
|
|
|
|
self.delay = delay;
|
|
|
|
self
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
|
2021-03-29 20:39:07 +00:00
|
|
|
/// Set a raw JSON payload for the job.
|
2021-03-28 01:57:17 +00:00
|
|
|
pub fn set_raw_json(&mut self, raw_json: &'a str) -> &mut Self {
|
|
|
|
self.payload_json = Some(Cow::Borrowed(raw_json));
|
|
|
|
self
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
|
2021-03-29 20:39:07 +00:00
|
|
|
/// Set a raw binary payload for the job.
|
2021-03-28 01:57:17 +00:00
|
|
|
pub fn set_raw_bytes(&mut self, raw_bytes: &'a [u8]) -> &mut Self {
|
|
|
|
self.payload_bytes = Some(raw_bytes);
|
|
|
|
self
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
|
2021-03-29 20:39:07 +00:00
|
|
|
/// Set a JSON payload for the job.
|
2021-03-29 02:05:20 +00:00
|
|
|
pub fn set_json<T: ?Sized + Serialize>(
|
|
|
|
&mut self,
|
|
|
|
value: &T,
|
|
|
|
) -> Result<&mut Self, serde_json::Error> {
|
2021-03-28 01:57:17 +00:00
|
|
|
let value = serde_json::to_string(value)?;
|
|
|
|
self.payload_json = Some(Cow::Owned(value));
|
|
|
|
Ok(self)
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
|
2021-03-29 20:39:07 +00:00
|
|
|
/// Spawn the job using the given executor. This might be a connection
|
2021-03-29 02:05:20 +00:00
|
|
|
/// pool, a connection, or a transaction.
|
2021-03-28 01:57:17 +00:00
|
|
|
pub async fn spawn<'b, E: sqlx::Executor<'b, Database = Postgres>>(
|
|
|
|
&self,
|
|
|
|
executor: E,
|
|
|
|
) -> Result<Uuid, sqlx::Error> {
|
|
|
|
sqlx::query(
|
|
|
|
"SELECT mq_insert(ARRAY[($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)::mq_new_t])",
|
|
|
|
)
|
|
|
|
.bind(self.id)
|
|
|
|
.bind(self.delay)
|
|
|
|
.bind(self.retries as i32)
|
|
|
|
.bind(self.retry_backoff)
|
|
|
|
.bind(self.channel_name)
|
|
|
|
.bind(self.channel_args)
|
|
|
|
.bind(self.commit_interval)
|
|
|
|
.bind(self.ordered)
|
|
|
|
.bind(self.name)
|
|
|
|
.bind(self.payload_json.as_deref())
|
|
|
|
.bind(self.payload_bytes)
|
|
|
|
.execute(executor)
|
|
|
|
.await?;
|
|
|
|
Ok(self.id)
|
|
|
|
}
|
|
|
|
}
|
2021-03-29 02:05:20 +00:00
|
|
|
|
2021-03-29 20:39:07 +00:00
|
|
|
/// Commit the specified jobs. The jobs should have been previously spawned
|
2021-03-29 02:05:20 +00:00
|
|
|
/// with the two-phase commit option enabled.
|
|
|
|
pub async fn commit<'b, E: sqlx::Executor<'b, Database = Postgres>>(
|
|
|
|
executor: E,
|
2021-03-29 20:39:07 +00:00
|
|
|
job_ids: &[Uuid],
|
2021-03-29 02:05:20 +00:00
|
|
|
) -> Result<(), sqlx::Error> {
|
|
|
|
sqlx::query("SELECT mq_commit($1)")
|
2021-03-29 20:39:07 +00:00
|
|
|
.bind(job_ids)
|
2021-03-29 02:05:20 +00:00
|
|
|
.execute(executor)
|
|
|
|
.await?;
|
|
|
|
Ok(())
|
|
|
|
}
|
2021-09-20 12:27:48 +00:00
|
|
|
|
2022-07-05 15:04:42 +00:00
|
|
|
/// Clear jobs from the specified channels.
|
2021-09-20 12:27:48 +00:00
|
|
|
pub async fn clear<'b, E: sqlx::Executor<'b, Database = Postgres>>(
|
|
|
|
executor: E,
|
|
|
|
channel_names: &[&str],
|
|
|
|
) -> Result<(), sqlx::Error> {
|
|
|
|
sqlx::query("SELECT mq_clear($1)")
|
|
|
|
.bind(channel_names)
|
|
|
|
.execute(executor)
|
|
|
|
.await?;
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2022-07-05 15:04:42 +00:00
|
|
|
/// Clear jobs from all channels.
|
2021-09-20 12:27:48 +00:00
|
|
|
pub async fn clear_all<'b, E: sqlx::Executor<'b, Database = Postgres>>(
|
|
|
|
executor: E,
|
|
|
|
) -> Result<(), sqlx::Error> {
|
|
|
|
sqlx::query("SELECT mq_clear_all()")
|
|
|
|
.execute(executor)
|
|
|
|
.await?;
|
|
|
|
Ok(())
|
|
|
|
}
|
2022-02-05 14:29:53 +00:00
|
|
|
|
|
|
|
/// Check if a job with that ID exists
|
|
|
|
pub async fn exists<'b, E: sqlx::Executor<'b, Database = Postgres>>(
|
|
|
|
executor: E,
|
|
|
|
id: Uuid,
|
|
|
|
) -> Result<bool, sqlx::Error> {
|
|
|
|
let exists = sqlx::query_scalar("SELECT EXISTS(SELECT id FROM mq_msgs WHERE id = $1)")
|
|
|
|
.bind(id)
|
|
|
|
.fetch_one(executor)
|
|
|
|
.await?;
|
|
|
|
Ok(exists)
|
|
|
|
}
|