Function get_changes_asc

src/sql-server-util/src/inspect.rs:235–254

Queries the specified capture instance and returns all changes from `[start_lsn, end_lsn)`, ordered by `start_lsn` in an ascending fashion.

(
    client: &mut Client,
    capture_instance: &str,
    start_lsn: Lsn,
    end_lsn: Lsn,
    filter: RowFilterOption,
)

/// Queries the specified capture instance and returns all changes from
/// `[start_lsn, end_lsn)`, ordered by `start_lsn` in an ascending fashion.
pub fn get_changes_asc(
    client: &mut Client,
    capture_instance: &str,
    start_lsn: Lsn,
    end_lsn: Lsn,
    filter: RowFilterOption,
) -> impl Stream<Item = Result<tiberius::Row, SqlServerError>> + Send {
    const START_LSN_COLUMN: &str = "__$start_lsn";
    let query = format!(
        "SELECT * FROM cdc.{function}(@P1, @P2, N'{filter}') ORDER BY {START_LSN_COLUMN} ASC;",
        function = quote_identifier(&format!("fn_cdc_get_all_changes_{capture_instance}"))
    );
    client.query_streaming(
        query,
        &[
            &start_lsn.as_bytes().as_slice(),
            &end_lsn.as_bytes().as_slice(),
        ],
    )
}

/// Cleans up the change table associated with the specified `capture_instance` by
/// deleting `max_deletes` entries with a `start_lsn` less than `low_water_mark`.