Subscriptions & Live Queries

SUBSCRIBE TO <namespace>.<table_name>
[WHERE <condition>]
[OPTIONS (last_rows=<n>, batch_size=<n>, from_seq_id=<n>)];

-- Subscribe to all messages in a conversation
SUBSCRIBE TO chat.messages
WHERE conversation_id = 1
OPTIONS (last_rows=50);
 
-- Subscribe to typing events with batching
SUBSCRIBE TO chat.typing_events
WHERE conversation_id = 1
OPTIONS (last_rows=10, batch_size=5);
 
-- Resume from a specific point
SUBSCRIBE TO chat.messages
WHERE conversation_id = 1
OPTIONS (from_seq_id=1234);

KILL LIVE QUERY '<subscription_id>';

import { createClient, Auth } from 'kalam-link';
 
const client = createClient({
  url: 'http://localhost:8080',
  auth: Auth.basic('admin', 'admin')
});
await client.connect();
 
// Subscribe with SQL WHERE clause
const unsub = await client.subscribeWithSql(
  'SELECT * FROM chat.messages WHERE conversation_id = 1',
  (event) => {
    if (event.type === 'change') {
      console.log('New message:', event.rows);
    }
  },
  { batch_size: 50 }
);
 
// Simple table subscription
const unsubTyping = await client.subscribe(
  'chat.typing_events',
  (event) => {
    console.log('Typing:', event.change_type, event.rows);
  }
);
 
// Check active subscription count
console.log(`Active: ${client.getSubscriptionCount()}`);
 
// Cleanup
await unsub();
await unsubTyping();
await client.disconnect();

kalam> \subscribe SELECT * FROM chat.messages WHERE conversation_id = 1 OPTIONS (last_rows=20);