Skip to Content
RealtimeLive Query Examples

Dart and Flutter Live Query Examples

Use KalamClient.live() for a Flutter list that follows database changes. Each stream value is the complete current row list, with inserts, updates, and deletes already reconciled by the SDK.

Prepare the live-query example

Create app.live_messages using the shared inbox schema. Install kalam_link, initialize it once, and obtain an authenticated client using Dart setup. Production apps should use their signed-in user’s auth provider.

BASH
flutter pub add kalam_link

The widget below accepts that connected client. Mount LiveInbox(client: client) after connection succeeds, and show connection errors in your app’s sign-in/connection screen. The widget owns its subscription; your session service owns and disposes the client.

Drop in a Flutter inbox

DART
import 'package:flutter/material.dart';import 'package:kalam_link/kalam_link.dart'; class Message {  const Message({required this.id, required this.body});  final String id;  final String body;   factory Message.fromRow(Map<String, KalamCellValue> row) => Message(    id: row['id']!.asString()!,    body: row['body']!.asString()!,  );} class LiveInbox extends StatefulWidget {  const LiveInbox({super.key, required this.client});  final KalamClient client;   @override  State<LiveInbox> createState() => _LiveInboxState();} class _LiveInboxState extends State<LiveInbox> {  late Stream<List<Message>> _rows;  int _attempt = 0;   Stream<List<Message>> _open() => widget.client.live<Message>(    "SELECT id, room, body FROM app.live_messages WHERE room = 'main'",    keyColumns: ['id'],    mapRow: Message.fromRow,  );   @override  void initState() {    super.initState();    _rows = _open(); // Keep the stream stable across rebuilds.  }   @override  void didUpdateWidget(covariant LiveInbox oldWidget) {    super.didUpdateWidget(oldWidget);    if (oldWidget.client != widget.client) {      _attempt++;      _rows = _open(); // A new session must not reuse the previous user's rows.    }  }   void _retry() => setState(() {    _attempt++;    _rows = _open();  });   @override  Widget build(BuildContext context) => StreamBuilder<List<Message>>(    key: ValueKey(_attempt), // Fresh loading state on explicit retry/session change.    stream: _rows,    builder: (context, snapshot) {      if (snapshot.hasError) {        final error = snapshot.error;        final message = error is SubscriptionError            ? '${error.code}: ${error.message}'            : error.toString();        return Column(          mainAxisSize: MainAxisSize.min,          children: [            Text('Could not update messages: $message'),            TextButton(onPressed: _retry, child: const Text('Retry')),          ],        );      }      if (snapshot.connectionState == ConnectionState.done) {        return Column(          mainAxisSize: MainAxisSize.min,          children: [            const Text('The live query ended.'),            TextButton(onPressed: _retry, child: const Text('Reconnect')),          ],        );      }      if (!snapshot.hasData) {        return const Center(child: CircularProgressIndicator());      }      final rows = snapshot.data!;      if (rows.isEmpty) return const Text('No messages yet.');      return ListView.builder(        itemCount: rows.length,        itemBuilder: (context, index) => ListTile(          key: ValueKey(rows[index].id),          title: Text(rows[index].body),        ),      );    },  );}

Place the widget in a bounded area such as Scaffold.body or Expanded. StreamBuilder cancels its subscription when removed and when its stream is replaced. Do not construct client.live() inside build(): an unrelated rebuild would otherwise create another stream.

The error branch comes before the empty branch, so a rejected query cannot look like an empty inbox. live() converts server subscription errors into stream errors. In contrast, low-level liveEvents() exposes SubscriptionError as an event; handle it in your event switch as well as handling stream failures.

Verify the screen

Run the insert, update, and delete statements from another connection with the same user. Expect the new body, then the edited body, then removal without duplicates. This USER table is private to that identity.

Temporarily use a nonexistent table to verify the error screen. Fix the name and restart the example. For a permissions or connection problem that has been corrected without changing code, use Retry. Navigate away and confirm the subscription disappears from client.getSubscriptions() after cancellation has completed.

ConnectionState.active describes the Dart stream, not socket health. A reconnecting stream can remain active with cached rows. Use your application’s connection state for an offline banner, and verify recovery with a normal SELECT before considering previously displayed rows current.

Use the stream without Flutter

In a service, keep the StreamSubscription and cancel it explicitly:

DART
final subscription = client.live<Message>(  "SELECT id, room, body FROM app.live_messages WHERE room = 'main'",  mapRow: Message.fromRow,  keyColumns: ['id'],).listen(  (rows) => print('Current messages: ${rows.length}'),  onError: (Object error, StackTrace stack) {    print('Live query failed: $error');  },  onDone: () => print('Live query ended'),); // In your service's async shutdown method:await subscription.cancel();// Only the session owner calls: await client.dispose();

Retry and persistence rules

  • Keep the primary key in the SELECT. mapRow converts cell wrappers into app models; keyColumns defines row identity for reconciliation.
  • Raw subscription SQL is SELECT ... FROM ... WHERE ..., without ORDER BY or LIMIT. Sort a copy for display. batchSize, lastRows, and the direct materializer’s limit have different purposes.
  • Let the SDK manage transient transport reconnects. Explicit Retry replaces this widget’s stream; do not add a timer that continually opens subscriptions.
  • Correct authentication, authorization, or SQL failures before retrying. For writes, check both thrown exceptions and QueryResponse.success; see Dart querying.
  • These rows live in memory. Do not persist only a checkpoint and resume into an empty list after app restart. Use Kalam Sync for a durable local cache, and Offline Actions for queued writes.
  • A gap or expired cursor requires recovery within the bounded replay contract. A live stream is not an unlimited event archive.
Live SubscriptionsOffline Flutter: kalam_sync