Struct PgmqNotifyClient 

pub struct PgmqNotifyClient { /* private fields */ }

Unified PGMQ client with comprehensive functionality and notification capabilities

Implementations

impl PgmqClient

pub async fn new(database_url: &str) -> Result<Self>

Create new unified PGMQ client using connection string

pub async fn new_with_config( database_url: &str, config: PgmqNotifyConfig, ) -> Result<Self>

Create new unified PGMQ client with custom configuration

pub async fn new_with_pool(pool: PgPool) -> Self

Create new unified PGMQ client using existing connection pool (BYOP - Bring Your Own Pool)

pub async fn new_with_pool_and_config( pool: PgPool, config: PgmqNotifyConfig, ) -> Self

Create new unified PGMQ client with existing pool and custom configuration

pub async fn create_queue(&self, queue_name: &str) -> Result<()>

Create queue if it doesn’t exist

pub async fn send_json_message<T>( &self, queue_name: &str, message: &T, ) -> Result<i64>

where T: Serialize,

Send generic JSON message to queue

pub async fn send_message_with_delay<T>( &self, queue_name: &str, message: &T, delay_seconds: u64, ) -> Result<i64>

where T: Serialize,

Send message with visibility timeout (delay)

pub async fn read_messages( &self, queue_name: &str, visibility_timeout: Option<i32>, limit: Option<i32>, ) -> Result<Vec<Message<Value>>>

Read messages from queue

pub async fn pop_message( &self, queue_name: &str, ) -> Result<Option<Message<Value>>>

Read messages from queue with pop (single read and delete)

pub async fn read_specific_message<T>( &self, queue_name: &str, message_id: i64, visibility_timeout: i32, ) -> Result<Option<Message<T>>>

where T: DeserializeOwned,

Read a specific message by ID using custom SQL function (for notification event handling)

pub async fn delete_message( &self, queue_name: &str, message_id: i64, ) -> Result<()>

Delete message from queue

pub async fn archive_message( &self, queue_name: &str, message_id: i64, ) -> Result<()>

Archive message (move to archive)

pub async fn set_visibility_timeout( &self, queue_name: &str, message_id: i64, vt_seconds: i32, ) -> Result<()>

Set visibility timeout for a message

Extends or resets the visibility timeout for a message. This is useful for:

• Heartbeat during long-running step processing
• Returning a message to the queue immediately (vt_seconds = 0)

Arguments

• queue_name - The queue containing the message
• message_id - The message ID to update
• vt_seconds - New visibility timeout in seconds from now

Returns

Returns Ok(()) on success, or an error if the message doesn’t exist.

pub async fn purge_queue(&self, queue_name: &str) -> Result<u64>

Purge queue (delete all messages)

pub async fn drop_queue(&self, queue_name: &str) -> Result<()>

Drop queue completely

pub async fn queue_metrics(&self, queue_name: &str) -> Result<QueueMetrics>

Get queue metrics/statistics

pub fn pool(&self) -> &PgPool

Get reference to underlying connection pool for advanced operations

pub fn pgmq(&self) -> &PGMQueue

Get reference to pgmq client for direct access

pub fn config(&self) -> &PgmqNotifyConfig

Get the configuration

pub fn has_notify_capabilities(&self) -> bool

Check if this client has notification capabilities enabled

pub async fn health_check(&self) -> Result<bool>

Health check - verify database connectivity

pub async fn get_client_status(&self) -> Result<ClientStatus>

Get client status information

pub fn extract_namespace(&self, queue_name: &str) -> Option<String>

Extract namespace from queue name using configured pattern

pub async fn create_listener( &self, buffer_size: usize, ) -> Result<PgmqNotifyListener>

Create a notify listener for this client

Arguments

• buffer_size - MPSC channel buffer size (TAS-51: bounded channels)

Note

TAS-51: Migrated from unbounded to bounded channel to prevent OOM during notification bursts. Buffer size should come from configuration based on context:

• Orchestration: config.mpsc_channels.orchestration.event_listeners.pgmq_event_buffer_size
• Worker: config.mpsc_channels.worker.event_listeners.pgmq_event_buffer_size

impl PgmqClient

Helper methods for common queue operations

pub async fn process_namespace_queue( &self, namespace: &str, visibility_timeout: Option<i32>, batch_size: i32, ) -> Result<Vec<Message<Value>>>

Process messages from namespace queue

pub async fn complete_message( &self, namespace: &str, message_id: i64, ) -> Result<()>

Complete message processing (delete from queue)

pub async fn initialize_namespace_queues( &self, namespaces: &[&str], ) -> Result<()>

Initialize standard namespace queues

pub async fn send_with_transaction<T>( &self, queue_name: &str, message: &T, tx: &mut Transaction<'_, Postgres>, ) -> Result<i64>

where T: Serialize,

Send message within a transaction (for atomic operations)

Trait Implementations

impl Clone for PgmqClient

fn clone(&self) -> PgmqClient

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for PgmqClient

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.