pgmq 0.34.0-alpha.5

#[cfg(feature = "rust-postgres")]
pub mod postgres;
#[cfg(feature = "tokio-postgres")]
pub mod tokio_postgres;

impl<T, H> TryFrom<::tokio_postgres::Row> for crate::types::Message<T, H>
where
    T: for<'de> serde::Deserialize<'de>,
    H: for<'de> serde::Deserialize<'de>,
{
    type Error = crate::errors::PgmqError;

    fn try_from(value: ::tokio_postgres::Row) -> Result<Self, Self::Error> {
        let message = T::deserialize(value.try_get::<_, serde_json::Value>("message")?)?;
        let headers =
            if let Some(headers) = value.try_get::<_, Option<serde_json::Value>>("headers")? {
                Some(H::deserialize(headers)?)
            } else {
                None
            };
        Ok(Self {
            msg_id: value.try_get("msg_id")?,
            read_ct: value.try_get("read_ct")?,
            enqueued_at: value.try_get("enqueued_at")?,
            last_read_at: value.try_get("last_read_at")?,
            vt: value.try_get("vt")?,
            message,
            headers,
        })
    }
}

type SqlParam<'a> = (&'a (dyn postgres_types::ToSql + Sync), postgres_types::Type);

/// This macro defines all the functions required to implement [`crate::queue::Queue`] for both
/// `postgres` and `tokio-postgres` crates. This is possible because the implementations are
/// nearly identical. The only differences are which connection trait is used, one needs the
/// connection/executor to be `&mut C` and one can be `&C`, and one requires `await`-ing DB results.
///
/// The functions defined by this macro are designed to be used by the [`crate::queue::Queue`]
/// definition generated by the [`crate::queue::macros::impl_queue`] macro.
macro_rules! rust_postgres_functions {
    (
        /// The common connection/executor trait to use to implement all the functions
        $executor_trait:path,
        /// Macro that expands to a reference (either mutable or immutable) of the provided type
        $ref_type:tt,
        /// Macro to transform the result (e.g., either simply use it directly, or perform an `await`)
        $transform_result:tt
    ) => {
        async fn create<C>(
            executor: $ref_type!(C),
            queue_name: crate::types::QueueName<'_>,
        ) -> Result<(), crate::PgmqError>
        where
            C: $executor_trait,
        {
            let params: [crate::queue::rust_postgres::SqlParam; _] =
                [(&*queue_name, postgres_types::Type::TEXT)];
            let result = executor.execute_typed(crate::queue::sql::CREATE, &params);
            $transform_result!(result)?;
            Ok(())
        }

        async fn send<C>(
            executor: $ref_type!(C),
            queue_name: crate::types::QueueName<'_>,
            message: serde_json::Value,
            headers: serde_json::Value,
            delay: crate::types::VisibilityTimeoutOffset,
        ) -> Result<i64, crate::PgmqError>
        where
            C: $executor_trait,
        {
            let params: [crate::queue::rust_postgres::SqlParam; _] = [
                (&*queue_name, postgres_types::Type::TEXT),
                (&message, postgres_types::Type::JSONB),
                (&headers, postgres_types::Type::JSONB),
                (&*delay, postgres_types::Type::INT4),
            ];
            let row = executor.query_typed_one(crate::queue::sql::SEND, &params);
            let row = $transform_result!(row)?;
            Ok(row.try_get(0)?)
        }

        async fn read<C, T, H>(
            executor: $ref_type!(C),
            queue_name: crate::types::QueueName<'_>,
            visibility_timeout: crate::types::VisibilityTimeoutOffset,
            quantity: i32,
        ) -> Result<Vec<crate::Message<T, H>>, crate::PgmqError>
        where
            C: $executor_trait,
            T: Send + for<'de> serde::Deserialize<'de>,
            H: Send + for<'de> serde::Deserialize<'de>,
        {
            let params: [crate::queue::rust_postgres::SqlParam; _] = [
                (&*queue_name, postgres_types::Type::TEXT),
                (&*visibility_timeout, postgres_types::Type::INT4),
                (&quantity, postgres_types::Type::INT4),
            ];
            let rows = executor.query_typed(crate::queue::sql::READ, &params);
            let rows = $transform_result!(rows)?;
            rows.into_iter()
                .map(|row| crate::Message::<T, H>::try_from(row))
                .collect::<Result<Vec<crate::Message<T, H>>, crate::errors::PgmqError>>()
        }
    };
}
pub(crate) use rust_postgres_functions;