Token streaming with message-per-token is a pattern where every token generated by your model is published as an independent Ably message. Each token then appears as one message in the channel history. This uses Ably Pub/Sub for realtime communication between agents and clients.
This pattern is useful when clients only care about the most recent part of a response and you are happy to treat the channel history as a short sliding window rather than a full conversation log. For example:
- Backend-stored responses: The backend writes complete responses to a database and clients load those full responses from there, while Ably is used only to deliver live tokens for the current in-progress response.
- Live transcription, captioning, or translation: A viewer who joins a live stream only needs sufficient tokens for the current "frame" of subtitles, not the entire transcript so far.
- Code assistance in an editor: Streamed tokens become part of the file on disk as they are accepted, so past tokens do not need to be replayed from Ably.
- Autocomplete: A fresh response is streamed for each change a user makes to a document, with only the latest suggestion being relevant.
Publishing tokens
Publish tokens from a Realtime client, which maintains a persistent connection to the Ably service. This allows you to publish at very high message rates with the lowest possible latencies, while preserving guarantees around message delivery order. For more information, see Realtime and REST.
Channels separate message traffic into different topics. For token streaming, each conversation or session typically has its own channel.
Use the get() method to create or retrieve a channel instance:
1
const channel = realtime.channels.get('job-map-new');When publishing tokens, don't await the channel.publish() call. Ably rolls up acknowledgments and debounces them for efficiency, which means awaiting each publish would unnecessarily slow down your token stream. Messages are still published in the order that publish() is called, so delivery order is not affected.
1
2
3
4
5
6
7
8
9
10
11
12
13
// ✅ Do this - publish without await for maximum throughput
for await (const event of stream) {
if (event.type === 'token') {
channel.publish('token', event.text);
}
}
// ❌ Don't do this - awaiting each publish reduces throughput
for await (const event of stream) {
if (event.type === 'token') {
await channel.publish('token', event.text);
}
}This approach maximizes throughput while maintaining ordering guarantees, allowing you to stream tokens as fast as your AI model generates them.
Streaming patterns
Ably is a pub/sub messaging platform, so you can structure your messages however works best for your application. Below are common patterns for streaming tokens, each showing both agent-side publishing and client-side subscription. Choose the approach that fits your use case, or create your own variation.
Continuous token stream
For simple streaming scenarios such as live transcription, where all tokens are part of a continuous stream, simply publish each token as a message.
Publish tokens
1
2
3
4
5
6
7
8
const channel = realtime.channels.get('job-map-new');
// Example: stream returns events like { type: 'token', text: 'Hello' }
for await (const event of stream) {
if (event.type === 'token') {
channel.publish('token', event.text);
}
}Subscribe to tokens
1
2
3
4
5
6
7
const channel = realtime.channels.get('job-map-new');
// Subscribe to token messages
await channel.subscribe('token', (message) => {
const token = message.data;
console.log(token); // log each token as it arrives
});This pattern is simple and works well when you're displaying a single, continuous stream of tokens.
Token stream with multiple responses
For applications with multiple responses, such as chat conversations, include a responseId in message extras to correlate tokens together that belong to the same response.
Publish tokens
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
const channel = realtime.channels.get('job-map-new');
// Example: stream returns events like { type: 'token', text: 'Hello', responseId: 'resp_abc123' }
for await (const event of stream) {
if (event.type === 'token') {
channel.publish({
name: 'token',
data: event.text,
extras: {
headers: {
responseId: event.responseId
}
}
});
}
}Subscribe to tokens
Use the responseId header in message extras to correlate tokens. The responseId allows you to group tokens belonging to the same response and correctly handle token delivery for distinct responses, even when delivered concurrently.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
const channel = realtime.channels.get('job-map-new');
// Track responses by ID
const responses = new Map();
await channel.subscribe('token', (message) => {
const token = message.data;
const responseId = message.extras?.headers?.responseId;
if (!responseId) {
console.warn('Token missing responseId');
return;
}
// Create an empty response
if (!responses.has(responseId)) {
responses.set(responseId, '');
}
// Append token to response
responses.set(responseId, responses.get(responseId) + token);
});Token stream with explicit start/stop events
In some cases, your AI model response stream may include explicit events to mark response boundaries. You can indicate the event type, such as a response start/stop event, using the Ably message name.
Publish tokens
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
const channel = realtime.channels.get('job-map-new');
// Example: stream returns events like:
// { type: 'message_start', responseId: 'resp_abc123' }
// { type: 'message_delta', responseId: 'resp_abc123', text: 'Hello' }
// { type: 'message_stop', responseId: 'resp_abc123' }
for await (const event of stream) {
if (event.type === 'message_start') {
// Publish response start
channel.publish({
name: 'start',
extras: {
headers: {
responseId: event.responseId
}
}
});
} else if (event.type === 'message_delta') {
// Publish tokens
channel.publish({
name: 'token',
data: event.text,
extras: {
headers: {
responseId: event.responseId
}
}
});
} else if (event.type === 'message_stop') {
// Publish response stop
channel.publish({
name: 'stop',
extras: {
headers: {
responseId: event.responseId
}
}
});
}
}Subscribe to tokens
Handle each event type to manage response lifecycle:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
const channel = realtime.channels.get('job-map-new');
const responses = new Map();
// Handle response start
await channel.subscribe('start', (message) => {
const responseId = message.extras?.headers?.responseId;
responses.set(responseId, '');
});
// Handle tokens
await channel.subscribe('token', (message) => {
const responseId = message.extras?.headers?.responseId;
const token = message.data;
const currentText = responses.get(responseId) || '';
responses.set(responseId, currentText + token);
});
// Handle response stop
await channel.subscribe('stop', (message) => {
const responseId = message.extras?.headers?.responseId;
const finalText = responses.get(responseId);
console.log('Response complete:', finalText);
});Client hydration
When clients connect or reconnect, such as after a page refresh, they often need to catch up on tokens that were published while they were offline or before they joined. Ably provides several approaches to hydrate client state depending on your application's requirements.
Using rewind for recent history
The simplest approach is to use Ably's rewind channel option to attach to the channel at some point in the recent past, and automatically receive all tokens since that point:
1
2
3
4
5
6
7
8
9
10
11
12
13
// Use rewind to receive recent historical messages
const channel = realtime.channels.get('job-map-new', {
params: { rewind: '2m' } // or rewind: 100 for message count
});
// Subscribe to receive both recent historical and live messages,
// which are delivered in order to the subscription
await channel.subscribe('token', (message) => {
const token = message.data;
// Process tokens from both recent history and live stream
console.log('Token received:', token);
});Rewind supports two formats:
- Time-based: Use a time interval like
'30s'or'2m'to retrieve messages from that time period - Count-based: Use a number like
50or100to retrieve the most recent N messages (maximum 100)
By default, rewind is limited to the last 2 minutes of messages. This is usually sufficient for scenarios where clients need only recent context, such as for continuous token streaming, or when the response stream from a given model request does not exceed 2 minutes. If you need more than 2 minutes of history, see Using history for longer persistence.
Using history for older messages
For applications that need to retrieve tokens beyond the 2-minute rewind window, enable persistence on your channel. Use channel history with the untilAttach option to paginate back through history to obtain historical tokens, while preserving continuity with the delivery of live tokens:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// Use a channel in a namespace called 'persisted', which has persistence enabled
const channel = realtime.channels.get('persisted:job-map-new');
let response = '';
// Subscribe to live messages (implicitly attaches the channel)
await channel.subscribe('token', (message) => {
// Append the token to the end of the response
response += message.data;
});
// Fetch history up until the point of attachment
let page = await channel.history({ untilAttach: true });
// Paginate backwards through history
while (page) {
// Messages are newest-first, so prepend them to response
for (const message of page.items) {
response = message.data + response;
}
// Move to next page if available
page = page.hasNext() ? await page.next() : null;
}Hydrating an in-progress response
A common pattern is to persist complete model responses in your database while using Ably for live token delivery of the in-progress response.
The client loads completed responses from your database, then reaches back into Ably channel history until it encounters a token for a response it's already loaded.
You can retrieve partial history using either the rewind or history pattern.
Hydrate using rewind
Load completed responses from your database, then use rewind to catch up on any in-progress responses, skipping any tokens that belong to a response that was already loaded:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
// Load completed responses from database
const completedResponses = await loadResponsesFromDatabase();
// Use rewind to receive recent historical messages
const channel = realtime.channels.get('job-map-new', {
params: { rewind: '2m' }
});
// Track in progress responses by ID
const inProgressResponses = new Map();
// Subscribe to receive both recent historical and live messages,
// which are delivered in order to the subscription
await channel.subscribe('token', (message) => {
const token = message.data;
const responseId = message.extras?.headers?.responseId;
if (!responseId) {
console.warn('Token missing responseId');
return;
}
// Skip tokens for responses already hydrated from database
if (completedResponses.has(responseId)) {
return;
}
// Create an empty in-progress response
if (!inProgressResponses.has(responseId)) {
inProgressResponses.set(responseId, '');
}
// Append tokens for new responses
inProgressResponses.set(responseId, inProgressResponses.get(responseId) + token);
});Hydrate using history
Load completed responses from your database, then paginate backwards through history to catch up on in-progress responses until you reach a token that belongs to a response you've already loaded:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
// Load completed responses from database
const completedResponses = await loadResponsesFromDatabase();
// Use a channel in a namespace called 'persisted', which has persistence enabled
const channel = realtime.channels.get('persisted:job-map-new');
// Track in progress responses by ID
const inProgressResponses = new Map();
// Subscribe to live tokens (implicitly attaches)
await channel.subscribe('token', (message) => {
const token = message.data;
const responseId = message.extras?.headers?.responseId;
if (!responseId) {
console.warn('Token missing responseId');
return;
}
// Skip tokens for responses already hydrated from database
if (completedResponses.has(responseId)) {
return;
}
// Create an empty in-progress response
if (!inProgressResponses.has(responseId)) {
inProgressResponses.set(responseId, '');
}
// Append live tokens for in-progress responses
inProgressResponses.set(responseId, inProgressResponses.get(responseId) + token);
});
// Paginate backwards through history until we encounter a hydrated response
let page = await channel.history({ untilAttach: true });
// Paginate backwards through history
let done = false;
while (page && !done) {
// Messages are newest-first, so prepend them to response
for (const message of page.items) {
const token = message.data;
const responseId = message.extras?.headers?.responseId;
// Stop when we reach a response already loaded from database
if (completedResponses.has(responseId)) {
done = true;
break;
}
// Create an empty in-progress response
if (!inProgressResponses.has(responseId)) {
inProgressResponses.set(responseId, '');
}
// Prepend historical tokens for in-progress responses
inProgressResponses.set(responseId, token + inProgressResponses.get(responseId));
}
// Move to next page if available
page = page.hasNext() ? await page.next() : null;
}