references/datastreams-basics.md
# Data Streams Basics
Real-time publish/subscribe: a server publishes to a channel, and every subscriber connected over WebSocket receives the message. Everything here was verified against live calls.
## How the pieces fit
```
Console: create channel + choose access control (User Roles or Custom Logic)
Server (Catalyst function):
publishData(channelId, data) (admin) -> every connected subscriber
getTokenPair(channelId, { userId | connectionName }) -> token pair for one client
Client (browser, Node, Java app):
get token pair from your function -> open WebSocket -> on 'open' subscribe(type)
-> on 'message' process event, then sendAck()
```
- **Channels are created only in the console** (Cloud Scale → Data Streams → Create Channel). There is no SDK method, REST endpoint, or MCP tool to create, edit, or delete a channel. The agent cannot see the console: ask the user for the channel ID, or list channels from a function.
- **Admin methods run on the server only.** `publishData`, `getAllChannels`, `getChannelDetails` and `getLiveCount` are server-side (admin) methods: call them in a Catalyst function, never in browser or mobile code.
- **`getTokenPair` is available in server-side and in the browser SDK.** On a **Custom Logic** channel the token must still come from the configured function (see below), so issue it there and return it to the client.
- **Subscribers** can run anywhere that opens a WebSocket: a browser, a Node service, a Java app.
## Channel access control
Chosen when the channel is created. It decides which identifier a token pair must use.
| Access control | `control_type` | Token identifier | Where the token must be generated |
|---|---|---|---|
| **User Roles** (pick roles, e.g. App Administrator, App User) | `0` | `userId`: an **app user's** ID from Catalyst Authentication | Any Catalyst function, or the REST API |
| **Custom Logic** (pick an Advanced I/O function) | `1` | `connectionName`: any string you choose per client | **Only inside the configured function** |
What happens with the wrong combination:
| Request | Result |
|---|---|
| User Roles channel + `connectionName` | `400 MISSING_PARAM` "Mandatory parameter app_user_id is missing." |
| User Roles channel + a console admin's user ID (not an app user) | `400 INVALID_INPUT` "Invalid input value for app_user_id" |
| User Roles channel + an app user who was invited but has not confirmed | Works |
| Custom Logic channel + `connectionName` from any function other than the configured one, or from REST | `400 DATASTREAMS_ERROR` "Invalid access control source! Kindly request from the configured one!" |
| Custom Logic channel + `userId` (from any function, or REST) | Token is issued, but the WebSocket rejects it: error `3000` "Authentication failed, please generate valid credentials…" |
**Custom Logic, as it really works:** the function you select in the console is the channel's token endpoint. Your client calls it; it runs whatever check you need (session, SSO, device ID) and then calls `getTokenPair(channelId, { connectionName })` itself. Catalyst does not call the function at token time or at connect time, and nothing reads an `{ approved: true }` response. Find users' IDs for User Roles channels in Console → Authentication, or with the Catalyst MCP tool that lists project users.
## Token pair
Response of `getTokenPair` (SDK) or `POST …/tokenpair` (REST). Keys are **hyphenated**:
```json
{
"wss-id": "sub_<appUserId>_<orgId>_<projectId>_<channelId>",
"channel-id": "<channelId>",
"key": "<352-character key>",
"url": "<data-center streaming host, e.g. us4-dms.zoho.com>",
"deployment": "<data-center deployment, e.g. us4>"
}
```
- Pass `wss-id` as `zuid` when creating the WebSocket client, with `url` and `key`. `token.wss_id` and `token.channel_id` are `undefined`.
- `url` is a bare host name for the user's data center. Use the value the token returns; do not hardcode a host.
- For Custom Logic tokens, `wss-id` starts with the connection name (`<connectionName>_subua_…`).
- Each token is scoped to one channel.
## Subscribe types
Pass to `subscribe(type)` after the WebSocket `open` event.
| Type | Behaviour (verified) |
|---|---|
| `'0'` | Live only: messages published after subscribing |
| `'-1'` | Earliest: replays every message still within the 48-hour retention, then live |
| `'-2'` | Resume the previous session. Clean only when the session missed nothing; see the warning below |
| `'<streamingId>'` | Resume from a specific message. Same warning as `'-2'` |
> **Don't rely on resume after a drop (verified live).** When a session that missed messages resubscribes with `'-2'` or a `streamingId`, the server delivers each missed message **twice** and also **replays the channel's whole 48-hour retention**, interleaved. With nothing missed, resume is clean. Reconnect the way that works instead: see [Reconnecting](#reconnecting).
## Acknowledgement
Delivery is sequential. After handling a message the subscriber must call `sendAck()`. Without it, the server holds every later message: with three messages published and no ack, only the first arrives.
- Handle the message, persist anything you need, **then** `sendAck()`. A crash before the ack leaves the message unacknowledged instead of lost.
- Each subscriber receives the next message about one network round trip after its ack (about 300 ms from India to a US data center). A burst of 20 messages takes a few seconds to drain per subscriber, so publish state snapshots (for example, totals once a second) rather than one message per tiny change.
## Reconnecting
What works in live tests, for both access-control types:
1. **Keep your data in your own store** (Data Store, your API). Treat the channel as "something changed", not as the record.
2. **Save the current token** while connected (for example, in local storage or the Keychain).
3. **Before reconnecting, release the old session:** open a socket with the *old* token, call `unsubscribe()` on `open` without subscribing, wait about a second, then close it. The release socket may still be sent a message that was published while you were disconnected: **ignore it and don't ack it** (step 4 reloads state anyway). The new session then starts clean, with no duplicates or retention replay.
4. **Connect with a fresh token and `subscribe('0')`.** Buffer live messages, load the current state from your store, then apply the buffer.
5. **Make updates idempotent:** carry a version or timestamp in each message and ignore anything older than what you hold.
6. **Unsubscribe before every deliberate close** (page unload, sign-out, idle). Sessions closed without `unsubscribe()` stay on the channel, and in tests many of them slowed `publishData` from ~50 ms to 130–380 ms, with spikes of several seconds.
Why step 3 matters: on a **User Roles** channel, one user has one live session per channel. After a drop without `unsubscribe()`, a fresh token for the same user connected and then kept closing every 2–4 s with no live messages, until the old session was released. On **Custom Logic** channels a `connectionName` also has one live session; after a clean `close()`, a second token for the same name fails with `DUPLICATE_VALUE`, so use a new name per connection (for example `kds-grill-4f2a`) or release the old one first. Connection names are at most **20 characters**, and `.` or `:` are rejected (with the misleading error "Either userId or connectionName must be provided").
## Messages
- Fields: `data` and `streamingId`; API-operation (bulk) messages also carry `url` and `method`. There is no `operation` field.
- `data` is **always a string**. The SDKs' `publishData(channelId, data)` requires a string, so `JSON.stringify` objects and `JSON.parse` on receipt. The REST API also accepts a JSON object as `data`; subscribers receive it as a JSON string.
- Messages arrive in publish order, typically within about 60 ms, and can reach a subscriber before the publisher's call returns.
- The Java message class also has `getUrl()` / `getMethod()` for bulk messages. Check `data` first; only use `url` when `data` is null.
## Retention and live count
- Messages are kept for **48 hours** per channel, then dropped.
- `getLiveCount(channelId)` returns the number of connected subscribers. The REST path accepts the channel name or ID.
- The count lags: it can still include a subscriber for a few seconds after it unsubscribes and closes. Allow for that before asserting 0 in a test.
## REST API
Base: `https://api.catalyst.zoho.com/baas/v1/project/{project_id}/datastreams` (use the user's data-center API domain).
Headers: `Authorization: Zoho-oauthtoken <token>`; `CATALYST-ORG: <org_id>` and `Environment: Development|Production` are optional.
| Operation | Method and path | Body | Scope |
|---|---|---|---|
| List channels | `GET /channel` | | `ZohoCatalyst.datastreams.channel.READ` |
| Channel details | `GET /channel/{channel_id}` | | `ZohoCatalyst.datastreams.channel.READ` |
| Live count | `GET /channel/{channel_name_or_id}/liveclient` | | `ZohoCatalyst.datastreams.connections.READ` |
| Token pair | `POST /channel/{channel_id}/tokenpair` | `{"app_user_id": <id>}` | `ZohoCatalyst.datastreams.connections.token.CREATE` |
| Publish | `POST /channel/{channel_id}/stream` | `{"data": <string or object>}` | `ZohoCatalyst.datastreams.channel.stream.CREATE` |
Responses are wrapped: `{"status": "success", "data": …}`. Publish returns `"data": true`. Live count returns `"data": {"connections": <n>}`. A REST token request can't use `connection_name` on a Custom Logic channel, because the request isn't coming from the configured function.
Channel object (from list and details): `id`, `name`, `control_type` (0 User Roles, 1 Custom Logic), `control_ids` (role IDs, or the function ID), `retention` (48), `project_id`, `created_by`, `modified_by`, timestamps.
## Common Errors
| Error | Cause | Fix |
|-------|-------|-----|
| `Mandatory parameter app_user_id is missing.` | `connectionName` used on a User Roles channel | Use an app user's `userId`, or create a Custom Logic channel for anonymous or device clients |
| `Invalid input value for app_user_id` | The ID is a console admin or not an app user of this project | Use an app user's ID from Console → Authentication (invited users work before they confirm) |
| `Invalid access control source! Kindly request from the configured one!` | Token for a Custom Logic channel requested outside its configured function, or via REST | Generate the token inside the function selected as the channel's access control |
| WebSocket error `3000` "Authentication failed…" | User-ID token on a Custom Logic channel, or an expired or wrong token | On Custom Logic channels use `connectionName` from the configured function; otherwise request a fresh token |
| `No Channel exist with the given identifer!` | Wrong channel ID, or the channel is in another project or environment | List channels and use the `id` field |
| `Invalid input value for Env Name` | `Environment: Production` sent for a project without a Production environment | Omit the header, or deploy the project to Production first |
| Subscriber gets the first message only | `sendAck()` not called | Call `sendAck()` after processing each message |
| `zuid` undefined / connection never opens | Read `token.wss_id` | Use `token['wss-id']` |
| Duplicates and old messages after reconnecting | Resubscribed a session that missed messages with `'-2'` or a `streamingId` | Release the old session, reconnect with a fresh token and `'0'`, reload state from your store ([Reconnecting](#reconnecting)) |
| Fresh token connects, then closes every few seconds with no messages | Old session for the same user or connection name still open | Release it first: open the old token and `unsubscribe()` |
## WebSocket error codes
| Code | Meaning | What to do |
|---|---|---|
| `3000` | Authentication failed | Generate a new token pair |
| `1000` | Session closing. Fired right after **your own `unsubscribe()`** ("Current session will be closed. New connection to be established."), in every SDK; otherwise the session expired | After your own `unsubscribe()`: expected, don't reconnect. Otherwise: reconnect with a new token pair |
| `1013` | Connection blocked | Contact support |
| `1011` | Server down | Retry after a minute |
| `1014` | Missing key parameter | Pass `token.key` when creating the client |
| `1006` | Failed to create the WebSocket connection | Retry; check the host and network |
| `1007` | Invalid host received in a switch-URL message | Reconnect with a fresh token pair |
references/datastreams-java-basics.md
# Data Streams with Java
Data Streams is part of the main Catalyst Java SDK (`com.zc.component.datastream`). Server calls run in a Catalyst Java function; subscribers can run anywhere.
## Dependency
There is no `datastreams` artifact. Add the Java SDK from Zoho's Maven repository:
```xml
<repositories>
<repository>
<id>java-sdk</id>
<url>https://maven.zohodl.com</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>com.zoho.catalyst</groupId>
<artifactId>java-sdk</artifactId>
<version>2.5.0</version>
</dependency>
</dependencies>
```
Version 2.5.0 is the first with Data Streams. It brings `function-sdk`, `Java-WebSocket`, `okhttp` and Jackson. For a CLI-deployed function, put the resolved jars in the function's `lib/` folder (`mvn dependency:copy-dependencies -DoutputDirectory=lib`): the CLI compiles the `.java` source against `lib/*` on deploy. Use stack `java17` (also `java8`, `java11`, `java21`). The copied jars include test and servlet jars (`junit`, `hamcrest`, `javax.servlet-api`); deploys work with them in place.
Function config (`functions/stream_api/catalyst-config.json`); `main` is the handler class name:
```json
{
"deployment": { "name": "stream_api", "stack": "java17", "type": "advancedio", "env_variables": {} },
"execution": { "main": "StreamApi" }
}
```
Project `catalyst.json`: `{ "functions": { "targets": ["stream_api"], "source": "functions" } }`, then `catalyst deploy --only functions:stream_api`. The function is reached at `https://<project-domain>/server/stream_api/<route>`; route on `req.getPathInfo()`.
## Server calls in a function
Inside a Catalyst function, `ZCDatastream.getInstance()` works with no extra setup.
```java
import com.catalyst.advanced.CatalystAdvancedIOHandler;
import com.zc.component.datastream.ZCDatastream;
import com.zc.component.datastream.beans.ZCChannelDetails;
import com.zc.component.datastream.beans.ZCChannelLiveCountResponse;
import com.zc.component.datastream.beans.ZCTokenResponse;
import org.json.JSONObject;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.List;
import java.util.stream.Collectors;
public class StreamApi implements CatalystAdvancedIOHandler {
@Override
public void runner(HttpServletRequest req, HttpServletResponse res) throws Exception {
ZCDatastream ds = ZCDatastream.getInstance();
JSONObject body = new JSONObject(req.getReader().lines().collect(Collectors.joining()));
String channelId = body.getString("channelId");
// Publish (data is a String)
Boolean ok = ds.publishData(channelId, new JSONObject().put("status", "shipped").toString());
// Token pair: Long userId for User Roles channels;
// String connectionName only inside a Custom Logic channel's configured function
ZCTokenResponse t = ds.getTokenPair(channelId, Long.valueOf(body.getString("userId")));
// Reads
List<ZCChannelDetails> channels = ds.getAllChannels();
ZCChannelDetails channel = ds.getChannelDetails(channelId);
ZCChannelLiveCountResponse live = ds.getLiveCount(channelId);
res.setContentType("application/json");
res.getWriter().write(new JSONObject()
.put("published", ok)
.put("live", live.getLiveConnections())
.put("token", new JSONObject()
.put("url", t.getUrl())
.put("wssId", t.getWssId())
.put("key", t.getKey())
.put("channelId", String.valueOf(t.getChannelId())))
.toString());
}
}
```
`ZCTokenResponse` has `getUrl()`, `getWssId()`, `getKey()` and `getChannelId()` (a `Long`). There is no deployment getter.
`ZCChannelDetails` has `getId()`, `getName()`, `getControlType()` (0 User Roles, 1 Custom Logic), `getControlIds()`, `getProjectId()`, `getParentId()`, `getService()`, `getSource()`, `getCreatedBy()` / `getModifiedBy()` (`ZCUserInfo`). Unlike the REST and JavaScript channel object, there is no retention field.
## Subscriber
`subscribe()` and `unsubscribe()` throw the checked `ZCDataStreamException`, so wrap them; `onOpen` cannot throw it.
```java
import com.zc.component.datastream.beans.ZCCustomEvent;
import com.zc.component.datastream.beans.ZCDataStreamMessageEvent;
import com.zc.component.datastream.beans.ZCDataStreamsConfig;
import com.zc.component.datastream.websocket.ZCDataStreamsEventListener;
import com.zc.component.datastream.websocket.ZCDataStreamsWebSocket;
import com.zc.exception.ZCDataStreamException;
public class Subscriber {
public static void main(String[] args) throws Exception { // ZCDataStreamException from unsubscribe() propagates here
// url, wssId, key come from your token endpoint
String url = args[0], wssId = args[1], key = args[2];
ZCDataStreamsConfig config = new ZCDataStreamsConfig(url, wssId, key, false); // last argument: loggingEnabled
// Outside a Catalyst function, construct the client directly
final ZCDataStreamsWebSocket socket = new ZCDataStreamsWebSocket(config);
socket.addEventListener(new ZCDataStreamsEventListener() {
@Override public void onOpen(ZCCustomEvent event) {
try {
socket.subscribe("0"); // "-1" earliest, "-2" resume
} catch (ZCDataStreamException e) {
System.err.println("Subscribe failed: " + e.getMessage());
}
}
@Override public void onMessage(ZCDataStreamMessageEvent event) {
if (event.getData() != null) {
System.out.println(event.getStreamingId() + " " + event.getData());
} else if (event.getUrl() != null) {
System.out.println("Bulk data at " + event.getMethod() + " " + event.getUrl());
}
socket.sendAck(); // required for the next message
}
@Override public void onClose(ZCCustomEvent event) { System.out.println("Closed: " + event.getMessage()); }
@Override public void onError(ZCCustomEvent event) { System.err.println("Error: " + event.getMessage()); }
@Override public void onPong(ZCCustomEvent event) { }
});
Thread.sleep(60_000);
socket.unsubscribe(); // free the session before closing; fires onError "Current session will be closed…", which is expected
Thread.sleep(1_000);
socket.close();
}
}
```
Inside a Catalyst function you can also use `ZCDatastream.getInstance().createWebSocketClient(config)`. Outside one, `getInstance()` fails because there is no project context. Other client methods: `isConnected()`, `getConnectionState()`, `getSessionInfo()`.
## Common Errors
| Error | Cause | Fix |
|-------|-------|-----|
| Maven can't resolve `com.zoho.catalyst:datastreams` | The artifact doesn't exist | Use `com.zoho.catalyst:java-sdk:2.5.0` with the `maven.zohodl.com` repository |
| `unreported exception ZCDataStreamException; must be caught or declared to be thrown` | `subscribe()` called in `onOpen` without try/catch | Wrap `subscribe()` / `unsubscribe()` in try/catch |
| `NullPointerException` in `ZCProject.initProject` / "Exception in parsing the JSON" | `ZCDatastream.getInstance()` outside a Catalyst function | `new ZCDataStreamsWebSocket(config)` |
| `ZCServerException` "Mandatory parameter app_user_id is missing." | `getTokenPair(channelId, String)` on a User Roles channel | Use `getTokenPair(channelId, Long userId)` |
| `cannot find symbol ... CatalystAdvancedIOHandler` | `function-sdk` jar missing from `lib/` | Copy all Maven dependencies into `lib/` |
| Only the first message arrives | No `sendAck()` | Call `sendAck()` in `onMessage` |
references/datastreams-javascript-basics.md
# Data Streams with JavaScript (Node.js and browser)
Server code runs in a Catalyst Advanced I/O function; subscribers run in Node or the browser. Package: `@zcatalyst/datastreams` (exports `DataStreams` and `DataStreamsWebSocket`).
## Install
```bash
# In the function folder
npm install @zcatalyst/datastreams @zcatalyst/auth
```
- Do a fresh install (delete `package-lock.json` and `node_modules`) if `@zcatalyst/auth-admin` 1.1.0 throws `Cannot read properties of undefined (reading 'api')`. That happens when a lockfile keeps `@zcatalyst/utils` at 1.0.0; 1.1.0 fixes it.
- The packages declare `engines: node >=22`, but they run on the `node20` function stack (verified); npm only warns.
- Subscribers only need `@zcatalyst/datastreams`. A bundler (Vite, webpack, esbuild) picks its browser build automatically through the package's `browser` export, with no Node `ws` dependency. For example: `npx esbuild client.js --bundle --platform=browser --format=iife --outfile=bundle.js` (about 60 KB).
- A browser page calls your token endpoint like any API. If the page is served from another origin, that origin must be allowed by your backend's CORS setup; serving the page from the same project avoids it.
Function config (`functions/<name>/catalyst-config.json`):
```json
{
"deployment": {
"name": "stream_api",
"stack": "node20",
"type": "advancedio",
"env_variables": {}
},
"execution": {
"main": "index.js"
}
}
```
Project `catalyst.json` (deploy with `catalyst deploy --only functions:stream_api`):
```json
{ "functions": { "targets": ["stream_api"], "source": "functions" } }
```
The function is reached at `https://<project-domain>/server/<name>/<route>`, so route on the end of `req.url`.
## Initialize in a function
```javascript
const { zcAuth } = require('@zcatalyst/auth');
const { DataStreams } = require('@zcatalyst/datastreams');
module.exports = async (req, res) => {
const app = await zcAuth.init(req); // per request, before any component
const datastreams = new DataStreams(app);
// ...
};
```
In Node.js, `DataStreams` exposes the admin methods (`publishData`, `getAllChannels`, `getChannelDetails`, `getLiveCount`) as well as `getTokenPair`. In the browser it exposes only `getTokenPair`.
## Raw request and response
A plain Advanced I/O function receives Node's `IncomingMessage` and `ServerResponse`. `req.body`, `res.status()` and `res.json()` are all `undefined` (they exist only if you export an Express app). Read and write JSON yourself:
```javascript
const readJson = (req) => new Promise((resolve, reject) => {
let raw = '';
req.on('data', (chunk) => (raw += chunk));
req.on('end', () => {
try { resolve(raw ? JSON.parse(raw) : {}); } catch (e) { reject(e); }
});
});
const sendJson = (res, status, body) => {
res.writeHead(status, { 'Content-Type': 'application/json' });
res.end(JSON.stringify(body));
};
```
## Publish
```javascript
const ok = await datastreams.publishData(channelId, JSON.stringify({ orderId: 12345, status: 'shipped' }));
// ok === true
```
`data` must be a non-empty string: stringify objects.
## Token endpoint (User Roles channel)
Authenticate the caller first, then issue a token for their app user ID.
```javascript
const { zcAuth } = require('@zcatalyst/auth');
const { DataStreams } = require('@zcatalyst/datastreams');
const readJson = (req) => new Promise((resolve, reject) => {
let raw = '';
req.on('data', (chunk) => (raw += chunk));
req.on('end', () => {
try { resolve(raw ? JSON.parse(raw) : {}); } catch (e) { reject(e); }
});
});
const sendJson = (res, status, body) => {
res.writeHead(status, { 'Content-Type': 'application/json' });
res.end(JSON.stringify(body));
};
module.exports = async (req, res) => {
try {
const app = await zcAuth.init(req);
const datastreams = new DataStreams(app);
const { channelId, userId } = await readJson(req);
const token = await datastreams.getTokenPair(channelId, { userId: String(userId) });
sendJson(res, 200, { token }); // { 'wss-id', 'channel-id', key, url, deployment }
} catch (error) {
sendJson(res, 500, { error: error.message, code: error.code });
}
};
```
## Custom Logic function (Custom Logic channel)
Select this function as the channel's access control in the console. Clients call it to get their token; it is the only place a Custom Logic token can be generated.
```javascript
const { zcAuth } = require('@zcatalyst/auth');
const { DataStreams } = require('@zcatalyst/datastreams');
const readJson = (req) => new Promise((resolve, reject) => {
let raw = '';
req.on('data', (chunk) => (raw += chunk));
req.on('end', () => {
try { resolve(raw ? JSON.parse(raw) : {}); } catch (e) { reject(e); }
});
});
const sendJson = (res, status, body) => {
res.writeHead(status, { 'Content-Type': 'application/json' });
res.end(JSON.stringify(body));
};
// Replace with your real check (session cookie, SSO token, device registry...)
const isAllowed = async (req, connectionName) =>
typeof connectionName === 'string' && connectionName.startsWith('device-');
module.exports = async (req, res) => {
try {
const { channelId, connectionName } = await readJson(req);
if (!(await isAllowed(req, connectionName))) {
return sendJson(res, 403, { approved: false, reason: 'Unauthorized' });
}
const app = await zcAuth.init(req);
const datastreams = new DataStreams(app);
const token = await datastreams.getTokenPair(channelId, { connectionName });
sendJson(res, 200, { approved: true, token });
} catch (error) {
sendJson(res, 500, { error: error.message });
}
};
```
## Read channels and live count
```javascript
const channels = await datastreams.getAllChannels(); // array of channel objects
const channel = await datastreams.getChannelDetails(channelId);
const live = await datastreams.getLiveCount(channelId); // { status: 'success', data: { connections: 1 } }
```
## Subscriber (Node or browser)
```javascript
const { DataStreamsWebSocket } = require('@zcatalyst/datastreams'); // or: import { DataStreamsWebSocket } from '@zcatalyst/datastreams';
// token = what your token endpoint returned
const socket = new DataStreamsWebSocket({
url: token.url,
zuid: token['wss-id'],
key: token.key
});
socket.on('open', () => socket.subscribe('0')); // '-1' earliest, '-2' resume
socket.on('message', (event) => {
const payload = JSON.parse(event.data); // data is always a string
console.log(event.streamingId, payload);
socket.sendAck(); // required for the next message
});
socket.on('error', (err) => console.error(err.code, err.message));
socket.on('close', () => console.log('closed'));
// Leaving: socket.unsubscribe() first (frees the session), then socket.close().
// Coming back: release the old session, then a fresh token + subscribe('0')
// (see "Reconnecting" in datastreams-basics.md; don't resume a dropped session with '-2').
```
- Client options: `url` (a host name, not a full URL), `key`, `zuid`, and optional `enableLogging: true`. The connection starts as soon as the object is created.
- Other methods: `isConnected()`, `getConnectionState()` (`'connected'`, `'connecting'`, `'closing'`, `'closed'`, `'disconnected'`), `getSessionInfo()` (`{ sid, uid }`), `unsubscribe()`, `close()`. Events: `open`, `message`, `error`, `close`, `pong`.
- `subscribe()` throws if the socket isn't open yet: call it from the `open` handler.
In Node the client logs one `close` about 1–2 s after connecting and then reconnects on its own. Delivery is unaffected and you do not need to subscribe again: messages keep arriving on the same `message` handler. Don't rely on the `close` event's code: the browser reports 1006, while in Node the event argument may have no `code`. After your own `unsubscribe()`, an `error` event with code `1000` ("Current session will be closed…") is expected. Both builds also close and reconnect every 27 minutes, resubscribing with `'-2'` once an ack has been sent; if that session missed messages, expect the duplicates described in `datastreams-basics.md`, and keep your message handling idempotent. The browser build does not auto-reconnect after a drop: handle `close` yourself.
## Common Errors
| Error | Cause | Fix |
|-------|-------|-----|
| `Cannot read properties of undefined (reading 'api')` from `auth-admin` | `@zcatalyst/auth-admin` 1.1.0 with `@zcatalyst/utils` 1.0.0 from a lockfile | Delete the lockfile and `node_modules`, reinstall |
| `TypeError: Cannot destructure property … of 'req.body'` | Advanced I/O `req` has no parsed body | Read the stream (`readJson` above) |
| `TypeError: res.status is not a function` | Raw `ServerResponse` | `res.writeHead(status, headers); res.end(body)` |
| `zuid` undefined, socket never opens | `token.wss_id` | `token['wss-id']` |
| Only the first message arrives | No `sendAck()` | Call `sendAck()` in every `message` handler |
| `subscribe` has no effect | Called before `open` | Subscribe inside the `open` handler |
| `publishData` validation error | `data` is an object or empty | Pass a non-empty string (`JSON.stringify`) |
references/datastreams-python-basics.md
# Data Streams with Python
Server code runs in a Catalyst Python function; subscribers run in any Python 3.10+ process. Package: `zcatalyst-sdk` **1.5.0 or later** (module `zcatalyst_sdk.datastreams`: `Datastreams`, `DataStreamsWebSocket`, `DataStreamsConfig`). Everything below was verified against live calls with 1.5.0.
## Install
```bash
pip install "zcatalyst-sdk>=1.5.0"
```
- Requires **Python 3.10+**. It pulls in `requests` and `websocket-client` (used by the subscriber).
- Function stacks: `python_3_10` to `python_3_13`. The CLI needs a local interpreter matching the stack, or `catalyst deploy` fails with "unable to locate python3.11 in your system"; set it with `catalyst config:set python3_11.bin=<path>`, or choose the stack that matches your local Python.
- Verified deploy: dependencies installed into the function folder for the Linux runtime, next to `main.py`:
`pip install -r requirements.txt -t . --platform manylinux2014_x86_64 --only-binary=:all: --python-version 3.13 --implementation cp`
(use the `--python-version` of your stack; `charset_normalizer` ships compiled Linux wheels).
Function config (`functions/<name>/catalyst-config.json`):
```json
{
"deployment": {
"name": "stream_api",
"stack": "python_3_13",
"type": "advancedio",
"env_variables": {}
},
"execution": {
"main": "main.py"
}
}
```
## Initialize in a function
```python
import zcatalyst_sdk
from flask import Request, jsonify, make_response
def handler(request: Request):
app = zcatalyst_sdk.initialize(req=request) # per request
datastreams = app.datastreams() # plural; there is no app.datastream()
...
```
- The SDK's own docstrings show `app.datastream()`; that method does not exist. Use `app.datastreams()`.
- Flask is provided by the runtime: don't vendor it. The handler receives every sub-path of `https://<project-domain>/server/<name>/`, so route on `request.path`.
- Project `catalyst.json`: `{ "functions": { "targets": ["stream_api"], "source": "functions" } }`, then `catalyst deploy --only functions:stream_api`.
- In an Advanced I/O function, both `initialize(req=request)` and `initialize(req=request, scope='admin')` worked for every call below. `get_token_pair` sends user credentials internally, so in functions without a request user (Job, Cron, Event), initialize with `scope='admin'`.
- `get_all_channels`, `get_channel_details`, `get_live_count` and `publish_data` are **admin** methods: server only.
## Publish
```python
import json
ok = datastreams.publish_data(channel_id, json.dumps({"orderId": 12345, "status": "shipped"}))
# ok is True
```
`data` must be a non-empty **string**: passing a dict fails validation with `Invalid-Argument` "Value provided for data is expected to be a non-empty string." `json.dumps` objects and `json.loads` on receipt.
## Token endpoint (User Roles channel)
Authenticate the caller first, then issue a token for their app user ID.
```python
def handler(request: Request):
app = zcatalyst_sdk.initialize(req=request)
body = request.get_json(silent=True) or {}
token = app.datastreams().get_token_pair(str(body["channelId"]), user_id=body["userId"])
return make_response(jsonify({"token": token}), 200)
```
- Signature: `get_token_pair(channel_id: str, user_id=None, connection_name=None)`. Pass `user_id` or `connection_name` as a **keyword**. If both are given, `user_id` wins.
- `channel_id` must be a **string** here: an `int` fails with `Invalid-Argument` "Value provided for channel_id is expected to be a non-empty string." (The other methods accept a string or a number.)
- The result is a plain dict with **hyphenated** keys: `wss-id`, `channel-id`, `key`, `url`, `deployment`. The SDK's type hints say `wss_id` / `channel_id`; those keys don't exist. Use `token["wss-id"]`.
- On a User Roles channel, `connection_name` fails with `MISSING_PARAM` "Mandatory parameter app_user_id is missing."
## Custom Logic channel
The token must be generated inside the function selected as the channel's access control, with `connection_name`:
```python
token = app.datastreams().get_token_pair(channel_id, connection_name="device-4f2a")
```
From any other function, Python included, the call fails with `DATASTREAMS_ERROR` "Invalid access control source! Kindly request from the configured one!" (verified). The rules in `datastreams-basics.md` apply unchanged: run your own check before issuing the token, and keep connection names at most 20 characters.
## Read channels and live count
```python
channels = datastreams.get_all_channels() # list of channel dicts
channel = datastreams.get_channel_details(channel_id)
live = datastreams.get_live_count(channel_id) # {'connections': 1}
```
## Subscriber
```python
import json, threading
from zcatalyst_sdk.datastreams import DataStreamsWebSocket
# token = what your token endpoint returned
ws = DataStreamsWebSocket(url=token["url"], zuid=token["wss-id"], key=token["key"])
def on_open(event): # event.code == 200
ws.subscribe("0") # subscribe only after open
def on_message(event):
payload = json.loads(event.data) # data is always a string
print(event.streaming_id, payload)
ws.send_ack() # required for the next message
ws.set_on_open(on_open)
ws.set_on_message(on_message)
ws.set_on_error(lambda e: print("error", e.code, e.message))
ws.set_on_close(lambda e: print("closed", e))
ws.connect() # nothing happens until connect()
threading.Event().wait() # keep the process alive; the socket runs on a daemon thread
# Leaving: ws.unsubscribe(), then ws.close()
```
- Unlike the JavaScript client, creating the object does not connect: call **`connect()`**. It runs the socket on a **daemon thread**, so the main thread must stay alive or the process exits.
- Events are set with `set_on_open`, `set_on_message`, `set_on_error`, `set_on_close`, `set_on_pong`. Message events have `data`, `streaming_id` (snake case), and `url` / `method` (set only on API-operation messages).
- For logging or custom timers, pass a config instead: `DataStreamsWebSocket(config=DataStreamsConfig(url=..., zuid=..., key=..., enable_logging=True))`.
- Other methods: `is_connected()`, `get_connection_state()` (a `ConnectionState` enum: `connected`, `connecting`, `closing`, `closed`, `disconnected`), `get_session_info()` (`{'sid', 'uid'}`), `unsubscribe()`, `close()`. `subscribe()` and `unsubscribe()` raise `ConnectionError` when the socket isn't open.
- `url` is checked against a built-in list of Data Streams hosts. The `url` from the token pair passes; don't hardcode or rewrite it.
- Verified: messages arrived in publish order, `get_live_count` showed 1 while connected, and without `send_ack()` only the first message arrived.
- `unsubscribe()` is followed by error `1000` "Current session will be closed. New connection to be established." and a lost-connection `1006`: that is the session ending, not a failure.
- A connection that drops without `unsubscribe()` (or `close()` without it) leaves the session open; release it as described in "Reconnecting" in `datastreams-basics.md`.
- Like the JavaScript client, it closes and reconnects every 27 minutes, resubscribing with `'-2'` once an ack has been sent. Follow "Reconnecting" in `datastreams-basics.md` and keep message handling idempotent.
## Common Errors
| Error | Cause | Fix |
|-------|-------|-----|
| `AttributeError: 'CatalystApp' object has no attribute 'datastream'` | Followed the SDK docstring | `app.datastreams()` |
| `KeyError: 'wss_id'` | Followed the type hints | `token["wss-id"]` |
| `Invalid-Argument` "…channel_id is expected to be a non-empty string" | `get_token_pair` with an `int` channel ID | Pass `str(channel_id)` |
| `Invalid-Argument` "…data is expected to be a non-empty string" | `publish_data` with a dict or empty string | `json.dumps(...)` |
| `MISSING_PARAM` "Mandatory parameter app_user_id is missing." | `connection_name` on a User Roles channel | Use `user_id=` with an app user's ID |
| `DATASTREAMS_ERROR` "Invalid access control source!…" | Custom Logic token requested outside the configured function | Generate it inside the configured function |
| `unable to locate python3.11 in your system` on deploy | No local interpreter for the stack | `catalyst config:set python3_11.bin=<path>`, or match the stack to your Python |
| Subscriber never receives anything, no error | `connect()` not called, or the script exited | Call `connect()` and keep the main thread alive |
| Only the first message arrives | No `send_ack()` | Call `send_ack()` in the message handler |
SKILL.md
---
name: catalyst-datastreams
description: "Catalyst Data Streams — real-time publish/subscribe channels over WebSocket. Publish from a function with publishData / publish_data, issue a token pair with getTokenPair / get_token_pair, subscribe with DataStreamsWebSocket / ZCDataStreamsWebSocket (Node.js, browser, Java, Python), sendAck, live/earliest/resume subscribe types, 48-hour retention, User Roles vs Custom Logic access control. Trigger on 'Data Streams', 'datastreams', 'real-time channel', 'live updates', 'pub/sub', 'WebSocket subscriber', 'token pair', 'wss-id', 'app_user_id is missing', 'Invalid access control source', 'Authentication failed 3000', '@zcatalyst/datastreams', or 'zcatalyst_sdk.datastreams'."
metadata:
version: "2.0.0"
---
## How It Works
1. **Channels and access control are console-only.** Create the channel and choose its access control in Console → Cloud Scale → Data Streams; no SDK, REST, or MCP tool creates channels. Ask the user for the channel ID and its access control type, or list channels with `getAllChannels()` (`control_type` 0 = User Roles, 1 = Custom Logic). Load `references/datastreams-basics.md` for concepts, the token rules, subscribe types, acknowledgement, and the REST API.
2. **Pick the token identifier from the access control — it is not interchangeable.**
- **User Roles** → `userId` = an app user's ID from Catalyst Authentication. A console admin's user ID is rejected; `connectionName` fails with "Mandatory parameter app_user_id is missing."
- **Custom Logic** → `connectionName`, and the token must be generated **inside the Advanced I/O function selected as the channel's access control**. That function is the token endpoint: it runs your check, then calls `getTokenPair`. Catalyst never calls it on its own. From any other function or REST: "Invalid access control source!". User-ID tokens on these channels are issued but the WebSocket rejects them (`3000`).
3. **Server code (publish, token pair, channel reads) — load the runtime reference.**
- Node.js / JavaScript → `references/datastreams-javascript-basics.md`. Initialize with the app: `new DataStreams(await zcAuth.init(req))`. Advanced I/O handlers get Node's raw `req`/`res`: no `req.body`, `res.status()`, or `res.json()`.
- Python → `references/datastreams-python-basics.md`. Requires `zcatalyst-sdk` 1.5.0+ (Python 3.10+). `app = zcatalyst_sdk.initialize(req=request)`, then `app.datastreams()` (plural). Pass `get_token_pair` a **string** channel ID and a keyword `user_id=` / `connection_name=`; the result's keys are hyphenated (`token["wss-id"]`). The subscriber does nothing until `connect()`.
- Java → `references/datastreams-java-basics.md`. Dependency is `com.zoho.catalyst:java-sdk` (2.5.0+) from `https://maven.zohodl.com`; no `datastreams` artifact exists.
- `publishData`, `getAllChannels`, `getChannelDetails` and `getLiveCount` are **admin** methods: server only, never in browser or mobile code. `getTokenPair` also exists in the browser SDK, but a Custom Logic token must come from the configured function.
4. **Subscriber code — read the token correctly and always acknowledge.** The token pair's keys are hyphenated: pass `token['wss-id']` as `zuid`, plus `token.url` and `token.key`. Subscribe only after the `open` event. Call `sendAck()` after every message, or the next message is never delivered. Messages carry `data` (always a string) and `streamingId` (plus `url`/`method` on API-operation messages). To reconnect, release the old session (old token → `unsubscribe()`), take a fresh token, `subscribe('0')`, and reload state from your own store; resuming a dropped session with `'-2'` redelivers missed messages twice and replays the 48-hour retention. Unsubscribe before every deliberate close; the `error` event with code `1000` that follows your own `unsubscribe()` is expected. In Node, `getLiveCount()` returns `{ status, data: { connections } }`. Java: `subscribe()` throws the checked `ZCDataStreamException`, and outside a Catalyst function create the client with `new ZCDataStreamsWebSocket(config)`, not `ZCDatastream.getInstance()`.
5. **Verify end to end before reporting success.** Open a subscriber, check `getLiveCount()` shows it, publish a test message, and confirm the subscriber received it. For anything the references do not cover, fetch the page under <https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/> as Markdown (`<page-url>/index.md`), and prefer the references where they disagree.
## Security Checklist
- `publishData` and the channel-read methods are admin operations: run them only in a Catalyst function, never in a browser or mobile app.
- Protect the token endpoint. Anyone who can call it can subscribe to the channel, so authenticate the caller before issuing a token.
- On Custom Logic channels, the configured function is the only gatekeeper: put the real validation there before calling `getTokenPair`.
- Do not log or return the token `key` anywhere except to the client that asked for it.
## Triggers
Use this skill for: "Data Streams", "Catalyst Data Streams", "datastreams", "real-time channel", "live updates", "push updates to the browser", "pub/sub", "publish data", `publishData`, `getTokenPair`, "token pair", "wss-id", `DataStreamsWebSocket`, `ZCDataStreamsWebSocket`, `ZCDatastream`, "WebSocket subscriber", `sendAck`, "subscribe type", "resume from last message", "live count", `getLiveCount`, "channel access control", "User Roles channel", "Custom Logic channel", "app_user_id is missing", "Invalid access control source", "Authentication failed 3000", "reconnect", "duplicate messages after reconnect", "DUPLICATE_VALUE", `@zcatalyst/datastreams`, `zcatalyst_sdk`, `get_token_pair`, `publish_data`, `send_ack`, or "Data Streams Python".
## References
| Reference | Load when the query is about… |
|-----------|-------------------------------|
| `references/datastreams-basics.md` | Concepts, channels, access control (User Roles vs Custom Logic), which token identifier to use, token pair shape, subscribe types, acknowledgement, retention, message fields, REST API endpoints and scopes |
| `references/datastreams-javascript-basics.md` | Node.js functions (init, publish, token endpoint, Custom Logic function, channel reads), Node and browser subscribers, package versions |
| `references/datastreams-python-basics.md` | Python functions and subscribers: `zcatalyst-sdk` 1.5.0+, `app.datastreams()`, `get_token_pair` argument rules, `DataStreamsWebSocket` with `connect()` and `set_on_*` callbacks, deploying Python functions |
| `references/datastreams-java-basics.md` | Java functions and subscribers: Maven setup, `ZCDatastream`, `ZCDataStreamsWebSocket`, checked exceptions, running outside Catalyst |