From d87923144ac493f415fb7e4bf8af7d24d4f72150 Mon Sep 17 00:00:00 2001
From: owenpearson <owen.pearson@ably.com>
Date: Thu, 12 Mar 2026 10:52:30 +0000
Subject: [PATCH] ait: add react token streaming examples

---
 src/data/languages/languageData.ts            |   1 +
 .../token-streaming/message-per-response.mdx  | 168 +++++++++++++++++
 .../token-streaming/message-per-token.mdx     | 178 ++++++++++++++++++
 3 files changed, 347 insertions(+)

diff --git a/src/data/languages/languageData.ts b/src/data/languages/languageData.ts
index 10710cf4a0..dda7e90139 100644
--- a/src/data/languages/languageData.ts
+++ b/src/data/languages/languageData.ts
@@ -44,6 +44,7 @@ export default {
   },
   aiTransport: {
     javascript: '2.19',
+    react: '2.19',
     java: '1.6',
     python: '3.1',
     swift: '1.2',
diff --git a/src/pages/docs/ai-transport/token-streaming/message-per-response.mdx b/src/pages/docs/ai-transport/token-streaming/message-per-response.mdx
index b8e6ea6026..d41f168b84 100644
--- a/src/pages/docs/ai-transport/token-streaming/message-per-response.mdx
+++ b/src/pages/docs/ai-transport/token-streaming/message-per-response.mdx
@@ -449,6 +449,31 @@ channel.subscribe(message -> {
     }
 });
 ```
+```react
+const [responses, setResponses] = useState(new Map());
+
+// Subscribe to live messages
+useChannel('ai:{{RANDOM_CHANNEL_NAME}}', (message) => {
+  setResponses((prev) => {
+    const next = new Map(prev);
+    switch (message.action) {
+      case 'message.create':
+        // New response started
+        next.set(message.serial, message.data);
+        break;
+      case 'message.append':
+        // Append token to existing response
+        next.set(message.serial, (next.get(message.serial) || '') + message.data);
+        break;
+      case 'message.update':
+        // Replace entire response content
+        next.set(message.serial, message.data);
+        break;
+    }
+    return next;
+  });
+});
+```
 </Code>
 
 ## Client hydration <a id="hydration"/>
@@ -549,6 +574,31 @@ channel.subscribe(message -> {
     }
 });
 ```
+```react
+// Set rewind via ChannelProvider options={{ params: { rewind: '2m' } }}
+
+const [responses, setResponses] = useState(new Map());
+
+// Receive both recent historical (via rewind) and live messages
+useChannel('ai:{{RANDOM_CHANNEL_NAME}}', (message) => {
+  setResponses((prev) => {
+    const next = new Map(prev);
+    switch (message.action) {
+      case 'message.create':
+        next.set(message.serial, message.data);
+        break;
+      case 'message.append':
+        const current = next.get(message.serial) || '';
+        next.set(message.serial, current + message.data);
+        break;
+      case 'message.update':
+        next.set(message.serial, message.data);
+        break;
+    }
+    return next;
+  });
+});
+```
 </Code>
 
 Rewind supports two formats:
@@ -678,6 +728,46 @@ while (page != null) {
     page = page.hasNext() ? page.next() : null;
 }
 ```
+```react
+const [responses, setResponses] = useState(new Map());
+const hydrated = useRef(false);
+
+// Subscribe to live messages and get the history function
+const { history } = useChannel('ai:{{RANDOM_CHANNEL_NAME}}', (message) => {
+  setResponses((prev) => {
+    const next = new Map(prev);
+    switch (message.action) {
+      case 'message.create':
+        next.set(message.serial, message.data);
+        break;
+      case 'message.append':
+        next.set(message.serial, (next.get(message.serial) || '') + message.data);
+        break;
+      case 'message.update':
+        next.set(message.serial, message.data);
+        break;
+    }
+    return next;
+  });
+});
+
+// Fetch history on mount
+useEffect(() => {
+  if (hydrated.current) return;
+  hydrated.current = true;
+
+  (async () => {
+    let page = await history({ untilAttach: true });
+    while (page) {
+      for (const message of page.items) {
+        // message.data contains the full concatenated text
+        setResponses((prev) => new Map(prev).set(message.serial, message.data));
+      }
+      page = page.hasNext() ? await page.next() : null;
+    }
+  })();
+}, [history]);
+```
 </Code>
 
 ### Hydrating an in-progress response <a id="in-progress-response"/>
@@ -911,6 +1001,37 @@ channel.subscribe(message -> {
     }
 });
 ```
+```react
+// Set rewind via ChannelProvider options={{ params: { rewind: '2m' } }}
+
+const [inProgressResponses, setInProgressResponses] = useState(new Map());
+
+// Receive both recent historical and live messages
+useChannel('ai:responses', (message) => {
+  const responseId = message.extras?.headers?.responseId;
+
+  if (!responseId) return;
+
+  // Skip messages for responses already loaded from database
+  if (completedResponses.has(responseId)) return;
+
+  setInProgressResponses((prev) => {
+    const next = new Map(prev);
+    switch (message.action) {
+      case 'message.create':
+        next.set(responseId, message.data);
+        break;
+      case 'message.append':
+        next.set(responseId, (next.get(responseId) || '') + message.data);
+        break;
+      case 'message.update':
+        next.set(responseId, message.data);
+        break;
+    }
+    return next;
+  });
+});
+```
 </Code>
 
 <Aside data-type="note">
@@ -1118,6 +1239,53 @@ while (page != null) {
     page = page.hasNext() ? page.next() : null;
 }
 ```
+```react
+const [inProgressResponses, setInProgressResponses] = useState(new Map());
+const hydrated = useRef(false);
+
+// Subscribe to live messages and get the history function
+const { history } = useChannel('ai:{{RANDOM_CHANNEL_NAME}}', (message) => {
+  const responseId = message.extras?.headers?.responseId;
+
+  if (!responseId) return;
+  if (completedResponses.has(responseId)) return;
+
+  setInProgressResponses((prev) => {
+    const next = new Map(prev);
+    switch (message.action) {
+      case 'message.create':
+        next.set(responseId, message.data);
+        break;
+      case 'message.append':
+        next.set(responseId, (next.get(responseId) || '') + message.data);
+        break;
+      case 'message.update':
+        next.set(responseId, message.data);
+        break;
+    }
+    return next;
+  });
+});
+
+// Fetch history from the last completed response until attachment
+useEffect(() => {
+  if (hydrated.current) return;
+  hydrated.current = true;
+
+  (async () => {
+    let page = await history({ untilAttach: true, start: latestTimestamp });
+    while (page) {
+      for (const message of page.items) {
+        const responseId = message.extras?.headers?.responseId;
+        if (responseId) {
+          setInProgressResponses((prev) => new Map(prev).set(responseId, message.data));
+        }
+      }
+      page = page.hasNext() ? await page.next() : null;
+    }
+  })();
+}, [history]);
+```
 </Code>
 
 <Aside data-type="note">
diff --git a/src/pages/docs/ai-transport/token-streaming/message-per-token.mdx b/src/pages/docs/ai-transport/token-streaming/message-per-token.mdx
index 4867624f35..10af6bccfc 100644
--- a/src/pages/docs/ai-transport/token-streaming/message-per-token.mdx
+++ b/src/pages/docs/ai-transport/token-streaming/message-per-token.mdx
@@ -228,6 +228,20 @@ channel.subscribe("response-complete", message -> {
     response.append((String) message.data);
 });
 ```
+
+```react
+const [response, setResponse] = useState('');
+
+// Handle individual tokens and recovery messages
+useChannel('{{RANDOM_CHANNEL_NAME}}', (message) => {
+  if (message.name === 'token') {
+    setResponse((prev) => prev + message.data);
+  } else if (message.name === 'response-complete') {
+    // Recovery message replaces all previous tokens
+    setResponse(message.data);
+  }
+});
+```
 </Code>
 
 When streaming multiple concurrent responses, include a `responseId` in message [`extras`](/docs/messages#properties) so subscribers can correctly associate the recovery message with the tokens it replaces. See [token stream with multiple responses](#multiple-responses) for details on correlating messages.
@@ -312,6 +326,16 @@ channel.subscribe("token", message -> {
     System.out.println(token); // log each token as it arrives
 });
 ```
+
+```react
+const [response, setResponse] = useState('');
+
+// Subscribe to token messages
+useChannel('{{RANDOM_CHANNEL_NAME}}', 'token', (message) => {
+  const token = message.data;
+  setResponse((prev) => prev + token);
+});
+```
 </Code>
 
 This pattern is simple and works well when you're displaying a single, continuous stream of tokens.
@@ -449,6 +473,24 @@ channel.subscribe("token", message -> {
     responses.put(responseId, responses.get(responseId) + token);
 });
 ```
+
+```react
+const [responses, setResponses] = useState(new Map());
+
+// Track responses by ID
+useChannel('{{RANDOM_CHANNEL_NAME}}', 'token', (message) => {
+  const token = message.data;
+  const responseId = message.extras?.headers?.responseId;
+
+  if (!responseId) return;
+
+  setResponses((prev) => {
+    const next = new Map(prev);
+    next.set(responseId, (next.get(responseId) || '') + token);
+    return next;
+  });
+});
+```
 </Code>
 
 ### Token stream with explicit start/stop events <a id="explicit-events"/>
@@ -662,6 +704,27 @@ channel.subscribe("stop", message -> {
     System.out.println("Response complete: " + finalText);
 });
 ```
+
+```react
+const [responses, setResponses] = useState(new Map());
+
+// Handle response lifecycle events
+useChannel('{{RANDOM_CHANNEL_NAME}}', (message) => {
+  const responseId = message.extras?.headers?.responseId;
+
+  if (message.name === 'start') {
+    setResponses((prev) => new Map(prev).set(responseId, ''));
+  } else if (message.name === 'token') {
+    setResponses((prev) => {
+      const next = new Map(prev);
+      next.set(responseId, (next.get(responseId) || '') + message.data);
+      return next;
+    });
+  } else if (message.name === 'stop') {
+    console.log('Response complete:', responses.get(responseId));
+  }
+});
+```
 </Code>
 
 ## Client hydration <a id="hydration"/>
@@ -725,6 +788,18 @@ channel.subscribe("token", message -> {
     System.out.println("Token received: " + token);
 });
 ```
+
+```react
+// Set rewind via ChannelProvider options={{ params: { rewind: '2m' } }}
+
+const [response, setResponse] = useState('');
+
+// Receive both recent historical and live messages
+useChannel('{{RANDOM_CHANNEL_NAME}}', 'token', (message) => {
+  const token = message.data;
+  setResponse((prev) => prev + token);
+});
+```
 </Code>
 
 Rewind supports two formats:
@@ -821,6 +896,35 @@ while (page != null) {
     page = page.hasNext() ? page.next() : null;
 }
 ```
+
+```react
+const [response, setResponse] = useState('');
+const hydrated = useRef(false);
+
+// Subscribe to live messages and get the history function
+const { history } = useChannel('persisted:{{RANDOM_CHANNEL_NAME}}', 'token', (message) => {
+  setResponse((prev) => prev + message.data);
+});
+
+// Fetch history on mount
+useEffect(() => {
+  if (hydrated.current) return;
+  hydrated.current = true;
+
+  (async () => {
+    let historicalTokens = '';
+    let page = await history({ untilAttach: true });
+    while (page) {
+      // Messages are newest-first, so prepend them to response
+      for (const message of page.items) {
+        historicalTokens = message.data + historicalTokens;
+      }
+      page = page.hasNext() ? await page.next() : null;
+    }
+    setResponse((prev) => historicalTokens + prev);
+  })();
+}, [history]);
+```
 </Code>
 
 ### Hydrating an in-progress response <a id="in-progress-response"/>
@@ -942,6 +1046,29 @@ channel.subscribe("token", message -> {
     inProgressResponses.put(responseId, inProgressResponses.get(responseId) + token);
 });
 ```
+
+```react
+// Set rewind via ChannelProvider options={{ params: { rewind: '2m' } }}
+
+const [inProgressResponses, setInProgressResponses] = useState(new Map());
+
+// Receive both recent historical and live messages
+useChannel('{{RANDOM_CHANNEL_NAME}}', 'token', (message) => {
+  const token = message.data;
+  const responseId = message.extras?.headers?.responseId;
+
+  if (!responseId) return;
+
+  // Skip tokens for responses already hydrated from database
+  if (completedResponses.has(responseId)) return;
+
+  setInProgressResponses((prev) => {
+    const next = new Map(prev);
+    next.set(responseId, (next.get(responseId) || '') + token);
+    return next;
+  });
+});
+```
 </Code>
 
 #### Hydrate using history
@@ -1133,4 +1260,55 @@ while (page != null && !done) {
     page = page.hasNext() ? page.next() : null;
 }
 ```
+
+```react
+const [inProgressResponses, setInProgressResponses] = useState(new Map());
+const hydrated = useRef(false);
+
+// Subscribe to live messages and get the history function
+const { history } = useChannel('persisted:{{RANDOM_CHANNEL_NAME}}', 'token', (message) => {
+  const token = message.data;
+  const responseId = message.extras?.headers?.responseId;
+
+  if (!responseId) return;
+  if (completedResponses.has(responseId)) return;
+
+  setInProgressResponses((prev) => {
+    const next = new Map(prev);
+    next.set(responseId, (next.get(responseId) || '') + token);
+    return next;
+  });
+});
+
+// Fetch history on mount
+useEffect(() => {
+  if (hydrated.current) return;
+  hydrated.current = true;
+
+  (async () => {
+    let page = await history({ untilAttach: true });
+    let done = false;
+    while (page && !done) {
+      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;
+        }
+
+        // Prepend historical tokens (newest-first order)
+        setInProgressResponses((prev) => {
+          const next = new Map(prev);
+          next.set(responseId, token + (next.get(responseId) || ''));
+          return next;
+        });
+      }
+      page = page.hasNext() ? await page.next() : null;
+    }
+  })();
+}, [history, completedResponses]);
+```
 </Code>