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

Guide Versions

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

Rust SDK

Go SDKRust SDKTypeScript SDK

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

Requirements

• Rust: Make sure you have Rust installed. You can install Rust by following the instructions on the official Rust website.
• API Credentials: Access to Data Streams requires API credentials. If you haven't already, contact us to request mainnet or testnet access.

Tutorial

Set up your Rust 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 Rust project:

   cargo init
   

3. Add the following dependencies to your Cargo.toml file:

   [dependencies]
   chainlink-data-streams-sdk = "1.0.3"
   chainlink-data-streams-report = "1.0.3"
   tokio = { version = "1.4", features = ["full"] }
   hex = "0.4"
   tracing = "0.1"
   tracing-subscriber = { version = "0.3", features = ["time"] }
   

   Note: The tracing feature is required for logging functionality.

Establish a WebSocket connection and listen for real-time reports

1. Replace the contents of src/main.rs with the following code:

   use chainlink_data_streams_report::feed_id::ID;
   // NOTE: Use the report version (v3, v8, etc.) that matches your stream
   use chainlink_data_streams_report::report::{ decode_full_report, v3::ReportDataV3 };
   use chainlink_data_streams_sdk::config::Config;
   use chainlink_data_streams_sdk::stream::Stream;
   use std::env;
   use std::error::Error;
   use tracing::{ info, warn };
   use tracing_subscriber::fmt::time::UtcTime;
   
   #[tokio::main]
   async fn main() -> Result<(), Box<dyn Error>> {
      // Initialize logging with UTC timestamps
      tracing_subscriber
         ::fmt()
         .with_timer(UtcTime::rfc_3339())
         .with_max_level(tracing::Level::INFO)
         .init();
   
      // Get feed IDs from command line arguments
      let args: Vec<String> = env::args().collect();
      if args.len() < 2 {
         eprintln!("Usage: cargo run [StreamID1] [StreamID2] ...");
         std::process::exit(1);
      }
   
      // Get API credentials from environment variables
      let api_key = env::var("API_KEY").expect("API_KEY must be set");
      let api_secret = env::var("API_SECRET").expect("API_SECRET must be set");
   
      // Parse feed IDs from command line arguments
      let mut feed_ids = Vec::new();
      for arg in args.iter().skip(1) {
         let feed_id = ID::from_hex_str(arg)?;
         feed_ids.push(feed_id);
      }
   
      // Initialize the configuration
      let config = Config::new(
         api_key,
         api_secret,
         "https://api.testnet-dataengine.chain.link".to_string(),
         "wss://ws.testnet-dataengine.chain.link".to_string()
      ).build()?;
   
      // Create and initialize the stream
      let mut stream = Stream::new(&config, feed_ids).await?;
      stream.listen().await?;
   
      info!("WebSocket connection established. Listening for reports...");
   
      // Process incoming reports
      loop {
         match stream.read().await {
               Ok(response) => {
                  info!("\nRaw report data: {:?}\n", response.report);
   
                  // Decode the report
                  let full_report = hex::decode(&response.report.full_report[2..])?;
                  let (_report_context, report_blob) = decode_full_report(&full_report)?;
                  // NOTE: Use the report version (v3, v8, etc.) that matches your stream
                  let report_data = ReportDataV3::decode(&report_blob)?;
   
                  // Print decoded report details
                  // NOTE: Adjust for your report and desired output
                  info!(
                     "\n--- Report Stream ID: {} ---\n\
                        ------------------------------------------\n\
                        Observations Timestamp : {}\n\
                        Price                 : {}\n\
                        Bid                   : {}\n\
                        Ask                   : {}\n\
                        Valid From Timestamp  : {}\n\
                        Expires At           : {}\n\
                        Link Fee             : {}\n\
                        Native Fee           : {}\n\
                        ------------------------------------------",
                     response.report.feed_id.to_hex_string(),
                     response.report.observations_timestamp,
                     report_data.benchmark_price,
                     report_data.bid,
                     report_data.ask,
                     response.report.valid_from_timestamp,
                     report_data.expires_at,
                     report_data.link_fee,
                     report_data.native_fee
                  );
   
                  // Print stream stats
                  info!(
                     "\n--- Stream Stats ---\n{:#?}\n\
                        --------------------------------------------------------------------------------------------------------------------------------------------",
                     stream.get_stats()
                  );
               }
               Err(e) => {
                  warn!("Error reading from stream: {:?}", e);
               }
         }
      }
   
      // Note: In a production environment, you should implement proper cleanup
      // by calling stream.close() when the application is terminated.
      // For example:
      //
      // tokio::select! {
      //     _ = tokio::signal::ctrl_c() => {
      //         info!("Received shutdown signal");
      //         stream.close().await?;
      //     }
      //     result = stream.read() => {
      //         // Process result
      //     }
      // }
   }
   

2. Set up your API credentials as environment variables:

   export API_KEY="<YOUR_API_KEY>"
   export API_SECRET="<YOUR_API_SECRET>"
   

   Replace <YOUR_API_KEY> and <YOUR_API_SECRET> with your API credentials.

   The Rust code sample reads these environment variables using std::env::var("API_KEY") and std::env::var("API_SECRET") when building the client configuration:

   let api_key = std::env::var("API_KEY").expect("API_KEY must be set");
   let api_secret = std::env::var("API_SECRET").expect("API_SECRET must be set");
   
   // Initialize the configuration
   let config = Config::new(
      api_key,
      api_secret,
      "https://api.testnet-dataengine.chain.link".to_string(),
      "wss://api.testnet-dataengine.chain.link/ws".to_string()
   ).build()?;
   

   This configuration also specifies the rest_url, which is the base URL for the API, along with the WebSocket endpoint for subscribing to a streamed data report. In this example, both are set to the testnet URLs for Data Streams.

   See the Rust SDK Reference page for more configuration options.

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

   cargo run -- 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782
   

   Expect output similar to the following in your terminal:

   2024-12-13T23:07:56.463719Z  INFO my_data_streams_project: WebSocket connection established. Listening for reports...
   2024-12-13T23:07:56.463824Z  INFO data_streams_sdk::stream::monitor_connection: Received ping: [49]
   2024-12-13T23:07:56.463868Z  INFO data_streams_sdk::stream::monitor_connection: Responding with pong: [49]
   2024-12-13T23:07:57.060504Z  INFO data_streams_sdk::stream::monitor_connection: Received new report from Data Streams Endpoint.
   2024-12-13T23:07:57.061078Z  INFO my_data_streams_project:
   Raw report data: Report { feed_id: 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782, valid_from_timestamp: 1734131277, observations_timestamp: 1734131277, full_report: "0x0006f9b553e393ced311551efd30d1decedb63d76ad41737462e2cdbbdff1578000000000000000000000000000000000000000000000000000000004f5ac90d000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028001010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba78200000000000000000000000000000000000000000000000000000000675cbe4d00000000000000000000000000000000000000000000000000000000675cbe4d000000000000000000000000000000000000000000000000000017489b06e8fc000000000000000000000000000000000000000000000000000c7badb615bd1400000000000000000000000000000000000000000000000000000000675e0fcd0000000000000000000000000000000000000000000000d3c0d34ca0d14d85600000000000000000000000000000000000000000000000d3bda64c97c9f3a3a00000000000000000000000000000000000000000000000d3c1a08e0cffd77690000000000000000000000000000000000000000000000000000000000000000238102110cad488ecf151a17276fcfad6ef1f05593edfe80f6823b729416f826972ba32d085525b1d7ab79e6ae8188928c86051a4fc75f500bffabda2acd1d1f900000000000000000000000000000000000000000000000000000000000000024dddbc660abf75c30cb3c2aa375c87d228b2ee8735e339f59c5214897c0b89af39a7602df754364cce029f6eb7699ee02ffded96d0c46b5919e81ee4f650d1cb" }
   
   2024-12-13T23:07:57.062344Z  INFO my_data_streams_project:
   --- Report Stream ID: 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 ---
   ------------------------------------------
   Observations Timestamp : 1734131277
   Price                 : 3906157533081673500000
   Bid                   : 3905928693886829700000
   Ask                   : 3906215307384792250000
   Valid From Timestamp  : 1734131277
   Expires At           : 1734217677
   Link Fee             : 3513685734964500
   Native Fee           : 25600606005500
   ------------------------------------------
   2024-12-13T23:07:57.062489Z  INFO my_data_streams_project:
   --- Stream Stats ---
   StatsSnapshot {
      accepted: 1,
      deduplicated: 0,
      total_received: 1,
      partial_reconnects: 0,
      full_reconnects: 0,
      configured_connections: 1,
      active_connections: 1,
   }
   --------------------------------------------------------------------------------------------------------------------------------------------
   2024-12-13T23:07:58.065686Z  INFO data_streams_sdk::stream::monitor_connection: Received new report from Data Streams Endpoint.
   2024-12-13T23:07:58.066315Z  INFO my_data_streams_project:
   Raw report data: Report { feed_id: 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782, valid_from_timestamp: 1734131278, observations_timestamp: 1734131278, full_report: "0x0006f9b553e393ced311551efd30d1decedb63d76ad41737462e2cdbbdff1578000000000000000000000000000000000000000000000000000000004f5ac911000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028000010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba78200000000000000000000000000000000000000000000000000000000675cbe4e00000000000000000000000000000000000000000000000000000000675cbe4e00000000000000000000000000000000000000000000000000001748ee300af4000000000000000000000000000000000000000000000000000c7b8b51304fbc00000000000000000000000000000000000000000000000000000000675e0fce0000000000000000000000000000000000000000000000d3bddf08d0b10a28e00000000000000000000000000000000000000000000000d3bb84af9f92f963c00000000000000000000000000000000000000000000000d3bf1d6bf14e501fc000000000000000000000000000000000000000000000000000000000000000021402b6b82c20826315384d74b3235b95f136ac65bba8c9e97c24d786e499894f298b51ae4aeba55cce0f85f2463e49e0a5e001b9a66f5b7b91e8be37d81d6cc5000000000000000000000000000000000000000000000000000000000000000217895cb599abc88d7b695edafed5ca5a5fc970f079b48bc2218888eec1fcccb0430c1ba2aa13b0d10f6c6b19a43cdb770029f4fb5804b0e2ef5ba3e73ca710f8" }
   
   2024-12-13T23:07:58.067395Z  INFO my_data_streams_project:
   --- Report Stream ID: 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 ---
   ------------------------------------------
   Observations Timestamp : 1734131278
   Price                 : 3905944663438106700000
   Bid                   : 3905775117434634200000
   Ask                   : 3906034281472429400000
   Valid From Timestamp  : 1734131278
   Expires At           : 1734217678
   Link Fee             : 3513538013319100
   Native Fee           : 25602001210100
   ------------------------------------------
   2024-12-13T23:07:58.067633Z  INFO my_data_streams_project:
   --- Stream Stats ---
   StatsSnapshot {
      accepted: 2,
      deduplicated: 0,
      total_received: 2,
      partial_reconnects: 0,
      full_reconnects: 0,
      configured_connections: 1,
      active_connections: 1,
   }
   
   [...]
   

   Your application has successfully subscribed to the report data.

   Learn more about the decoded report details.

Subscribing to multiple streams

You can subscribe to multiple streams by providing additional stream IDs as command-line arguments:

cargo run -- \
  0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 \
  0x00037da06d56d083fe599397a4769a042d63aa73dc4ef57709d31e9971a5b439

This will subscribe to both ETH/USD and BTC/USD streams.

High Availability (HA) mode

The example above demonstrates streaming data from a single crypto stream. For production environments, especially when subscribing to multiple streams, it's recommended to enable High Availability (HA) mode.

High Availability (HA) mode creates multiple WebSocket connections to different origin endpoints for improved reliability. When HA mode is enabled, the Stream will maintain at least 2 concurrent connections to different instances to ensure high availability, fault tolerance and minimize the risk of report gaps.

Enabling HA mode

To enable HA mode in your streaming application, make these changes to the basic example. You also must use a mainnet endpoint, as HA mode is not currently supported on testnet.

// ... existing code ...

use chainlink_data_streams_sdk::config::{Config, WebSocketHighAvailability}; // Import WebSocketHighAvailability

// ... existing code ...

// Initialize the configuration with HA mode
let config = Config::new(
    api_key,
    api_secret,
    "https://api.dataengine.chain.link".to_string(), // Mainnet endpoint
    "wss://ws.dataengine.chain.link".to_string(), // Mainnet WebSocket endpoint
)
.with_ws_ha(WebSocketHighAvailability::Enabled) // Enable WebSocket High Availability Mode
.build()?;


// ... existing code ...

When WebSocketHighAvailability::Enabled is set, the SDK automatically discovers multiple origin endpoints behind the single URL and establishes separate connections to each origin.

See more details about HA mode in the SDK Reference.

Decoded report details

The decoded report details include:

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

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.

Adapting code for different report schema versions

When working with different versions of Data Stream reports, you'll need to adapt your code to handle the specific report schema version they use:

1. Import the correct schema version module. Examples:

   • For v3 schema (as used in this example):

      use chainlink_data_streams_report::report::{ decode_full_report, v3::ReportDataV3 };
     

   • For v8 schema:

     use chainlink_data_streams_report::report::{ decode_full_report, v8::ReportDataV8 };
     

2. Update the decode function to use the correct schema version. Examples:

   • For v3 schema (as used in this example):

     let report_data = ReportDataV3::decode(&report_blob)?;
     

   • For v8 schema:

     let report_data = ReportDataV8::decode(&report_blob)?;
     

3. Access fields according to the schema version structure. Refer to the Report Schemas documentation for complete field references for each version.

Explanation

Establishing a WebSocket connection and listening for reports

The WebSocket connection is established in two steps:

1. Stream::new initializes a new stream instance with your configuration and feed IDs. This function prepares the connection parameters but doesn't establish the connection yet.

2. stream.listen() establishes the actual WebSocket connection and starts the background tasks that maintain the connection. These tasks handle:

   • Automatic reconnection if the connection is lost
   • Ping/pong messages to keep the connection alive
   • Message queueing and delivery

Decoding a report

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

1. Reading streams: The read method on the Stream object is called within a loop. This asynchronous method:

   • Awaits the next report from the WebSocket connection
   • Handles backpressure automatically
   • Returns a WebSocketReport containing the report data

2. Decoding reports: Each report is decoded in two stages:

   • decode_full_report parses the raw hexadecimal data, separating the report context (containing metadata) from the report blob
   • ReportDataV3::decode transforms the report blob into a structured format containing:
     • The benchmark price
     • Bid and ask prices for liquidity-weighted pricing
     • Fee information for onchain verification
     • Timestamp information

Handling the decoded data

The example demonstrates several best practices for handling the decoded data:

1. Logging:

   • Uses the tracing crate for structured logging
   • Configures UTC timestamps for consistent time representation
   • Includes both raw report data and decoded fields for debugging

2. Error handling:

   • Uses Rust's Result type for robust error handling
   • Implements the ? operator for clean error propagation
   • Logs errors with appropriate context using warn! macro

3. Stream monitoring:

   • Tracks stream statistics through get_stats()
   • Monitors connection status and reconnection attempts
   • Reports message acceptance and deduplication counts

The decoded data can be used for further processing, analysis, or display in your application. For production environments, it's recommended to verify the data onchain using the provided full_report payload.