Stream and decode Data Streams reports via WebSocket using the TypeScript SDK

Guide Versions

This guide is available in multiple versions. Choose the one that matches your needs.

TypeScript SDK

Go SDK Rust SDK TypeScript SDK

In this tutorial, you'll learn how to use the Data Streams SDK for TypeScript to subscribe to real-time reports via a WebSocket connection. You'll set up your TypeScript project, listen for real-time reports from the Data Streams Aggregation Network, decode the report data, and log their attributes to your terminal.

Requirements

• Git: Make sure you have Git installed. You can check your current version by running git --version in your terminal and download the latest version from the official Git website if necessary.
• Node.js: Make sure you have Node.js 20.0 or higher. You can check your current version by running node --version in your terminal and download the latest version from the official Node.js website if necessary.
• TypeScript: Make sure you have TypeScript 5.3 or higher. You can check your current version by running npx tsc --version in your terminal and install or update TypeScript by running npm install -g typescript if necessary.
• API Credentials: Access to Data Streams requires API credentials. If you haven't already, contact us to request mainnet or testnet access.

Tutorial

First, you'll set up a basic TypeScript project, installing the SDK and pasting example code. This will let you stream reports for streams, logging their attributes to your terminal.

Set up your TypeScript project

1. Create a new directory for your project and navigate to it:

   mkdir my-data-streams-project
   cd my-data-streams-project
   

2. Initialize a new Node.js project:

   npm init -y
   

3. Install the TypeScript SDK and other required packages:

    npm install @chainlink/data-streams-sdk dotenv
    npm install -D tsx
   

4. Set your API credentials:

   Option 1 - Environment variables:

   export API_KEY="your_api_key_here"
   export USER_SECRET="your_user_secret_here"
   

   Option 2 - .env file:

   # Create .env file
   touch .env
   
   # Add your credentials
   API_KEY="your_api_key_here"
   USER_SECRET="your_user_secret_here"
   

Establish a WebSocket connection and listen for real-time reports

1. Create a new TypeScript file, stream.ts, in your project directory:

   touch stream.ts
   

2. Insert the following code example and save your stream.ts file:

   import { createClient, LogLevel, decodeReport, getReportVersion, formatReport } from "@chainlink/data-streams-sdk"
   import "dotenv/config"
   
   async function main() {
     if (process.argv.length < 3) {
       console.error("Please provide one or more feed IDs as arguments")
       console.error("\nExamples:")
       console.error("  Single feed:")
       console.error("    npx tsx stream.ts 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782")
       console.error("  Multiple feeds:")
       console.error(
         "    npx tsx stream.ts 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782,0x00036fe43f87884450b4c7e093cd5ed99cac6640d8c2000e6afc02c8838d0265"
       )
       process.exit(1)
     }
   
     const feedIds = process.argv[2].split(",")
   
     console.log("Chainlink Data Streams - Report Streaming")
     console.log("=".repeat(60))
     console.log(`📊 Feeds: ${feedIds.length} feed(s)`)
     console.log("=".repeat(60))
   
     try {
       const client = createClient({
         apiKey: process.env.API_KEY || "YOUR_API_KEY",
         userSecret: process.env.USER_SECRET || "YOUR_USER_SECRET",
         endpoint: "https://api.testnet-dataengine.chain.link",
         wsEndpoint: "wss://ws.testnet-dataengine.chain.link",
   
         // Comment to disable SDK logging:
         logging: {
           logger: console,
           logLevel: LogLevel.INFO,
           enableConnectionDebug: false, // Enable WebSocket ping/pong and connection state logs (logLevel should be DEBUG)
         },
       })
   
       let reportCount = 0
   
       // Create stream with custom options
       const stream = client.createStream(feedIds, {
         maxReconnectAttempts: 10,
         reconnectInterval: 3000,
       })
   
       // Event: Process incoming reports
       stream.on("report", (report) => {
         reportCount++
   
         try {
           console.log(`\n📈 Report #${reportCount} - ${new Date().toISOString()}`)
   
           // Show raw report blob
           console.log(`\nRaw Report Blob: ${report.fullReport}`)
   
           // Decode the report
           const decodedData = decodeReport(report.fullReport, report.feedID)
           const version = getReportVersion(report.feedID)
   
           // Combine decoded data with report metadata
           const decodedReport = {
             ...decodedData,
             feedID: report.feedID,
             validFromTimestamp: report.validFromTimestamp,
             observationsTimestamp: report.observationsTimestamp,
           }
   
           console.log(formatReport(decodedReport, version))
         } catch (error) {
           console.error(`❌ Error processing report: ${error instanceof Error ? error.message : error}`)
         }
   
         // Display stats every 5 reports
         if (reportCount % 5 === 0) {
           const stats = stream.getMetrics()
           console.log(
             `\n📊 Stats: ${stats.accepted} reports | ${stats.activeConnections}/${stats.configuredConnections} connections`
           )
         }
       })
   
       // Event: Handle errors
       stream.on("error", (error) => {
         console.error(`\n❌ Error: ${error.message}`)
   
         if (error.message.includes("authentication")) {
           console.error("💡 Check your API_KEY and USER_SECRET environment variables")
         }
       })
   
       // Event: Handle disconnections
       stream.on("disconnected", () => {
         console.log("\n🔴 Stream disconnected - reconnecting...")
       })
   
       // Event: Monitor reconnections
       stream.on("reconnecting", (info: { attempt: number; delayMs: number; origin?: string; host?: string }) => {
         console.log(
           `🔄 Reconnecting... attempt ${info.attempt} in ~${info.delayMs}ms${info.host ? ` (${info.host})` : ""}`
         )
       })
   
       console.log("⏳ Connecting...\n")
       await stream.connect()
       console.log("✅ Connected! Listening for reports...\n")
   
       // Graceful shutdown
       const shutdown = async () => {
         console.log("\n🛑 Shutting down...")
         await stream.close()
         console.log("✅ Shutdown complete")
         process.exit(0)
       }
   
       process.on("SIGINT", shutdown)
       process.on("SIGTERM", shutdown)
     } catch (error) {
       console.error("❌ Failed to start stream:", error instanceof Error ? error.message : error)
       process.exit(1)
     }
   }
   
   main()
   

3. Subscribe to a testnet crypto stream. The below example executes the application, subscribing to the ETH/USD crypto stream:

   npx tsx stream.ts 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782
   

   Expect output similar to the following in your terminal:

   Chainlink Data Streams - Report Streaming
   ============================================================
   📊 Feeds: 1 feed(s)
   ============================================================
   ⏳ Connecting...
   
   [2025-09-23T01:15:23.456Z] [DataStreams] Data Streams client initialized
   [2025-09-23T01:15:23.789Z] [DataStreams] WebSocket connected
   ✅ Connected! Listening for reports...
   
   📈 Report #1 - 2025-09-23T01:15:24.123Z
   
   Raw Report Blob: 0x00090d9e8d96765a0c49e03a6ae05c82e8f8de70cf179baa632f18313e54bd690000000000000000000000000000000000000000000000000000000001f71291000000000000000000000000000000000000000000000000000000030000000100000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028001000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba7820000000000000000000000000000000000000000000000000000000068d1efd30000000000000000000000000000000000000000000000000000000068d1efd30000000000000000000000000000000000000000000000000000455f50d8377000000000000000000000000000000000000000000000000000349983c7f97bd70000000000000000000000000000000000000000000000000000000068f97cd30000000000000000000000000000000000000000000000e36d996215f00840000000000000000000000000000000000000000000000000e36835d6045c9700000000000000000000000000000000000000000000000000e36fabf5a9955e40000000000000000000000000000000000000000000000000000000000000000002d2fe013d18dcdc7d08d61c79d31ff7067487d86b6eb64b1b6c6010d31038b46d53e9324e546ea84b16d7f1b5ace563611ce5f758d11780d51ca737dade8a49dd
   
   Report Metadata:
   Feed ID: 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782
   Valid From: 1758588883
   Observations: 1758588883
   
   Decoded Data:
   Native Fee: 76275680556912
   LINK Fee: 14805490063735767
   Expires At: 1761180883
   Price: 4195308356000000000000
   Bid Price: 4194920048000000000000
   Ask Price: 4195457700000000000000
   --------------------------------------------------
   
   📈 Report #2 - 2025-09-23T01:15:25.234Z
   
   Raw Report Blob: 0x00090d9e8d96765a0c49e03a6ae05c82e8f8de70cf179baa632f18313e54bd690000000000000000000000000000000000000000000000000000000001f71292000000000000000000000000000000000000000000000000000000030000000100000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028001000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba7820000000000000000000000000000000000000000000000000000000068d1efd40000000000000000000000000000000000000000000000000000000068d1efd40000000000000000000000000000000000000000000000000000455f45d2349000000000000000000000000000000000000000000000000000349968c9f98bd70000000000000000000000000000000000000000000000000000000068f97cd40000000000000000000000000000000000000000000000e36d8b5b2b6084000000000000000000000000000000000000000000000000000e36824f6b8c970000000000000000000000000000000000000000000000000e36fc15a99955e400000000000000000000000000000000000000000000000000000000000000000028f3c0139b8c7d08d61c79d31ff7067487d86b6eb64b1b6c6010d31038b46d53e932
   
   Report Metadata:
   Feed ID: 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782
   Valid From: 1758588884
   Observations: 1758588884
   
   Decoded Data:
   Native Fee: 76275680556912
   LINK Fee: 14805490063735767
   Expires At: 1761180884
   Price: 4195298740000000000000
   Bid Price: 4194910430000000000000
   Ask Price: 4195448050000000000000
   --------------------------------------------------
   
   📊 Stats: 5 reports | 1/1 connections
   
   [...]
   

   Your application has successfully subscribed to the report data.

   Learn more about the decoded report details.

Decoded report details

The decoded report details include:

For descriptions and data types of other report schemas, see the Report Schema Overview.

High Availability (HA) mode

High Availability (HA) mode creates multiple WebSocket connections to different origin endpoints for improved reliability. When enabled, the SDK automatically handles failover, deduplicates reports, and provides connection-level metrics.

Enabling HA mode

To enable HA mode in your streaming application, make these changes to the basic example:

1. Add haMode: true to the client configuration. You also must use a mainnet endpoint, as HA mode is not currently supported on testnet.

const client = createClient({
  apiKey: process.env.API_KEY || "YOUR_API_KEY",
  userSecret: process.env.USER_SECRET || "YOUR_USER_SECRET",
  endpoint: "https://api.dataengine.chain.link", // Mainnet endpoint
  wsEndpoint: "wss://ws.dataengine.chain.link", // Mainnet WebSocket
  haMode: true, // Enable High Availability mode

  // Optional: Monitor connection status changes
  connectionStatusCallback: (isConnected: boolean, host: string, origin?: string) => {
    const status = isConnected ? "🟢 Connected" : "🔴 Disconnected"
    console.log(`${status}: ${host}${origin ? ` (${origin})` : ""}`)
  },

  logging: {
    logger: console,
    logLevel: LogLevel.INFO,
  },
})

2. Optionally, monitor HA metrics using the existing getMetrics() calls in your report handler:

stream.on("report", (report) => {
  // ... your report processing logic

  // HA-specific metrics
  const stats = stream.getMetrics()
  console.log(
    `📊 HA Stats: ${stats.accepted} reports | ${stats.activeConnections}/${stats.configuredConnections} connections`
  )
  console.log(`Reconnects: ${stats.partialReconnects} partial, ${stats.fullReconnects} full`)

  // Check deduplication (important for HA mode)
  if (stats.totalReceived > 0) {
    const deduplicationRate = ((stats.deduplicated / stats.totalReceived) * 100).toFixed(1)
    console.log(`📈 Deduplication: ${deduplicationRate}% (${stats.deduplicated} filtered)`)
  }
})

When haMode is true, the SDK automatically discovers multiple origin endpoints behind the single URL and establishes separate connections to each origin.

The optional connectionStatusCallback can be used to integrate with external monitoring systems. The SDK already provides comprehensive connection logs, so this callback is primarily useful for custom alerting or metrics collection.

Payload for onchain verification

In this tutorial, you logged and decoded the full_report payloads to extract the report data. However, in a production environment, you should verify the data to ensure its integrity and authenticity.

Refer to the Verify report data onchain tutorial to learn more.

Explanation

Establishing a WebSocket connection and listening for reports

Your application uses the createClient function from the Data Streams SDK to create a client, then uses client.createStream() to establish a real-time WebSocket connection with the Data Streams Aggregation Network.

Once the WebSocket connection is established, your application subscribes to one or more streams by passing an array of feed IDs to the createStream function. This subscription lets the client receive real-time updates whenever new report data is available for the specified streams.

Fore further reference, see the WebSocket Interface section of the SDK Reference.

Event-driven streaming

The TypeScript SDK uses an event-driven approach for handling streaming data:

• Connection events: The stream emits connected, disconnected, and reconnecting events to track connection status.
• Report events: When new reports arrive, the report event is triggered with the decoded report data.
• Error handling: The stream emits error events for any issues that occur during streaming.
• Metrics: The getMetrics() method provides real-time statistics about the stream performance.

Decoding and processing reports

As data reports arrive via the established WebSocket connection, they are processed in real-time:

• Automatic decoding: The SDK's decodeReport function automatically detects the report version and parses the raw data into a structured format.
• Real-time processing: Each received report triggers the report event handler, where you can process, log, or store the decoded data.
• Error resilience: Individual report processing errors don't interrupt the stream, allowing continuous operation.

Handling the decoded data

In this example, the application logs the structured report data to the terminal. However, this data can be used for further processing, analysis, or display in your own application. The decoded data includes essential information such as benchmark prices, bid/ask spreads, and fee data for onchain verification.

For more information about SDK streaming configuration and advanced options, see the SDK Reference.