# Implementation



--------------------------------------------------------------------------------
title: "Create a Data Streams Channel"
description: "Step-by-step guide to creating a new Data Streams channel in the Catalyst Console with access control configuration."
last_updated: "2026-10-08T06:47:52.524Z"
source: "https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/implementation/create-channel/"
service: "Cloud Scale"
related:
- Channels Overview (/en/cloud-scale/help/data-streams/channels-overview/)
- Setup Access Control (/en/cloud-scale/help/data-streams/implementation/setup-access-control/)
- Publish Data with SDK (/en/cloud-scale/help/data-streams/implementation/publish-data-sdk/)
- Key Concepts (/en/cloud-scale/help/data-streams/key-concepts/)

--------------------------------------------------------------------------------


# Create a Channel

To create a new Data Streams channel:

1. Access **Data Streams** from the service menu under *Cloud Scale*.
  <br />

2. Click the **Create Channel** button.
  <br />

3. Enter a descriptive name for your channel (e.g., "user-notifications", "order-updates", "live-chat").
  <br />

4. Select an access control type to determine how subscribers will be authenticated when connecting to this channel:
   - **User Roles** — Restricts channel access to authenticated users with specific roles. Select **User Roles** from the dropdown and choose one or more roles from your Catalyst Authentication configuration. Use this when subscribers are authenticated users in your application with defined roles.
   - **Custom Logic** — Delegates authentication to an Advanced I/O function you define. Select **Custom Logic** from the dropdown and choose an existing function. Use this when you need custom authentication logic (SSO, API keys, anonymous users, device-based access).

Note: For detailed access control configuration, refer to Setup Access Control. To create user roles, refer to the User Roles documentation.

5. Click **Create**.
  <br />

Your channel is ready to receive and distribute data. You will be redirected to the *Test Data Flow* section where you can verify that data flows correctly between publishers and subscribers. The **Channel ID** (required for all SDK and API interactions) is available in the *General Details* section of the *Dashboard* tab.

<br />

Note: Always use the Channel ID in your code, not the Channel Name. The Channel ID never changes.




--------------------------------------------------------------------------------
title: "Manage Data Streams Channels"
description: "Edit channel names, update access control configuration, and delete Data Streams channels from the Catalyst Console."
last_updated: "2026-10-08T06:47:52.524Z"
source: "https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/implementation/manage-channels/"
service: "Cloud Scale"
related:
- Create a Channel (/en/cloud-scale/help/data-streams/implementation/create-channel/)
- Setup Access Control (/en/cloud-scale/help/data-streams/implementation/setup-access-control/)
- Channels Overview (/en/cloud-scale/help/data-streams/channels-overview/)

--------------------------------------------------------------------------------


# Manage Channels

Note: You must have a Data Streams channel already created before performing these operations.

### Edit Channel Name

You can update the Channel Name at any time. The Channel ID remains unchanged.




1. Click the **three-dot menu** next to the channel.
2. Select **Edit**.
  <br />

3. Update the **Channel Name**.
4. Click **Update** to save changes.
  <br />




1. While viewing a channel, click the **three-dot menu** in the top-right corner.
2. Select **Edit**.
  <br />

3. Update the **Channel Name**.
4. Click **Update** to save changes.
  <br />




Note: If your publishers or subscribers reference the channel by name in your code, you must update those references. If you use the Channel ID (recommended), no code changes are needed.

### Edit Access Control

To update subscriber authentication rules:

1. Go to the **Access Control** tab.
<br />

2. Click **Edit** in the **Subscriber Access** section.
  <br />

3. Modify the access control configuration:
   - **User Roles** — Add or remove roles
   - **Custom Logic** — Select a different Advanced I/O function



4. Click **Update** to save changes.

Note: When revoking access, subscribers lose access immediately and active connections are terminated. When granting access, new users can subscribe and already-connected users will continue receiving data. Refer to Setup Access Control for detailed configuration guidance.

### Delete a Channel

Note: This action cannot be undone. All active connections will be terminated immediately. Publishers will no longer be able to send data, and subscribers will no longer receive events.




1. Click the **three-dot menu** next to the channel.
2. Select **Delete**.
  <br />

3. Type the **Channel Name** in the confirmation dialog.


4. Click **Delete** to confirm.




1. While viewing a channel, click the **three-dot menu** in the top-right corner.
2. Select **Delete**.
  <br />

3. Type the **Channel Name** in the confirmation dialog.
<br />

4. Click **Delete** to confirm.







--------------------------------------------------------------------------------
title: "Publish Data to Channel Using SDK"
description: "Publish data to Data Streams channels from server-side Catalyst functions using Java, JavaScript, or Python SDK."
last_updated: "2026-10-08T06:47:52.527Z"
source: "https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/implementation/publish-data-sdk/"
service: "Cloud Scale"
related:
- Create a Channel (/en/cloud-scale/help/data-streams/implementation/create-channel/)
- Key Concepts - Publishing Methods (/en/cloud-scale/help/data-streams/key-concepts/#publishing-methods)
- Advanced I/O Functions (/en/serverless/help/functions/advanced-io/)
- Java SDK (/en/sdk/java/v1/cloud-scale/data-streams/)
- JavaScript SDK (/en/sdk/javascript/v1/cloudscale/data-streams/)
- Python SDK (/en/sdk/python/v1/cloud-scale/data-streams/)

--------------------------------------------------------------------------------


# Publish Data with SDK

To publish data to a Data Streams channel using the Catalyst SDK:

Note: You must have a Data Streams channel created (you'll need the Channel ID) and an Advanced I/O function for publishing. Publishing requires admin-level authentication. The publishData() method can only be called from a server-side Catalyst function, not from client/browser applications.

1. Navigate to your Advanced I/O function directory and install the SDK for your runtime.





npm install @zcatalyst/datastreams





Add the following dependency to your pom.xml:


<!-- Refer to SDK documentation for the latest version -->
<dependency>
    <groupId>com.zoho.catalyst</groupId>
    <artifactId>datastreams</artifactId>
    <version>LATEST</version>
</dependency>






pip install zcatalyst-sdk





2. Initialize the SDK in your function code. The SDK automatically resolves project credentials from the execution environment.





const { DataStreams } = require('@zcatalyst/datastreams');

module.exports = async (req, res) => {
    const datastreams = new DataStreams();
    // ... use datastreams methods
};






import com.zc.component.datastream.ZCDatastream;

ZCDatastream datastream = ZCDatastream.getInstance();






import zcatalyst_sdk

app = zcatalyst_sdk.initialize_app()
data_streams = app.datastreams()





Note: Do not use DataStreamsWebSocket for publishing. That class is for subscribers only.

3. Use the publishData() method to send data to your channel.

**Method Parameters:**

<table class="content-table">
<thead>
<tr>
<th>Parameter</th>
<th>Type</th>
<th>Description</th>
</tr>
</thead>
<tbody>
<tr>
<td>channelId</td>
<td>string</td>
<td>The unique Channel ID from the console</td>
</tr>
<tr>
<td>data</td>
<td>string (JavaScript/Java)<br>string or dict (Python)</td>
<td>The data to publish</td>
</tr>
</tbody>
</table>

**Publish text data:**





const result = await datastreams.publishData('YOUR_CHANNEL_ID', 'Hello, subscribers!');
console.log('Published successfully:', result);






Boolean result = datastream.publishData("YOUR_CHANNEL_ID", "Hello, subscribers!");
System.out.println("Published successfully: " + result);






result = data_streams.publish_data("YOUR_CHANNEL_ID", "Hello, subscribers!")
print("Published successfully:", result)





4. For structured data, stringify your JSON before publishing (JavaScript/Java) or pass a dictionary (Python).





const payload = JSON.stringify({
    event: 'temperature_update',
    value: 23.5,
    unit: 'celsius',
    timestamp: Date.now()
});

const result = await datastreams.publishData('YOUR_CHANNEL_ID', payload);
console.log('Published successfully:', result);


Note: Always use JSON.stringify(). Passing an object directly will throw a validation error.





String payload = new Gson().toJson(Map.of(
    "event", "temperature_update",
    "value", 23.5,
    "unit", "celsius",
    "timestamp", System.currentTimeMillis()
));

Boolean result = datastream.publishData("YOUR_CHANNEL_ID", payload);
System.out.println("Published successfully: " + result);


Note: Use Gson or another JSON library to convert objects to strings.





# Python SDK accepts dictionaries directly
result = data_streams.publish_data("YOUR_CHANNEL_ID", {
    "event": "temperature_update",
    "value": 23.5,
    "unit": "celsius",
    "timestamp": time.time()
})
print("Published successfully:", result)

# Or pass a JSON string
import json
payload = json.dumps({"event": "order_placed", "order_id": 12345})
result = data_streams.publish_data("YOUR_CHANNEL_ID", payload)


Note: Python SDK handles both dictionaries and JSON strings.




After publishing, data is sent to the Data Streams channel where it is distributed to all active subscribers in real-time. Publishing succeeds even if no subscribers are connected; data is retained for 48 hours.

### Common Errors

<table class="content-table">
<thead>
<tr>
<th>Error</th>
<th>Cause</th>
<th>Solution</th>
</tr>
</thead>
<tbody>
<tr>
<td>Invalid Channel ID</td>
<td>Wrong or non-existent channel</td>
<td>Verify Channel ID in console</td>
</tr>
<tr>
<td>Data must be a string</td>
<td>Passed object instead of string</td>
<td>Use JSON.stringify() (JavaScript/Java)</td>
</tr>
<tr>
<td>Unauthorized</td>
<td>Called from client-side</td>
<td>Move to server-side function</td>
</tr>
<tr>
<td>Channel not found</td>
<td>Channel deleted</td>
<td>Recreate channel or update Channel ID</td>
</tr>
</tbody>
</table>




--------------------------------------------------------------------------------
title: "Publish Data Using Signals Integration"
description: "Automatically publish data to Data Streams channels when events occur in connected services using Catalyst Signals event bus."
last_updated: "2026-10-08T06:47:52.528Z"
source: "https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/implementation/publish-data-signals/"
service: "Cloud Scale"
related:
- Create a Channel (/en/cloud-scale/help/data-streams/implementation/create-channel/)
- Publish Data with SDK (/en/cloud-scale/help/data-streams/implementation/publish-data-sdk/)
- Catalyst Signals (/en/signals/getting-started/introduction/)
- Key Concepts - Publishing Methods (/en/cloud-scale/help/data-streams/key-concepts/#publishing-methods)

--------------------------------------------------------------------------------


# Publish Data with Signals

To automatically publish data to a Data Streams channel when events occur in connected services:

Note: You must have created a Data Streams channel (you'll need the Channel ID), Catalyst Signals enabled in your project, and an event source configured (e.g., Catalyst Authentication, Data Store, Zoho CRM, or custom webhooks). Signals is an event bus service that routes events from publishers to targets without requiring custom code.

1. Configure a Signals Publisher (the event source that emits events).




1. Navigate to your Data Streams channel.
2. Go to the **Dashboard** tab.
3. Find the *Triggers* section.
4. Click **Go to Signals** next to "Signals".
  <br />

This redirects you to Catalyst Signals with the channel context pre-loaded.




1. Navigate to **Catalyst Signals** in the console.
2. Go to the *Publishers* section.
3. Click **Create Publisher**.
4. Select your event source:
   - **Catalyst Services** — Authentication, Data Store, Cache, Stratus
   - **Zoho Services** — CRM, Desk, Projects
   - **Custom Webhooks** — External services

5. Configure the publisher settings.
6. Click **Create**.




Note: For detailed publisher configuration, refer to Catalyst Signals Publishers documentation.

2. Create a Signals Rule (defines when and how events should be routed).
   - In Catalyst Signals, navigate to *Rules*.
   - Click **Create Rule**.
   - Configure the rule:
     - **Name** — Descriptive name (e.g., "Route user signups to notifications channel")
     - **Publisher** — Select the publisher you configured
     - **Event Filter** — (Optional) Filter specific events
     - **Condition** — (Optional) Add conditions to route only certain events
   - Click **Create**.

Note: For detailed rule configuration, refer to Signals Rules documentation.

3. Add a Data Streams Target (specifies where the event should be sent).
   - While creating or editing a rule, scroll to the *Targets* section.
   - Click **Add Target**.
   - Select **Data Streams** from the target type dropdown.
   - Enter your **Channel ID** (from the Data Streams console).
   - Click **Add**.
   - Save the rule.

Note: For detailed target configuration, refer to Signals Targets documentation.

4. Test the configuration by triggering an event in your source service:
   - **Catalyst Authentication** — Create a test user signup
   - **Data Store** — Insert a row in the configured table
   - **Zoho CRM** — Create or update a record
   - **Custom Webhook** — Send a test POST request

5. Verify in Data Streams:
   - Navigate to your Data Streams channel.
   - Go to the *Test Data Flow* tab.
   - Click the **Receive** tab.
   - Click **Activate Stream**.
   - Watch the event appear in real-time.
  <br />

When events occur in connected services, Signals automatically publishes them to your Data Streams channel where all connected subscribers receive the data in real-time.

--------------------------------------------------------------------------------
title: "Setup Access Control for Data Streams"
description: "Configure User Roles or Custom Logic authentication to control subscriber access to Data Streams channels."
last_updated: "2026-10-08T06:47:52.528Z"
source: "https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/implementation/setup-access-control/"
service: "Cloud Scale"
related:
- Create a Channel (/en/cloud-scale/help/data-streams/implementation/create-channel/)
- Manage Channels (/en/cloud-scale/help/data-streams/implementation/manage-channels/)
- User Roles - Catalyst Authentication (/en/cloud-scale/help/authentication/user-management/roles/introduction/)
- Advanced I/O Functions (/en/serverless/help/functions/advanced-io/)

--------------------------------------------------------------------------------


# Setup Access Control

To configure access control for your channel:

Note: You should have a channel created before configuring access control. Access control applies only to subscribers. Publishers are authenticated through the Catalyst SDK or the REST API and do not require access control configuration.

1. Select an access control type during channel creation or from the *Access Control* section.




Use this when subscribers are authenticated users in Catalyst Authentication with predefined roles.

2. Select **User Roles**.
3. Select one or more roles from the dropdown menu.
  <br />

4. Click **Create** or **Update**.

Only users belonging to the selected roles can subscribe to the channel. When a role is removed, subscribers with that role lose access immediately. When a role is granted, users with that role can subscribe immediately.

Note: To create new roles, refer to the Catalyst Authentication User Roles documentation.




Use this when you need custom authentication logic beyond role-based access (SSO, session-based connections, anonymous users, or custom criteria).

2. Create an Advanced I/O function that handles subscriber authentication:


module.exports = async (req, res) => {
    const { userId, connectionName } = req.body;
    
    // Your custom validation logic
    const isAuthorized = await yourCustomValidationLogic(userId);
    
    if (isAuthorized) {
        res.status(200).json({ approved: true });
    } else {
        res.status(403).json({ approved: false, reason: "Unauthorized" });
    }
};


The function must accept connection requests, perform validation logic, and return approval or rejection.

3. Select **Custom Logic**.
4. Select your Advanced I/O function from the dropdown. If the function doesn't exist, click **Create new Advanced I/O function** to create one first.
  <br />

5. Click **Create** or **Update**.

Note: Only Advanced I/O functions are supported. Ensure the function is deployed and available in your project before configuring it.




### Manage Subscriber Connections

After access control is configured, navigate to the *Manage Connections* section in the *Access Control* tab to view all subscribers.

<br />

Each connection displays:

<table class="content-table">
<thead>
<tr>
<th>Field</th>
<th>Description</th>
</tr>
</thead>
<tbody>
<tr>
<td>Connection Name</td>
<td>Unique identifier for the subscriber connection (generated during token pair creation)</td>
</tr>
<tr>
<td>User Type</td>
<td>Type of user (App User or Custom User) based on access control configuration</td>
</tr>
<tr>
<td>Status</td>
<td>Current connection state: Active, Inactive, or Blocked</td>
</tr>
</tbody>
</table>

**To block a subscriber:**

1. Locate the subscriber connection.
2. Click the **three-dot menu** icon.
3. Select **Block**.
  <br />

Blocking terminates the active connection immediately.

**To delete a subscriber connection:**

1. Locate the subscriber connection.
2. Click the **three-dot menu** icon.
3. Select **Delete**.
4. Confirm deletion in the dialog.


--------------------------------------------------------------------------------
title: "Generate Token Pair for Subscribers"
description: "Create authentication credentials for subscribers to establish WebSocket connections to Data Streams channels."
last_updated: "2026-10-08T06:47:52.528Z"
source: "https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/implementation/generate-token-pair/"
service: "Cloud Scale"
related:
- Create a Channel (/en/cloud-scale/help/data-streams/implementation/create-channel/)
- Setup Access Control (/en/cloud-scale/help/data-streams/implementation/setup-access-control/)
- Subscribe with WebSocket (/en/cloud-scale/help/data-streams/implementation/subscribe-websocket/)
- Key Concepts - Tokens (/en/cloud-scale/help/data-streams/key-concepts/#token-pair)

--------------------------------------------------------------------------------


# Generate Token Pair

A token pair provides the authentication credentials required for subscribers to establish WebSocket connections to a Data Streams channel.

Before you begin, ensure you have created a channel, configured access control, and an Advanced I/O function to host the token generation endpoint.

Note: Token pair generation is a server-side operation that requires admin-level authentication. You must call getTokenPair() from a Catalyst Advanced I/O function and return the token pair to the client for WebSocket connection. Never expose token generation logic in client-side code.

A token pair contains the WebSocket server URL (url), session identifier (wss_id or zuid), authentication key (key), and channel ID (channel_id).

### Create Token Generation Endpoint

Create an Advanced I/O function that generates and returns token pairs to clients. You can generate a token pair using either User ID (for authenticated users with Catalyst Authentication) or Connection Name (for anonymous or device-based connections). You must provide at least one.

<table class="content-table">
<thead>
<tr>
<th>Parameter</th>
<th>Type</th>
<th>Description</th>
</tr>
</thead>
<tbody>
<tr>
<td>channelId</td>
<td>string</td>
<td>The unique identifier of the channel to subscribe to</td>
</tr>
<tr>
<td>userId</td>
<td>string / Long (Java)</td>
<td>The Catalyst project user ID (for authenticated users)</td>
</tr>
<tr>
<td>connectionName</td>
<td>string</td>
<td>Custom connection identifier (for anonymous/device connections)</td>
</tr>
</tbody>
</table>





const { DataStreams } = require('@zcatalyst/datastreams');

module.exports = async (req, res) => {
    const datastreams = new DataStreams();
    
    // Option 1: Using a user ID (for authenticated users)
    const tokenByUser = await datastreams.getTokenPair('YOUR_CHANNEL_ID', {
        userId: 'YOUR_USER_ID'
    });
    
    // Option 2: Using a connection name (for anonymous users)
    const tokenByName = await datastreams.getTokenPair('YOUR_CHANNEL_ID', {
        connectionName: 'YOUR_CONNECTION_NAME'
    });
    
    // Return token pair to client
    res.setHeader('Content-Type', 'application/json');
    res.writeHead(200);
    res.end(JSON.stringify({ status: 'success', data: tokenByUser }));
};





import com.zc.component.datastream.ZCDatastream;
import com.zc.component.datastream.beans.ZCTokenResponse;

ZCDatastream datastream = ZCDatastream.getInstance();

// Option 1: Using a user ID (Long type for authenticated users)
Long userId = 30772000000631094L;
ZCTokenResponse tokenByUser = datastream.getTokenPair("YOUR_CHANNEL_ID", userId);

// Option 2: Using a connection name (String type for anonymous users)
ZCTokenResponse tokenByName = datastream.getTokenPair("YOUR_CHANNEL_ID", "YOUR_CONNECTION_NAME");

// Return tokenByUser or tokenByName to client





import zcatalyst_sdk

app = zcatalyst_sdk.initialize_app()
data_streams = app.datastreams()

# Option 1: Using a user ID (for authenticated users)
token_by_user = data_streams.get_token_pair("YOUR_CHANNEL_ID", user_id="YOUR_USER_ID")

# Option 2: Using a connection name (for anonymous users)
token_by_name = data_streams.get_token_pair("YOUR_CHANNEL_ID", connection_name="YOUR_CONNECTION_NAME")

# Return token to client




### Deploy and Test

To deploy and test your token generation endpoint:

1. Deploy your Advanced I/O function
2. Call the function endpoint from your client application
3. The client receives the token pair response:


{
    "url": "wss://datastreams.zoho.com",
    "wss_id": "abc123sessionid",
    "key": "authentication_key_here",
    "channel_id": "1234567890"
}


### Example Implementation

Server (Advanced I/O Function):


const { DataStreams } = require('@zcatalyst/datastreams');

module.exports = async (req, res) => {
    try {
        const datastreams = new DataStreams();
        const { channelId, userId, connectionName } = req.body;
        
        // Generate token pair
        const tokenPair = await datastreams.getTokenPair(channelId, {
            userId: userId || undefined,
            connectionName: connectionName || undefined
        });
        
        res.status(200).json({
            success: true,
            token: tokenPair
        });
    } catch (error) {
        res.status(500).json({
            success: false,
            error: error.message
        });
    }
};


Client (Request token from server):


// Client-side: Fetch token pair from your server
const response = await fetch('https://your-function-url.com/get-token', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
        channelId: 'YOUR_CHANNEL_ID',
        userId: 'current_user_id'  // or connectionName: 'device_123'
    })
});

const { token } = await response.json();
// Now use token to create WebSocket connection


Always validate subscriber identity before generating tokens, use User ID for authenticated users, and use Connection Name for anonymous connections. Tokens are scoped to specific channels.

You can find further steps on subscribing with WebSocket to create connections using the token pair, testing data flow to verify token generation, Token Pairs in Key Concepts, and WebSocket Connections.


--------------------------------------------------------------------------------
title: "Subscribe to Data Streams via WebSocket"
description: "Connect to Data Streams channels and receive real-time data using WebSocket connections with Java, JavaScript, or Python SDK."
last_updated: "2026-10-08T06:47:52.529Z"
source: "https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/implementation/subscribe-websocket/"
service: "Cloud Scale"
related:
- Generate Token Pair (/en/cloud-scale/help/data-streams/implementation/generate-token-pair/)
- Key Concepts - Subscriber Types (/en/cloud-scale/help/data-streams/key-concepts/#subscriber-types)
- Key Concepts - Acknowledgement (/en/cloud-scale/help/data-streams/key-concepts/#acknowledgement)
- Setup Access Control (/en/cloud-scale/help/data-streams/implementation/setup-access-control/)

--------------------------------------------------------------------------------


# Subscribe to Channel via WebSocket

Subscribe to a Data Streams channel to receive data from publishers in real time over a WebSocket connection.

Before you begin, ensure you have a channel created, access control configured, and a token pair available. Subscriptions are typically established from client-side applications (web or mobile) because WebSocket connections are long-lived and must remain open to receive data continuously.

Note: The JavaScript SDK supports browser environments for subscriber operations.

### Install the Data Streams SDK





npm install @zcatalyst/datastreams




Add to pom.xml:


<!-- Refer to SDK documentation for latest version -->
<dependency>
    <groupId>com.zoho.catalyst</groupId>
    <artifactId>datastreams</artifactId>
    <version>LATEST</version>
</dependency>





pip install zcatalyst-sdk




### Get Token Pair from Server

First, request a token pair from your server endpoint (created in generate-token-pair):


// Client-side example
const response = await fetch('https://your-server.com/get-token', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
        channelId: 'YOUR_CHANNEL_ID',
        userId: 'current_user_id'
    })
});

const { token } = await response.json();
// token contains: { url, wss_id, key, channel_id }


### Create WebSocket Connection

Use the token pair to create a WebSocket client instance. The SDK handles authentication automatically.




const { DataStreamsWebSocket } = require('@zcatalyst/datastreams');

const socket = new DataStreamsWebSocket({
    url: token.url,
    zuid: token.wss_id,
    key: token.key,
    enableLogging: true  // Optional: enable debug logging
});





import com.zc.component.datastream.ZCDatastream;
import com.zc.component.datastream.beans.ZCDataStreamsConfig;
import com.zc.component.datastream.websocket.ZCDataStreamsWebSocket;

ZCDatastream datastream = ZCDatastream.getInstance();

ZCDataStreamsConfig config = new ZCDataStreamsConfig(
    tokenResponse.getUrl(),
    tokenResponse.getWssId(),
    tokenResponse.getKey(),
    false  // enableLogging
);

ZCDataStreamsWebSocket socket = datastream.createWebSocketClient(config);





from zcatalyst_sdk.datastreams.websocket_client import DataStreamsWebSocket
from zcatalyst_sdk.types.datastreams import DataStreamsConfig

config = DataStreamsConfig(
    url=token["url"],
    zuid=token["wss-id"],
    key=token["key"],
    enable_logging=True
)

socket = DataStreamsWebSocket(config=config)




Note: In Python, the WebSocket connection is not opened until you call socket.connect() after registering your event listeners.

### Choose Your Subscriber Type

Before subscribing, decide which subscriber type fits your use case:

<table class="content-table">
<thead>
<tr>
<th>Type</th>
<th>Value</th>
<th>When to Use</th>
</tr>
</thead>
<tbody>
<tr>
<td>Live events only</td>
<td>'0'</td>
<td>Start receiving only NEW messages published after subscribing</td>
</tr>
<tr>
<td>Earliest available</td>
<td>'-1'</td>
<td>Receive ALL available messages from the earliest stored event (within 48-hour retention)</td>
</tr>
<tr>
<td>Resume</td>
<td>'-2'</td>
<td>Resume from where you last left off (or live events for new subscribers)</td>
</tr>
<tr>
<td>Specific Streaming ID</td>
<td>'<streaming-id>'</td>
<td>Resume from a specific message (use the streamingId from a previous message)</td>
</tr>
</tbody>
</table>

Note: Data retention is 48 hours. Events older than 48 hours are not available.

### Register Event Listeners and Subscribe




// Listen for connection opened
socket.on('open', () => {
    console.log('Connected! Subscribing to channel...');
    
    // Subscribe with your chosen subscriber type
    socket.subscribe('0');  // Live events only
    // socket.subscribe('-1');  // Earliest available
    // socket.subscribe('-2');  // Resume
    // socket.subscribe('16965000000027481');  // From specific streaming ID
});

// Listen for incoming messages
socket.on('message', (event) => {
    console.log('Received:', event.data);
    console.log('Streaming ID:', event.streamingId);
    
    // Process your message here
    // ...
    
    // CRITICAL: Always send acknowledgement to receive next message
    socket.sendAck();
});

// Listen for errors
socket.on('error', (err) => {
    console.error('Error:', err.code, err.message);
});

// Listen for connection closed
socket.on('close', () => {
    console.log('Connection closed');
});





import com.zc.component.datastream.websocket.ZCDataStreamsEventListener;
import com.zc.component.datastream.beans.ZCCustomEvent;
import com.zc.component.datastream.beans.ZCDataStreamMessageEvent;

socket.addEventListener(new ZCDataStreamsEventListener() {
    @Override
    public void onOpen(ZCCustomEvent event) {
        System.out.println("Connected! Subscribing to channel...");
        
        // Subscribe with your chosen subscriber type
        socket.subscribe("0");  // Live events only
        // socket.subscribe("-1");  // Earliest available
        // socket.subscribe("-2");  // Resume
    }
    
    @Override
    public void onMessage(ZCDataStreamMessageEvent event) {
        System.out.println("Received: " + event.getData());
        System.out.println("Streaming ID: " + event.getStreamingId());
        
        // Process your message here
        // ...
        
        // CRITICAL: Always send acknowledgement to receive next message
        socket.sendAck();
    }
    
    @Override
    public void onError(ZCCustomEvent event) {
        System.err.println("Error: " + event.getMessage());
    }
    
    @Override
    public void onClose(ZCCustomEvent event) {
        System.out.println("Connection closed");
    }
    
    @Override
    public void onPong(ZCCustomEvent event) {
        // Keep-alive pong received
    }
});





def on_open(event):
    print("Connected! Subscribing to channel...")
    
    # Subscribe with your chosen subscriber type
    socket.subscribe("0")  # Live events only
    # socket.subscribe("-1")  # Earliest available
    # socket.subscribe("-2")  # Resume

def on_message(event):
    print("Received:", event.data)
    print("Streaming ID:", event.streaming_id)
    
    # Process your message here
    # ...
    
    # CRITICAL: Always send acknowledgement to receive next message
    socket.send_ack()

def on_error(event):
    print("Error:", event.code, event.message)

def on_close(event):
    print("Connection closed")

# Register event listeners
socket.set_on_open(on_open)
socket.set_on_message(on_message)
socket.set_on_error(on_error)
socket.set_on_close(on_close)

# Connect (Python requires explicit connect call)
socket.connect()




### Handle Message Events

Each message event received contains:

<table class="content-table">
<thead>
<tr>
<th>Field</th>
<th>Type</th>
<th>Description</th>
</tr>
</thead>
<tbody>
<tr>
<td>operation</td>
<td>string</td>
<td>Event type: 'event' for inline data or 'api' for bulk data</td>
</tr>
<tr>
<td>streamingId</td>
<td>string</td>
<td>Unique identifier for this message in the stream</td>
</tr>
<tr>
<td>data</td>
<td>string</td>
<td>The published data (for inline events)</td>
</tr>
<tr>
<td>url</td>
<td>string</td>
<td>URL to fetch bulk data (for bulk/api events)</td>
</tr>
<tr>
<td>method</td>
<td>string</td>
<td>HTTP method for fetching bulk data (for bulk/api events)</td>
</tr>
</tbody>
</table>

### Example: Processing Different Event Types


socket.on('message', (event) => {
    if (event.operation === 'event') {
        // Inline data
        console.log('Data:', event.data);
        const parsedData = JSON.parse(event.data);
        // Process parsedData...
    } else if (event.operation === 'api') {
        // Bulk data - fetch from URL
        console.log('Bulk data available at:', event.url);
        // Fetch and process bulk data...
    }
    
    // Always acknowledge
    socket.sendAck();
});


### Send Acknowledgement

You must call sendAck() after processing each message.

- Until acknowledgement is sent, the next message will NOT be delivered
- This ensures ordered, reliable message delivery
- If you disconnect before acknowledging, you can reconnect using subscriber type '-2' to resume


// JavaScript
socket.sendAck();

// Java
socket.sendAck();

// Python
socket.send_ack()


### Unsubscribe and Disconnect

When you're done receiving data:




// Unsubscribe: ends the session. Generate a new token pair to subscribe again.
socket.unsubscribe();

// Or disconnect only: reconnect later and resume with subscribe type "-2".
socket.close();





// Unsubscribe: ends the session. Generate a new token pair to subscribe again.
socket.unsubscribe();

// Or disconnect only: reconnect later and resume with subscribe type "-2".
socket.close();





# Unsubscribe: ends the session. Generate a new token pair to subscribe again.
socket.unsubscribe()

# Or disconnect only: reconnect later and resume with subscribe type "-2".
socket.close()




### Example Implementation


const { DataStreamsWebSocket } = require('@zcatalyst/datastreams');

async function subscribeToChannel() {
    // 1. Get token from server
    const response = await fetch('https://your-server.com/get-token', {
        method: 'POST',
        body: JSON.stringify({ channelId: 'YOUR_CHANNEL_ID', userId: 'user123' })
    });
    const { token } = await response.json();
    
    // 2. Create WebSocket connection
    const socket = new DataStreamsWebSocket({
        url: token.url,
        zuid: token.wss_id,
        key: token.key,
        enableLogging: true
    });
    
    // 3. Register event listeners
    socket.on('open', () => {
        console.log('Connected!');
        socket.subscribe('0');  // Live events
    });
    
    socket.on('message', (event) => {
        console.log('Received:', event.data);
        
        // Process message
        const data = JSON.parse(event.data);
        updateUI(data);
        
        // Acknowledge
        socket.sendAck();
    });
    
    socket.on('error', (err) => {
        console.error('Error:', err);
    });
    
    socket.on('close', () => {
        console.log('Disconnected');
    });
}

subscribeToChannel();


The SDK manages the connection lifecycle automatically: opens the WebSocket connection, validates the token pair, subscribes when you call subscribe(), and pushes messages with the acknowledgement flow.

You can find further details on Subscriber Types, Message Event Structure, Acknowledgement, Connection Lifecycle, and Data Retention.


--------------------------------------------------------------------------------
title: "Test Data Flow from Console"
description: "Verify Data Streams channel setup by testing publishing and subscribing directly from the Catalyst Console without code."
last_updated: "2026-10-08T06:47:52.530Z"
source: "https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/implementation/test-data-flow-console/"
service: "Cloud Scale"
related:
- Create a Channel (/en/cloud-scale/help/data-streams/implementation/create-channel/)
- Publish Data with SDK (/en/cloud-scale/help/data-streams/implementation/publish-data-sdk/)
- Subscribe with WebSocket (/en/cloud-scale/help/data-streams/implementation/subscribe-websocket/)
- Key Concepts - Message Event Structure (/en/cloud-scale/help/data-streams/key-concepts/#message-event-structure)

--------------------------------------------------------------------------------


# Test Data Flow

After creating a channel or any time you open an existing channel from the Channels list page, you will be directed to the Test Data Flow section. This section enables you to verify how data is published and received through the channel before implementing it in your application.

Using this interface, you can simulate both a publisher and a subscriber directly from the Catalyst Console, without writing any code or deploying a function. 

Note: Data sent through the Test Data Flow section is treated as a real streaming event. It will be delivered to all active subscribers connected to the channel, including any subscriber connections established through the SDK or API.

The Test Data Flow section contains two primary actions: Send and Receive.

### Send Test Data

The Send tab allows you to simulate a publisher and send sample data to the channel. The data you send will be treated as a real streaming event and will be delivered to any active subscribers connected to the channel, including the test receiver in the console. 



To test publishing:

1. Navigate to the Send tab in the Test Data Flow section of your channel.
2. Enter the sample payload in the Enter Data Payload field.
3. Click Send to publish the data to the channel.



This allows you to verify how event data will be published by your application services. You can use this to confirm that the channel is correctly configured before integrating the publishing logic in your application.

### Receive Channel Data

The Receive tab allows you to simulate a subscriber and listen for incoming events from the channel.



To start receiving events:

1. Navigate to the Receive tab in the Test Data Flow section of your channel.
2. Click Activate Stream to establish a test subscription.
3. The console will begin listening for events published to the channel.



4. Any incoming data will appear in the Receive channel data panel in real time.



Each event displayed will include the following fields:

* data – the payload that was published to the channel.
* streamingId – the unique identifier assigned to that event in the stream.

These fields correspond to the message event structure that your application will receive when subscribing through the SDK.

### Event Acknowledgement

After receiving an event, you must click Send Acknowledgement to confirm that the message has been processed.

Data Streams delivers events in sequence, and the acknowledgement ensures that the next event in the stream can be delivered. Until the acknowledgement is sent, subsequent events will not be displayed in the receiver.



This behavior mirrors the sendAck() method used in the SDK. In your application code, you must call sendAck() after processing each message to receive the next one. Refer to the Subscribe with WebSocket guide for implementation details.

### Closing the Test Stream

You can stop the test receiver at any time by clicking Disconnect. Once disconnected, the test subscriber will no longer receive events from the channel. If you do not manually disconnect, the test connection will be closed automatically after two minutes.



Note: Clicking Disconnect only stops the test connection created in the console. It does not affect any subscriber connections established through the SDK or API. The test connection will also be closed automatically after two minutes if not manually disconnected.

### Set Up Data Streams

The right side of the Test Data Flow section provides example code snippets for sending or receiving events using the Catalyst SDK. Snippets are available in supported languages including Java, JavaScript, and Python. 

These examples help you implement the same publishing and subscribing behavior in your application.

For complete implementation details, refer to the SDK and API documentation.


--------------------------------------------------------------------------------
title: "Monitor Data Streams Channel Activity"
description: "Track live channel metrics, data transfer, triggers, and performance insights from the Data Streams Dashboard."
last_updated: "2026-10-08T06:47:52.531Z"
source: "https://docs.catalyst.zoho.com/en/cloud-scale/help/data-streams/implementation/monitor-channel-activity/"
service: "Cloud Scale"
related:
- Create a Channel (/en/cloud-scale/help/data-streams/implementation/create-channel/)
- Test Data Flow (/en/cloud-scale/help/data-streams/implementation/test-data-flow-console/)
- Dashboard Overview (/en/cloud-scale/help/data-streams/dashboard/)

--------------------------------------------------------------------------------


# Monitor Channel Activity

The Channel Dashboard provides real-time insights into your Data Streams channel's activity, data flow, and performance metrics. Before you begin, ensure you have a channel created with data flowing through it (publishers and/or subscribers active).

## Access the Dashboard

Navigate to your channel from the Channels list page. The dashboard is displayed by default when you open a channel.



## Live Channel Activity

Monitor real-time metrics to understand current channel performance.



### Metrics Available

<table class="content-table">
<thead>
<tr>
<th>Metric</th>
<th>Description</th>
</tr>
</thead>
<tbody>
<tr>
<td>Messages Streamed</td>
<td>Total number of messages streamed through the channel for the day</td>
</tr>
<tr>
<td>Live Subscribers</td>
<td>Number of subscribers currently connected and actively receiving data (real-time count)</td>
</tr>
</tbody>
</table>

### Programmatic Access

You can retrieve the live subscriber count programmatically:


// JavaScript
const liveCount = await datastreams.getLiveCount('YOUR_CHANNEL_ID');
console.log('Live subscribers:', liveCount);


Note: Refer to SDK documentation or REST API for implementation details.

## Configure Triggers

The **Triggers** section displays available methods for publishing data to your channel.



### Available Triggers

#### Catalyst Signals (Event-Driven)

Use Catalyst Signals to automatically publish data when events occur in connected services:
- Zoho CRM record updates
- Data Store row insertions
- Authentication user signups
- Custom service events

To configure:
1. Click **Go to Signals**
2. Configure Signals Publishers, Rules, and Targets
3. Route events to your Data Streams channel

Refer to the Publish Data with Signals guide for complete integration details.

#### Catalyst SDK (Manual Publishing)

Use the SDK to publish data from your application code:
- Backend services and functions
- Custom business logic
- Data processing workflows

To configure:
1. Click **Learn more** to view publishing guide
2. Implement publishing in your Advanced I/O function

Refer to the Publish Data with SDK guide for complete implementation details.

Note: Refer to Publishing Methods in Key Concepts to learn about all available publishing options.

## Analyze Data Transfer

The **Data Transfer** section visualizes data volume over time, helping you identify usage patterns and activity spikes.



### Select Time Period

Choose from preset ranges or define a custom period:



<table class="content-table">
<thead>
<tr>
<th>Time Period</th>
<th>Range</th>
<th>Use Case</th>
</tr>
</thead>
<tbody>
<tr>
<td>Today</td>
<td>12:00:00 AM → Current time</td>
<td>Monitor today's activity</td>
</tr>
<tr>
<td>Yesterday</td>
<td>Previous day (12:00:00 AM → 11:59:59 PM)</td>
<td>Compare with today</td>
</tr>
<tr>
<td>Last 7 days</td>
<td>Rolling 7-day window</td>
<td>Identify weekly trends</td>
</tr>
<tr>
<td>Last 30 days</td>
<td>Rolling 30-day window</td>
<td>Monitor monthly patterns</td>
</tr>
<tr>
<td>Custom Range</td>
<td>Up to 2 months</td>
<td>Specific date analysis</td>
</tr>
</tbody>
</table>

The chart displays **Inbound Data Size (KB)** over your selected time period. The X-axis shows time intervals, the Y-axis shows data volume in kilobytes, and you can hover over data points to view exact timestamps and data sizes.



Note: Inbound Data Size represents the total data received by the channel from publishers before distribution to subscribers.

## View General Details

The **General Details** section displays channel configuration and metadata.



### Available Information

<table class="content-table">
<thead>
<tr>
<th>Field</th>
<th>Description</th>
</tr>
</thead>
<tbody>
<tr>
<td>Channel ID</td>
<td>Unique identifier (required for SDK/API operations)</td>
</tr>
<tr>
<td>Created By</td>
<td>User who created the channel</td>
</tr>
<tr>
<td>Created On</td>
<td>Timestamp when the channel was created</td>
</tr>
<tr>
<td>Last Modified On</td>
<td>Timestamp of most recent configuration change</td>
</tr>
<tr>
<td>Control Type</td>
<td>Access control method (User Roles or Custom Logic)</td>
</tr>
<tr>
<td>Retention Time</td>
<td>Data retention period (48 hours)</td>
</tr>
</tbody>
</table>

### Data Retention Details

- **Retention Period**: 48 hours from publication
- **Subscriber Access**: Can receive events within retention period using subscriber types:
  - `'-1'` (earliest available)
  - `'-2'` (resume)
  - `'<streaming-id>'` (from specific point)
- **After 48 Hours**: Events are no longer available

Note: Learn more about Subscriber Types and Data Retention.

## Dashboard Actions

From the three-dot menu in the top-right corner:

- **Edit Channel** → Change channel name
- **Delete Channel** → Permanently remove channel

Note: For details on editing or deleting channels, refer to Manage Channels.