Function main

src/sql-server-util/examples/cdc.rs:49–152

()

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::from_default_env())
        .init();

    let mut config = tiberius::Config::new();

    config.host("localhost");
    config.port(1433);
    config.database("materialize");
    config.authentication(tiberius::AuthMethod::sql_server("SA", "password123?"));
    config.trust_cert();

    // Open one client to stream changes.
    let tunnel = TunnelConfig::Direct {
        resolved_addresses: Default::default(),
    };
    let mz_config = Config::new(config.clone(), tunnel.clone(), InTask::No);
    let mut client_1 = Client::connect(mz_config).await?;
    tracing::info!("connection 1 successful!");

    let capture_instances = ["materialize_t1", "materialize_t2"];
    let metrics = LoggingSqlServerCdcMetrics;
    let mut cdc_handle = client_1.cdc(capture_instances, metrics);

    cdc_handle.wait_for_ready().await?;

    // Open a second client that we can use to cleanup the underlying change tables.
    let mz_config = Config::new(config.clone(), tunnel, InTask::No);
    let mut client_2 = Client::connect(mz_config).await?;
    tracing::info!("connection 2 successful!");

    let tables = mz_sql_server_util::inspect::get_tables_for_capture_instance(
        &mut client_2,
        capture_instances,
    )
    .await?;
    let mut instance_to_lsn = BTreeMap::new();

    for table in tables {
        // Get an initial snapshot of the table.
        let (lsn, snapshot) = cdc_handle
            .snapshot(&table, 1, mz_repr::GlobalId::User(1))
            .await?;

        instance_to_lsn.insert(Arc::clone(&table.capture_instance.name), lsn);

        let mut snapshot = std::pin::pin!(snapshot);
        while let Some(result) = snapshot.next().await {
            let row = result?;
            tracing::info!("snapshot: {} {row:?}", &table.capture_instance.name);
        }
    }

    // Initialize all capture instances at the LSN we just snapshotted.
    for instance in capture_instances {
        let lsn = instance_to_lsn
            .remove(instance)