Struct CdcStream

pub struct CdcStream<'a> {
    client: &'a mut Client,
    capture_instances: BTreeMap<Arc<str>, Option<Lsn>>,
    poll_interval: Duration,
    max_lsn_wait: Duration,
}

A stream of changes from a table in SQL Server that has CDC enabled.

SQL Server does not have an API to push or notify consumers of changes, so we periodically poll the upstream source.

See: https://learn.microsoft.com/en-us/sql/relational-databases/system-tables/change-data-capture-tables-transact-sql?view=sql-server-ver16

Fields

client: &'a mut Client

Client we use for querying SQL Server.

capture_instances: BTreeMap<Arc<str>, Option<Lsn>>

Upstream capture instances we’ll list changes from.

poll_interval: Duration

How often we poll the upstream for changes.

max_lsn_wait: Duration

How long we’ll wait for SQL Server to return a max LSN before taking a snapshot.

Note: When CDC is first enabled in an instance of SQL Server it can take a moment for it to “completely” startup. Before starting a TRANSACTION for our snapshot we’ll wait this duration for SQL Server to report an LSN and thus indicate CDC is ready to go.

Implementations

impl<'a> CdcStream<'a>

pub(crate) fn new( client: &'a mut Client, capture_instances: BTreeMap<Arc<str>, Option<Lsn>>, ) -> Self

pub fn start_lsn(self, capture_instance: &str, lsn: Lsn) -> Self

Set the Lsn that we should start streaming changes from.

If the provided Lsn is not available, the stream will return an error when first polled.

pub fn poll_interval(self, interval: Duration) -> Self

The cadence at which we’ll poll the upstream SQL Server database for changes.

Default is 1 second.

pub fn max_lsn_wait(self, wait: Duration) -> Self

The max duration we’ll wait for SQL Server to return an LSN before taking a snapshot.

When CDC is first enabled in SQL Server it can take a moment before it is fully setup and starts reporting LSNs.

Default is 10 seconds.

pub async fn snapshot<'b>( &'b mut self, instances: Option<BTreeSet<Arc<str>>>, ) -> Result<(Lsn, BTreeMap<Arc<str>, usize>, impl Stream<Item = (Arc<str>, Result<Row, SqlServerError>)> + use<'b, 'a>), SqlServerError>

Takes a snapshot of the upstream table that the specified capture_instance is replicating changes from.

An optional instances parameter can be provided to only snapshot the specified instances.

pub fn into_stream( self, ) -> impl Stream<Item = Result<CdcEvent, SqlServerError>> + use<'a>

Consume self returning a Stream of CdcEvents.

async fn initialize_start_lsns(&mut self) -> Result<(), SqlServerError>

Determine the Lsn to start streaming changes from.

async fn wait_for_ready(&mut self) -> Result<(), SqlServerError>

If CDC was recently enabled on an instance of SQL Server then it will report NULL for the minimum LSN of a capture instance and/or the maximum LSN of the entire database.

This method runs a retry loop that waits for the upstream DB to report good values. It should be called before taking the initial CdcStream::snapshot to ensure the system is ready to proceed with CDC.