Struct ReplicationClient 

pub struct ReplicationClient { /* private fields */ }

PostgreSQL logical replication client.

This client spawns a background worker task that maintains the replication connection and streams events to the consumer via a bounded channel.

Example

use pgwire_replication::client::{ReplicationClient, ReplicationEvent};
use pgwire_replication::config::ReplicationConfig;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = ReplicationConfig::new(
        "localhost",
        "postgres",
        "password",
        "mydb",
        "my_slot",
        "my_pub",
    );

    let mut client = ReplicationClient::connect(config).await?;

    while let Some(ev) = client.recv().await? {
        match ev {
            ReplicationEvent::XLogData { data, wal_end, .. } => {
                process_change(&data);
                client.update_applied_lsn(wal_end);
            }
            ReplicationEvent::KeepAlive { .. } => {}
            ReplicationEvent::StoppedAt { reached } => {
                println!("Reached stop LSN: {reached}");
                break;
            }
            _ => {}
        }
    }

    Ok(())
}

fn process_change(_data: &bytes::Bytes) {
    // user-defined
}

Implementations

impl ReplicationClient

pub async fn connect(cfg: ReplicationConfig) -> Result<Self>

Connect to PostgreSQL and start streaming replication events.

This establishes a TCP connection (optionally upgrading to TLS), authenticates, and starts the replication stream. Events are buffered in a channel of size config.buffer_events.

Errors

Returns an error if:

• TCP connection fails
• TLS handshake fails (when enabled)
• Authentication fails
• Replication slot doesn’t exist
• Publication doesn’t exist
• Unix socket does not exist (when host starts with /)
• TLS requested with Unix socket connection

pub async fn connect_gated(cfg: ReplicationConfig) -> Result<Self>

Connect and authenticate like connect, but do not start replication: the session waits for simple_query (and the IDENTIFY_SYSTEM / TIMELINE_HISTORY helpers) until start sends START_REPLICATION. Everything runs on the one authenticated session, so what is read before the start describes the server the stream then opens on. Dropping the client before start ends the session.

pub async fn simple_query(&mut self, sql: &str) -> Result<Vec<QueryRow>>

Run sql with the simple query protocol on a gated session before start; returns its rows (text format). A server error leaves the session usable.

pub async fn identify_system(&mut self) -> Result<IdentifySystem>

IDENTIFY_SYSTEM on a gated session.

pub async fn timeline_history( &mut self, timeline: u32, ) -> Result<TimelineHistory>

TIMELINE_HISTORY for timeline on a gated session (the server has no history file for timeline 1).

pub async fn start(&mut self) -> Result<()>

Send START_REPLICATION on a gated session; then wait_started reports whether the server accepted it.

pub async fn wait_started(&mut self) -> Result<()>

Wait until the server has accepted START_REPLICATION: from then on the replication slot is held by this session, and no other session can acquire or advance it. connect returns before that, as connecting, authentication and START_REPLICATION run in the background worker. Returns the worker’s error if it ended first.

pub async fn recv(&mut self) -> Result<Option<ReplicationEvent>>

Receive the next replication event.

• Ok(Some(event)) => received an event
• Ok(None) => replication ended normally (stop requested or stop_at_lsn reached)
• Err(e) => replication ended abnormally

pub fn update_applied_lsn(&self, lsn: Lsn)

Update the applied/durable LSN reported to the server.

Semantics: call this only once you have durably persisted all events up to lsn. This update is monotonic and cheap. A raised LSN wakes the worker, which sends it once status_interval has passed since its last standby status update, also on an idle stream; updates in between coalesce into one status carrying the newest LSN.

pub fn feedback_pending_since(&self) -> Option<Instant>

When the oldest applied-position raise not yet reported to the server happened, with a feedback observer configured; None otherwise or when everything raised was reported. Readable after the session ended, so a caller replacing a failed session can carry it over with carry_feedback_pending.

pub fn carry_feedback_pending(&self, since: Instant)

Carry over a pending start from a session this one replaces (from its feedback_pending_since): the next update that advances the reported position reports its wait from there, or from an older start of this session. No effect without a feedback observer.

pub fn metrics(&self) -> Arc<ReplicationMetrics>

Returns a handle to the live replication metrics.

The returned Arc shares the same counters the background worker updates, so reads reflect current progress. Cheap to clone and call repeatedly; nothing here blocks the worker.

pub fn stop(&self)

Request the worker to stop gracefully.

After calling this, recv() will return remaining buffered events, then Ok(None) once the worker exits cleanly.

This sends a CopyDone message to the server to cleanly terminate the replication stream.

pub fn is_running(&self) -> bool

pub async fn join(self) -> Result<()>

Wait for the worker task to complete and return its result.

This consumes the client. Use this for diagnostics or to ensure clean shutdown after calling stop().

pub fn abort(&mut self)

Abort the worker task immediately.

This is a hard cancel and does not send CopyDone. Prefer stop()/shutdown() for graceful termination.

pub async fn shutdown(&mut self) -> Result<()>

Request a graceful stop and wait for the worker to exit.

Trait Implementations

impl Drop for ReplicationClient

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.