Rust Live Queries

Use KalamLinkClient::live() to receive materialized row snapshots. The SDK merges inserts, updates, and deletes; your async loop handles rows, server errors, transport failures, and shutdown.

Prepare the live-query example

Create app.live_messages with the shared inbox schema. Use the same authenticated identity for the subscriber and the test writes because this is a USER table.

BASH

1cargo new live-inbox2cd live-inbox3cargo add kalam-client --path ../KalamDB/link/kalam-client4cargo add tokio --features macros,rt-multi-thread,signal,time5export KALAMDB_URL=http://localhost:29006export KALAMDB_USERNAME=your-user7export KALAMDB_PASSWORD=your-password

Run the Rust inbox

Save this as src/main.rs, then run cargo run. It loads a fresh snapshot, prints each current row list, and waits for changes until Ctrl+C. Credentials come from the local terminal; use your application’s authentication strategy for deployment.

RUST

1use std::{error::Error, time::Duration};2use kalam_client::{AuthProvider, KalamLinkClient, LiveRowsEvent};3 4#[tokio::main]5async fn main() -> Result<(), Box<dyn Error>> {6    let client = KalamLinkClient::builder()7        .base_url(std::env::var("KALAMDB_URL")8            .unwrap_or_else(|_| "http://localhost:2900".into()))9        .auth(AuthProvider::basic_auth(10            std::env::var("KALAMDB_USERNAME")?,11            std::env::var("KALAMDB_PASSWORD")?,12        ))13        .timeout(Duration::from_secs(30))14        .build()?;15 16    // Disconnect even when connect/subscription setup or the loop fails.17    let result = watch(&client).await;18    client.disconnect().await;19    result20}21 22async fn watch(client: &KalamLinkClient) -> Result<(), Box<dyn Error>> {23    client.connect().await?;24    let mut live = client.live(25        "SELECT id, room, body FROM app.live_messages WHERE room = 'main'",26    ).await?;27 28    println!("Loading inbox…");29    let mut received_rows = false;30    let first_rows_deadline = tokio::time::sleep(Duration::from_secs(15));31    tokio::pin!(first_rows_deadline);32    let shutdown = tokio::signal::ctrl_c();33    tokio::pin!(shutdown);34 35    let outcome: Result<(), Box<dyn Error>> = loop {36        tokio::select! {37            signal = &mut shutdown => break signal.map_err(Into::into),38            _ = &mut first_rows_deadline, if !received_rows => {39                break Err("No initial rows received within 15 seconds".into());40            }41            event = live.next() => {42                match event {43                    Some(Ok(LiveRowsEvent::Rows { rows, .. })) => {44                        received_rows = true;45                        if rows.is_empty() { println!("Inbox is empty"); }46                        else { println!("{} messages", rows.len()); }47                        for row in rows {48                            let id = row.get("id").and_then(|v| v.as_text()).unwrap_or("");49                            let body = row.get("body").and_then(|v| v.as_text()).unwrap_or("");50                            println!("{id}: {body}");51                        }52                    }53                    Some(Ok(LiveRowsEvent::Error { code, message, .. })) => {54                        break Err(format!("Subscription [{code}]: {message}").into());55                    }56                    Some(Err(error)) => break Err(error.into()),57                    None => break Err("Live subscription ended unexpectedly".into()),58                }59            }60        }61    };62 63    let close_result = live.close().await;64    outcome?;65    close_result?;66    Ok(())67}

Handle each outcome deliberately

Result of `live.next()`	Meaning	Example behavior
`Some(Ok(LiveRowsEvent::Rows { .. }))`	Current rows, including an empty snapshot	Replace/print the list
`Some(Ok(LiveRowsEvent::Error { .. }))`	Server subscription error with a code	Report it and exit cleanly
`Some(Err(error))`	SDK/transport failure	Return the error after closing
`None`	Stream ended	Report unexpected termination

event? alone does not handle LiveRowsEvent::Error: that variant is inside Ok. Keep both error paths. The first-row deadline applies only to startup; a quiet inbox after loading is normal. live.close() runs after success or loop failure, and disconnect() runs even if subscription setup fails.

This CLI deliberately exits on reported errors. A long-running service can reopen after correcting the cause, using bounded backoff for retryable failures. Do not spin on invalid SQL or denied access, and do not start a second subscription while the first is still owned by another loop.

Verify changes and cleanup

Run the test insert, update, and delete one at a time using the same identity. On a fresh table, expect empty → one message → edited message → empty. The whole current list prints after each change; this is not an append-only event stream.

Press Ctrl+C and confirm the process exits. Temporarily use a missing table to exercise server errors. Stop the server to exercise transport recovery or failure handling. After restarting the example, it obtains a new snapshot with the current rows.

Options and recovery

Use live_with_config(config, LiveRowsConfig { limit: Some(100), ..Default::default() }) to cap the direct materializer’s retained rows. SubscriptionOptions::with_last_rows(100) restricts the starting history; it is separate from retention and snapshot batch_size. The complete inbox above leaves these unset so it can show every matching row.

Keep the SQL in SELECT ... FROM ... WHERE ... form; no joins, aggregates, ORDER BY, or LIMIT. Include the primary key. Sort for presentation if needed. For raw events, use live_events().

The shared connection manages reconnects. Durable resume requires cached state and a checkpoint belonging to the same query and identity. Do not save only the cursor and later expect a full list from an empty materializer. Replay is bounded; follow the WebSocket recovery contract for gaps and expired cursors.

More examples

The repository also includes a self-contained live-inbox smoke example that creates a temporary table, inserts a row, and cleans it up. Its table setup differs from the long-running inbox above.