Skip to Content
RealtimeLive Subscriptions

Realtime Subscriptions

For most Flutter lists, use liveTable() or live(). They emit the current row set. Use liveEvents() only when you need raw insert/update/delete protocol events.

Live SQL must be SELECT ... FROM ... WHERE ... — no ORDER BY or LIMIT.

Need a local cache that works offline? Use Kalam Sync instead of reconciling events yourself.

Start with the Flutter inbox example for a complete widget with loading, empty, error, retry, and cleanup behavior.

Which API

  • liveTable(name) — SELECT * FROM name, materialized rows
  • live(sql) — custom SELECT, materialized rows
  • liveEvents(sql) — Ack / snapshot batches / insert / update / delete

Materialized live rows

live<T>(...) keeps the current row set updated inside the SDK and emits the latest list whenever the query result changes.

DART
final rowsStream = client.live<Map<String, KalamCellValue>>(  "SELECT id, room, body, created_at FROM app.messages WHERE room = 'main'",  lastRows: 20,  limit: 20,  onCheckpoint: (checkpoint) {    print('Applied version: ${checkpoint.lastVersionId}');  },); await for (final rows in rowsStream) {  for (final row in rows) {    print('${row['id']?.asInt()} ${row['body']?.asString()}');  }}

Use liveTable<T>(...) when SELECT * FROM table is enough:

DART
final taskRows = client.liveTable<Map<String, KalamCellValue>>('app.tasks'); await for (final rows in taskRows) {  print('tasks: ${rows.length}');}

If your live query uses a natural or composite key instead of a plain id column, pass keyColumns so reconciliation still stays inside the shared Rust core:

DART
final messages = client.live<Map<String, KalamCellValue>>(  "SELECT room_id, message_id, body FROM app.messages WHERE room_id = 'main'",  keyColumns: ['room_id', 'message_id'],);

You can also cap the in-memory materialized row set with limit, map rows directly into app models with mapRow, and observe checkpoint progress with onCheckpoint. For durable cached state, use Kalam Sync.

Materialized rows vs change events

  • liveEvents(...) gives you protocol-level events and incremental changes.
  • live(...) and liveTable(...) give you the fully reconciled current row list.
  • Materialized helpers are usually the better default for UI code.
  • Low-level events are better when you need custom reconciliation, audit logging, or protocol inspection.

Materialized-row limits

The SQL for materialized live rows still follows the same strict supported shape:

  • SELECT ... FROM ... WHERE ...
  • no ORDER BY
  • no LIMIT

The controls apply at different stages:

  • batchSize chunks the initial snapshot sent by the server.
  • lastRows selects how much history to rewind before live changes begin.
  • limit caps the materialized live row set the SDK keeps after startup.

Keep Dart-side presentation logic for sorting and grouping after rows arrive. Use limit only when the materialized live state itself should stay bounded.

Raw events (liveEvents)

DART
final stream = client.liveEvents(  "SELECT * FROM app.messages WHERE room = 'main'",); await for (final event in stream) {  switch (event) {    case AckEvent(:final subscriptionId, :final schema, :final totalRows):      print('ack $subscriptionId: $totalRows rows, columns=${schema.length}');    case InitialDataBatch(:final rows, :final hasMore):      print('snapshot batch: ${rows.length}, hasMore=$hasMore');    case InsertEvent(:final row):      print('insert: ${row['id']?.asInt()}');    case UpdateEvent(:final row, :final oldRow):      print('update: $oldRow -> $row');    case DeleteEvent(:final row):      print('delete: $row');    case SubscriptionError(:final code, :final message):      print('subscription error [$code]: $message');  }}

Snapshot vs live changes

Most subscriptions follow this pattern:

  1. AckEvent confirms the subscription and includes schema metadata.
  2. One or more InitialDataBatch events deliver the initial snapshot.
  3. InsertEvent / UpdateEvent / DeleteEvent deliver live changes.

If you only care about live changes, you can ignore InitialDataBatch after initial UI hydration.

Controlling the initial snapshot

batchSize

If the initial query returns many rows, the server can send it in batches. batchSize controls the maximum rows per snapshot batch.

DART
final stream = client.liveEvents(  'SELECT * FROM app.large_table',  batchSize: 100,);

You will receive one or more InitialDataBatch events with hasMore=true, followed by hasMore=false and then live changes.

lastRows

lastRows asks the server to “rewind” and include the last N rows before live changes begin. This is useful for chat timelines or activity feeds.

DART
final stream = client.liveEvents(  "SELECT * FROM chat.messages WHERE room_id = 'main'",  lastRows: 50,);

subscriptionId

Provide an explicit subscriptionId if you want stable IDs for client-side bookkeeping:

DART
final stream = client.liveEvents(  "SELECT * FROM app.messages WHERE room = 'main'",  subscriptionId: 'messages-feed',);

from

from accepts a VersionId checkpoint. Use it only when restoring the matching cached data for the same query and authenticated identity. A cursor alone does not reconstruct a complete list after an app restart; ordinary screens should start with a fresh snapshot and omit from.

DART
final stream = client.liveEvents(  "SELECT * FROM app.messages WHERE room = 'main'",  from: VersionId.parse('123'),);

For materialized rows, onCheckpoint exposes checkpoint.lastVersionId. This callback is not a transaction that atomically commits your app’s data and cursor. Use Kalam Sync for durable local state.

Replay is bounded. A delivery gap, expired cursor, or changed history can require resubscription or a fresh snapshot. Do not present retained rows as current until recovery completes. See the WebSocket recovery contract.

Listing active subscriptions

Use getSubscriptions() to inspect all subscriptions on the shared connection.

DART
final subs = await client.getSubscriptions();for (final sub in subs) {  print('${sub.id}: ${sub.query} (lastVersionId=${sub.lastVersionId}, closed=${sub.closed})');}

For server-side observability while debugging, query SELECT * FROM system.live. That view is node-local and reflects the subscriptions currently held in memory on the node you query. It is not persisted to RocksDB and it is not replicated through Raft.

Cancelling a subscription

The SDK returns a Stream<ChangeEvent>. Cancel by cancelling your StreamSubscription.

DART
final sub = client.liveEvents('SELECT * FROM app.messages').listen((_) {}); // laterawait sub.cancel();

Cancellation stays lightweight on the Dart side. The SDK cancels the Rust subscription handle on the shared connection, waits briefly for the pull loop to unblock, and then closes the stream. That avoids leaving a hanging Dart-side pull loop after a widget or provider disposes.

Data types

All subscription events expose rows as Map<String, KalamCellValue> for type-safe access:

  • InitialDataBatch.rows → List<Map<String, KalamCellValue>>
  • InsertEvent.row / InsertEvent.rows
  • UpdateEvent.columnIndexes / UpdateEvent.row / UpdateEvent.oldRow / UpdateEvent.oldRows
  • DeleteEvent.row / DeleteEvent.oldRows

UPDATE event behavior

For UpdateEvent, row and rows are sparse typed maps reconstructed from the Ack schema. They contain only the changed columns plus the primary key columns and _seq.

oldRow contains the previous values for that same sparse column set.

columnIndexes exposes the original positional update layout from the wire protocol. It lines up with the values sent on the Rust bridge raw event if you need exact sparse-column ordering.

DART
case UpdateEvent(:final row, :final oldRow):  // row is a sparse patch: only keys, changed columns, and metadata.  final id = row['id']?.asString();  final updateType = row['update_type']?.asString();  final targetId = row['target_id']?.asString();  print('Trigger: $updateType on $targetId');   // Do not replace a full cached row with this sparse map.  // oldRow carries the previous values for the sparse column set.  for (final key in oldRow.keys) {    if (!key.startsWith('_') && key != 'id') {      print('  changed: $key: ${oldRow[key]} → ${row[key]}');    }  }

Use the typed accessors to extract values:

DART
case InsertEvent(:final row):  final id    = row['id']?.asInt();  final name  = row['name']?.asString();  final score = row['score']?.asDouble();  final flag  = row['active']?.asBool();

These event accessors are already decoded from JSON row payloads; no manual List<List<dynamic>> parsing is needed.

See Cell Values for the full KalamCellValue API.

Next

SQL QueriesLive Query Examples