Struct PgListener 

pub struct PgListener { /* private fields */ }

PostgreSQL LISTEN handler for receiving NOTIFY events

Creates a long-lived connection that subscribes to one or more channels and streams incoming notifications.

Implementations

impl PgListener

pub async fn from_url(url: &str, channels: Vec<String>) -> Result<Self, PgError>

Create a new listener from a connection URL

Arguments

• url - PostgreSQL connection string
• channels - List of channel names to subscribe to

Errors

Returns error if:

• Cannot connect to database
• Cannot subscribe to channels

Example

let channels = vec!["warpdrive:cache:invalidate".to_string()];
let listener = PgListener::from_url(
    "postgresql://localhost/mydb",
    channels
).await?;

pub async fn new( _pool: &PgPool, _channels: Vec<String>, ) -> Result<Self, PgError>

Create a new listener for the specified channels using a pool

Note: This creates a dedicated connection separate from the pool. The pool is used only to verify connectivity and extract the connection URL.

Arguments

• pool - Connection pool (used to get database URL)
• channels - List of channel names to subscribe to

Errors

Returns error if:

• Cannot obtain connection from pool
• Cannot subscribe to channels

Example

let channels = vec![
    "warpdrive:cache:invalidate".to_string(),
    "warpdrive:circuit:state".to_string(),
];
// Use from_url instead, as pool doesn't expose the connection string
let listener = PgListener::from_url(url, channels).await?;

pub fn stream( &mut self, ) -> Pin<Box<dyn Stream<Item = PgNotification> + Send + '_>>

Get a stream of notifications

Returns a stream that yields PgNotification items as they arrive.

Example

let mut stream = listener.stream();
while let Some(notification) = stream.next().await {
    println!("Channel: {}, Payload: {}", notification.channel, notification.payload);
}

pub async fn subscribe(&self, channel: &str) -> Result<(), PgError>

Subscribe to additional channels

Errors

Returns error if LISTEN command fails

pub async fn unsubscribe(&self, channel: &str) -> Result<(), PgError>

Unsubscribe from channels

Errors

Returns error if UNLISTEN command fails