Stream and decode reports via WebSocket

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

Before you begin

Go Version: Make sure you have Go version 1.21 or higher. You can check your current version by running go version in your terminal and download the latest version from the official Go website if necessary.
API Credentials: Access to the Streams Direct implementation requires API credentials. If you haven't already, contact us to talk to an expert about integrating Chainlink Data Streams with your applications.

Tutorial

Set up your Go project

Create and navigate to your project directory:

mkdir data-streams-direct-ws && cd data-streams-direct-ws

Initialize a Go module:
```
go mod init data-streams-direct-ws
```

Open the go.mod file at the root of your project directory and include the necessary module and package information:

module data-streams-direct-ws

go 1.21

require (
	github.com/ethereum/go-ethereum v1.12.2 // Ethereum blockchain interaction library
	github.com/gorilla/websocket v1.5.0 // Websocket library
	github.com/pkg/errors v0.9.1 // Library for handling errors
	github.com/smartcontractkit/chainlink/v2 v2.2.1-0.20230823171354-1ead9ee6f6bb // Chainlink core components library
)

replace (
	// Resolves version mismatch between cosmosSDK and hdevalence/ed25519consensus
	filippo.io/edwards25519 => filippo.io/edwards25519 v1.0.0-rc.1

	// Adds ARM support by updating CosmWasm to v1.2.4
	github.com/CosmWasm/wasmvm => github.com/CosmWasm/wasmvm v1.2.4

	//// Fix go mod tidy issue for ambiguous imports from go-ethereum
	//// See https://github.com/ugorji/go/issues/279
	github.com/btcsuite/btcd => github.com/btcsuite/btcd v0.22.1

	// Aligns protobuf version with cosmos SDK requirements
	github.com/gogo/protobuf => github.com/regen-network/protobuf v1.3.3-alpha.regen.1
)

Create client and internal subfolders within your project directory:
```
mkdir client internal
```
Create a clientWs.go file in the client subfolder and a decoder.go file in the internal subfolder:
```
touch client/clientWs.go
touch internal/decoder.go
```

Insert the provided code into clientWs.go and decoder.go to allow your application to establish a WebSocket connection, listen for new reports, and decode them:

// decoder.go

package internal

import (
	"encoding/binary"
	"fmt"

	"github.com/ethereum/go-ethereum/accounts/abi"
	"github.com/pkg/errors"
	mercuryutils "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/mercury/utils"
	v3report "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/mercury/v3/types"
)

type ReportWithContext struct {
	FeedId      mercuryutils.FeedID
	FeedVersion mercuryutils.FeedVersion
	V3Report    *v3report.Report
	Round       uint8
	Epoch       uint32
	Digest      []byte
}

type FullReport struct {
	ReportContext [3][32]byte
	ReportBlob    []byte
	RawRs         [][32]byte
	RawSs         [][32]byte
	RawVs         [32]byte
}

func mustNewType(t string) abi.Type {
	result, err := abi.NewType(t, "", []abi.ArgumentMarshaling{})
	if err != nil {
		panic(fmt.Sprintf("Unexpected error during abi.NewType: %s", err))
	}
	return result
}

var schema = abi.Arguments{
	{Name: "reportContext", Type: mustNewType("bytes32[3]")},
	{Name: "reportBlob", Type: mustNewType("bytes")},
	{Name: "rawRs", Type: mustNewType("bytes32[]")},
	{Name: "rawSs", Type: mustNewType("bytes32[]")},
	{Name: "rawVs", Type: mustNewType("bytes32")},
}

/*
	DecodeFullReport reads the "fullReport" from the API response into a struct containing the report context, report data,
	and raw signatures. This functions requires no prep to use, because the schema for the "fullReport" blob is
	common among all report versions (basic, premium, etc),
*/

func DecodeFullReport(fullReport []byte) (*FullReport, error) {
	values, err := schema.Unpack(fullReport)
	if err != nil {
		return nil, fmt.Errorf("failed to decode FullReport: %w", err)
	}
	decoded := new(FullReport)
	if err = schema.Copy(decoded, values); err != nil {
		return nil, fmt.Errorf("failed to copy FullReport values to struct: %w", err)
	}

	return decoded, nil
}

/*
DecodeReportData takes the report blob (FullReport.ReportBlob), extracts the feeds id, calculates the version from the feed id,
and finally decodes the report blob using the lib that correlates with the version. The resulting interface can be cast into
the correct report type as needed.
*/
func DecodeReportData(reportBlob []byte) (mercuryutils.FeedID, interface{}, error) {
	feedIdAbi := abi.Arguments{
		{Name: "feedId", Type: mustNewType("bytes32")},
	}
	reportElements := map[string]interface{}{}
	if err := feedIdAbi.UnpackIntoMap(reportElements, reportBlob); err != nil {
		return mercuryutils.FeedID{}, nil, err
	}
	feedIdInterface, ok := reportElements["feedId"]
	if !ok {
		return mercuryutils.FeedID{}, nil, errors.Errorf("unpacked ReportBlob has no 'feedId'")
	}
	feedIdBytes, ok := feedIdInterface.([32]byte)
	if !ok {
		return mercuryutils.FeedID{}, nil, errors.Errorf("cannot cast ReportBlob feedId to [32]byte, type is %T", feedIdBytes)
	}
	feedID := mercuryutils.FeedID(feedIdBytes)

	switch feedID.Version() {
	case mercuryutils.REPORT_V3:
		res, err := v3report.Decode(reportBlob)
		return feedID, res, err
	default:
		return mercuryutils.FeedID{}, nil, errors.Errorf("unknown report version %d", feedID.Version())
	}
}

/*
DecodeFullReportAndReportData takes the full report payload, decodes the fullReport blob, and then decodes the report data.
*/
func DecodeFullReportAndReportData(payload []byte) (*ReportWithContext, error) {
	fullReport, err := DecodeFullReport(payload)
	if err != nil {
		return nil, err
	}

	feedID, report, err := DecodeReportData(fullReport.ReportBlob)
	if err != nil {
		return nil, err
	}

	result := &ReportWithContext{
		FeedId:      feedID,
		FeedVersion: feedID.Version(),
		Digest:      fullReport.ReportContext[0][:],
		Round:       fullReport.ReportContext[1][31],
		Epoch:       binary.BigEndian.Uint32(fullReport.ReportContext[1][32-5 : 32-1]),
	}

	switch feedID.Version() {
	case mercuryutils.REPORT_V3:
		result.V3Report = report.(*v3report.Report)
	default:
		return nil, errors.Errorf("unknown report version %d", feedID.Version())
	}

	return result, nil
}

// clientWs.go

package client

import (
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/ethereum/go-ethereum/common/hexutil"
	"github.com/gorilla/websocket"
)

// Constants for ping interval, pong timeout, write timeout, and retry backoff
const pingInterval = 5 * time.Second
const pongTimeout = 10 * time.Second
const writeTimeout = 5 * time.Second

// NewReportWSMessage is the struct that we'll send to a subscribed client's MessageChan to be forwarded to the client's websocket connection
type NewReportWSMessage struct {
	Report struct {
		FeedId     hexutil.Bytes `json:"feedID"`
		FullReport hexutil.Bytes `json:"fullReport"`
	} `json:"report"`
}

const (
	wsPath = "/api/v1/ws"
)

func GenerateHMAC(method string, path string, body []byte, clientId string, timestamp int64, userSecret string) string {
	serverBodyHash := sha256.New()
	serverBodyHash.Write(body)
	serverBodyHashString := fmt.Sprintf("%s %s %s %s %d",
		method,
		path,
		hex.EncodeToString(serverBodyHash.Sum(nil)),
		clientId,
		timestamp)
	fmt.Println("Generating HMAC with the following: ", serverBodyHashString)
	signedMessage := hmac.New(sha256.New, []byte(userSecret))
	signedMessage.Write([]byte(serverBodyHashString))
	userHmac := hex.EncodeToString(signedMessage.Sum(nil))
	return userHmac
}

func GenerateAuthHeaders(method string, pathAndParams string, clientId string, userSecret string) http.Header {
	header := http.Header{}
	timestamp := time.Now().UTC().UnixMilli()
	hmacString := GenerateHMAC(method, pathAndParams, []byte(""), clientId, timestamp, userSecret)

	header.Add("Authorization", clientId)
	header.Add("X-Authorization-Timestamp", strconv.FormatInt(timestamp, 10))
	header.Add("X-Authorization-Signature-SHA256", hmacString)
	return header
}

// connectAndListen connects to the WebSocket server and starts listening for messages.
// It also handles ping/pong communication to keep the connection alive.
func ConnectAndListen(ctx context.Context, feedIds []string) error {
	conn, err := openWebsocketConnection(ctx, feedIds)
	if err != nil {
		return err
	}
	defer conn.Close()

	// Start the ping/pong handling
	go pingLoop(ctx, conn)

	// Set the initial read deadline
	err = conn.SetReadDeadline(time.Now().Add(pongTimeout))
	if err != nil {
		return err
	}

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
			_, msg, err := conn.ReadMessage()
			if err != nil {
				return err
			}

			var decoded NewReportWSMessage
			err = json.Unmarshal(msg, &decoded)
			if err != nil {
				return fmt.Errorf("failed to unmarshal message: %w", err)
			}

			// From here, you have decoded.FeedId and decoded.FullReport
			// See examples for decoding the full report in other code snippets
			fmt.Println("Received the following message: ", decoded)

			// There may be some latency between when you receive a message and when the report is retrievable
			// So pause for a minute before taking action on the message
			time.Sleep(500 * time.Millisecond)
		}
	}
}

// openWebsocketConnection opens a WebSocket connection to the server.
func openWebsocketConnection(ctx context.Context, feedIds []string) (*websocket.Conn, error) {
	baseUrl := os.Getenv("BASE_URL") // Example: https://ws.testnet-dataengine.chain.link
	clientId := os.Getenv("CLIENT_ID") // Example: "00000000-0000-0000-0000-000000000000"
	userSecret := os.Getenv("CLIENT_SECRET") // Example: "your-secret"

	if len(feedIds) == 0 {
        return nil, fmt.Errorf("no feed ID(s) provided")
    }

	params := url.Values{
		"feedIDs":   {strings.Join(feedIds, ",")},
	}

	reqURL := &url.URL{
		Scheme:   "wss", // Note the scheme here
		Host:     baseUrl,
		Path:     wsPath,
		RawQuery: params.Encode(),
	}

	headers := GenerateAuthHeaders("GET", reqURL.RequestURI(), clientId, userSecret)
	conn, _, err := websocket.DefaultDialer.DialContext(ctx, reqURL.String(), headers)
	if err != nil {
		return nil, err
	}

	// Add the Pong handler
	conn.SetPongHandler(func(string) error {
		log.Println("Websocket:", "Received pong...")
		err := conn.SetReadDeadline(time.Now().Add(pongTimeout))
		return err
	})

	return conn, nil
}

// PingPongLoop is a function that handles sending ping messages to the websocket and handles pong messages received
func pingLoop(ctx context.Context, conn *websocket.Conn) {
	ticker := time.NewTicker(pingInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			log.Println("Websocket:", "Sending ping...")
			err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeTimeout))
			if err != nil {
				log.Printf("Failed to send ping: %v", err)
				return
			}
		}
	}
}

Execute the command below to download dependencies and generate the go.sum file:

go mod tidy

Your project directory should now have the following structure:

data-streams-direct-ws/
├── client/
│   └── clientWs.go
├── go.mod
├── go.sum
├── internal/
│   └── decoder.go

Set environment variables

Set the required environment variables in your terminal session to authenticate with the Data Streams Aggregation Network:

export BASE_URL="ws.testnet-dataengine.chain.link"
export CLIENT_ID="YOUR_CLIENT_ID"
export CLIENT_SECRET="YOUR_CLIENT_SECRET"

BASE_URL is the WebSocket endpoint to subscribe to price updates. See the Streams Direct Interface guide for more information.
Replace CLIENT_ID and CLIENT_SECRET with your API credentials.

Establish a WebSocket connection and listen for real-time reports

Create main.go at the root of your project directory:
```
touch main.go
```

Open main.go and insert the following code to establish a WebSocket connection and listen for real-time data reports from the Data Streams Aggregation Network:

// main.go

package main

import (
	"context"
	"data-streams-direct-ws/client"
	"log"
	"os"
)

func main() {
    if len(os.Args) < 2 {
        log.Fatalf("Usage: %s <feedID1> <feedID2> ...", os.Args[0])
    }

    feedIds := os.Args[1:]

    ctx := context.Background()

    // Pass feed IDs to the ConnectAndListen function
    if err := client.ConnectAndListen(ctx, feedIds); err != nil {
        log.Fatalf("Error connecting and listening: %v", err)
    }
}

For this example, you will read from the ETH/USD Data Streams feed on Arbitrum Sepolia. This feed ID is 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782. See the Data Streams Feed IDs page for a complete list of available assets.

Launch the WebSocket listener by running the following command in your terminal:

go run main.go 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782

Expect output similar to the following in your terminal:

Generating HMAC with the following:  GET /api/v1/ws?feedIDs=0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855 <YOUR_CLIENT_ID> 1712557137856
Received the following message:  {{0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 0x0006f9b553e393ced311551efd30d1decedb63d76ad41737462e2cdbbdff1578000000000000000000000000000000000000000000000000000000001d49aa08000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028001000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba7820000000000000000000000000000000000000000000000000000000066138c540000000000000000000000000000000000000000000000000000000066138c5400000000000000000000000000000000000000000000000000001a8d2d99af3c0000000000000000000000000000000000000000000000000014193c2106c39c000000000000000000000000000000000000000000000000000000006614ddd40000000000000000000000000000000000000000000000b9b0fee093d4178d800000000000000000000000000000000000000000000000b9b0c6cc20c60f80400000000000000000000000000000000000000000000000b9b136f506e21f9ac00000000000000000000000000000000000000000000000000000000000000002cbfff26b44d58842991f29cd912099b90672f032fbca13b4d35b870c176b50b60396640cc31466f16bcd09a1afe01597fe194870af103d7d4f5c77648358559e000000000000000000000000000000000000000000000000000000000000000232408446b215b7e0ee343144575cb39c63f694ba162ac127336996e1f8b428547a2ce48e6acee6bcb3cab020f92c9bdf3d2fd5fd8e3b832d5c2267bbe5a3a633}}
Received the following message:  {{0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 0x0006f9b553e393ced311551efd30d1decedb63d76ad41737462e2cdbbdff1578000000000000000000000000000000000000000000000000000000001d49aa0c000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028001010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba7820000000000000000000000000000000000000000000000000000000066138c550000000000000000000000000000000000000000000000000000000066138c5500000000000000000000000000000000000000000000000000001a8d2d80abb80000000000000000000000000000000000000000000000000014193b117c4380000000000000000000000000000000000000000000000000000000006614ddd50000000000000000000000000000000000000000000000b9b0ff8f82d63faa400000000000000000000000000000000000000000000000b9b0c9ce73c00f5a000000000000000000000000000000000000000000000000b9b1355091ec6ced4000000000000000000000000000000000000000000000000000000000000000028b26f5ce5210ba244241da391b852d53bb97d76b561c752f12d147c0d0e07565294f878ac10b4f8f94940374339965d3cb1f7194b0d7303a081c26bdd3f4706e000000000000000000000000000000000000000000000000000000000000000278f418c832ad07cab0522e623875c0f5b4c4359cb0ad76c571dbaa02de699ec172b5e9bd4bfc7e2d6667301555434bd01db348112666c264891c59874272c520}}
Received the following message:  {{0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 0x0006f9b553e393ced311551efd30d1decedb63d76ad41737462e2cdbbdff1578000000000000000000000000000000000000000000000000000000001d49aa10000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028000010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba7820000000000000000000000000000000000000000000000000000000066138c560000000000000000000000000000000000000000000000000000000066138c5600000000000000000000000000000000000000000000000000001a8cefac07b000000000000000000000000000000000000000000000000000141937477f35e0000000000000000000000000000000000000000000000000000000006614ddd60000000000000000000000000000000000000000000000b9b2afff39fd7358400000000000000000000000000000000000000000000000b9b293198c5103b0000000000000000000000000000000000000000000000000b9b2cce4e7a9e300800000000000000000000000000000000000000000000000000000000000000002753b3027be86222709eff93f98059486c278e442535b0c53127908dab09826f22e4fd9d1148550720de61813bd294e3084d724f3d83d972d5abe684a6d9d9ce900000000000000000000000000000000000000000000000000000000000000024236585d1840548b88d2dab324ecb414fbb898f6a9dc4a80979b8f9460802955389e98a95dd26ffc9813489dd2747b6394f95934034ebe1e47fa822bb7419b13}}
Received the following message:  {{0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 0x0006f9b553e393ced311551efd30d1decedb63d76ad41737462e2cdbbdff1578000000000000000000000000000000000000000000000000000000001d49aa14000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba7820000000000000000000000000000000000000000000000000000000066138c570000000000000000000000000000000000000000000000000000000066138c5700000000000000000000000000000000000000000000000000001a8cef5e26300000000000000000000000000000000000000000000000000014191020cfc724000000000000000000000000000000000000000000000000000000006614ddd70000000000000000000000000000000000000000000000b9b2b21fed369590400000000000000000000000000000000000000000000000b9b28e01befceded800000000000000000000000000000000000000000000000b9b2d381feb661a9a00000000000000000000000000000000000000000000000000000000000000002c99d39f03b2e1dc88597401992d0ab2a2ebb829725a7aff4a5b2312dd9d79ccf0447a62620c088d68d577c63029cad7bd46e354775e9d5e1ddb30db3854bb9380000000000000000000000000000000000000000000000000000000000000002158b93031523cda28c976edd47c4b9b6500c13e65026422be3e623de4251fca45ed0107225c022c1a3480cdebc0cb86d5365bc8538702d13d7d80934c12a3275}}
Received the following message:  {{0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 0x0006f9b553e393ced311551efd30d1decedb63d76ad41737462e2cdbbdff1578000000000000000000000000000000000000000000000000000000001d49aa18000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028000010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba7820000000000000000000000000000000000000000000000000000000066138c580000000000000000000000000000000000000000000000000000000066138c5800000000000000000000000000000000000000000000000000001a8cf081ff1400000000000000000000000000000000000000000000000000141946a6b6b830000000000000000000000000000000000000000000000000000000006614ddd80000000000000000000000000000000000000000000000b9b2aa26b724837a200000000000000000000000000000000000000000000000b9ae56d1e39f64f0000000000000000000000000000000000000000000000000b9b39e26a94459d6a00000000000000000000000000000000000000000000000000000000000000002c633d7ac6c69851ef98d890df1b4ba2d31d9a44fcb4e7d40ed91a6a088cf66c58f194573f0ff92f72d87c49fbb07637a35591a7d5c0746666415a46d6a5300f800000000000000000000000000000000000000000000000000000000000000027be80f9819c263925579e18bae6660cd509f4723ffa2417efe9bcc659045af021e8dad3f288a6e166e1bc43fe8c0e0910becdc395f1ff09602ebec10da2dd120}}
2024/04/08 08:19:04 Websocket: Sending ping...
Received the following message:  {{0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 0x0006f9b553e393ced311551efd30d1decedb63d76ad41737462e2cdbbdff1578000000000000000000000000000000000000000000000000000000001d49ab03000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba7820000000000000000000000000000000000000000000000000000000066138c590000000000000000000000000000000000000000000000000000000066138c5900000000000000000000000000000000000000000000000000001a8cf0bc3f300000000000000000000000000000000000000000000000000014194ce58b2658000000000000000000000000000000000000000000000000000000006614ddd90000000000000000000000000000000000000000000000b9b2a88f4e6b50f2e00000000000000000000000000000000000000000000000b9aeb43cd0a5cd7ec00000000000000000000000000000000000000000000000b9b4d02e31df9123400000000000000000000000000000000000000000000000000000000000000002de39c78c325f14159a4dc9797ccafb0d18a5ec278cd13f5ebc43c3988ce27d304d5174161b14636fcfac22bc06ea83fefd630e6ff826dc5e8ee5fed38adb42f800000000000000000000000000000000000000000000000000000000000000021537709709f273854856f78e0c36acf6580eb87cb23798ff0f8d5fd14b14beed7439265bdd11aca9d9a7f981251fc995c687750cb324d0a61b1e70155d0075b1}}
2024/04/08 08:19:05 Websocket: Received pong...
Received the following message:  {{0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 0x0006f9b553e393ced311551efd30d1decedb63d76ad41737462e2cdbbdff1578000000000000000000000000000000000000000000000000000000001d49ab07000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028000010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba7820000000000000000000000000000000000000000000000000000000066138c5a0000000000000000000000000000000000000000000000000000000066138c5a00000000000000000000000000000000000000000000000000001a8cf593f46c000000000000000000000000000000000000000000000000001419360d1b077c000000000000000000000000000000000000000000000000000000006614ddda0000000000000000000000000000000000000000000000b9b286b0a60b5f3ca00000000000000000000000000000000000000000000000b9ae8a4a15117180000000000000000000000000000000000000000000000000b9b4eabe86f969652000000000000000000000000000000000000000000000000000000000000000023ede508913fc4066c754edba91d60e12c71911d616231da453e27eba07f34538dc3f32ae21d6ab44cff5fbba64b240f38252ee0a483c6450518257dc748926e900000000000000000000000000000000000000000000000000000000000000021650e1dc132ec7d7ef19672d3ae8c4e0bd1a069439d017664bb665d29a3b415163c6aa6172b2fb6e1658dee2e09292b741fc81cd17cce224e6e790fe3749df2f}}

When you examine a message from the WebSocket connection's output, you notice that it contains two values:

{{0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 0x0006f9b553e393ced311551efd30d1decedb63d76ad41737462e2cdbbdff1578000000000000000000000000000000000000000000000000000000001d49ab07000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e00000000000000000000000000000000000000000000000000000000000000220000000000000000000000000000000000000000000000000000000000000028000010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000120000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba7820000000000000000000000000000000000000000000000000000000066138c5a0000000000000000000000000000000000000000000000000000000066138c5a00000000000000000000000000000000000000000000000000001a8cf593f46c000000000000000000000000000000000000000000000000001419360d1b077c000000000000000000000000000000000000000000000000000000006614ddda0000000000000000000000000000000000000000000000b9b286b0a60b5f3ca00000000000000000000000000000000000000000000000b9ae8a4a15117180000000000000000000000000000000000000000000000000b9b4eabe86f969652000000000000000000000000000000000000000000000000000000000000000023ede508913fc4066c754edba91d60e12c71911d616231da453e27eba07f34538dc3f32ae21d6ab44cff5fbba64b240f38252ee0a483c6450518257dc748926e900000000000000000000000000000000000000000000000000000000000000021650e1dc132ec7d7ef19672d3ae8c4e0bd1a069439d017664bb665d29a3b415163c6aa6172b2fb6e1658dee2e09292b741fc81cd17cce224e6e790fe3749df2f}}

The first value 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 is the feedId. In this example, it is the ETH/USD feed on Arbitrum Sepolia.
The second value is the FullReport payload in hexadecimal. It contains the encoded report data, which you'll decode in the next section. Save the FullReport payload value for the next step.

Note: In this example, you listen to a single feed. You can listen to multiple feeds by providing the feed IDs as command line arguments separated by spaces. For example:

go run main.go 0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 0x00036fe43f87884450b4c7e093cd5ed99cac6640d8c2000e6afc02c8838d0265

Decode the reports

Create decodeFullReportHex.go at the root of your project directory:
```
touch decodeFullReportHex.go
```

Open decodeFullReportHex.go and insert the sample code provided below. It uses the DecodeFullReportAndReportData function from the internal package to decode the FullReport part of the message.

Replace the fullReportHex variable value with the hexadecimal value you saved earlier from your WebSocket message output, without the 0x prefix.

// decodeFullReportHex.go

package main

import (
	"data-streams-direct-ws/internal"
	"encoding/hex"
	"fmt"
	"log"
)

func main() {
    // Sample FullReport payload extracted from the WebSocket message as a hex string
    fullReportHex := "00067f14c763070bec1de1118aceeed1546878ab24e3213de21127249adabcbd0000000000000000000000000000000000000000000000000000000011f0c90b000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e000000000000000000000000000000000000000000000000000000000000001e00000000000000000000000000000000000000000000000000000000000000240010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e000027bbaff688c906a3e20a34fe951715d1018d262a5b66e38eda027a674cd1b0000000000000000000000000000000000000000000000000000000065cdcd950000000000000000000000000000000000000000000000000000000065cdcd95000000000000000000000000000000000000000000000000000020b5e9c686e80000000000000000000000000000000000000000000000000011b8f926846fb80000000000000000000000000000000000000000000000000000000065cf1f15000000000000000000000000000000000000000000000096ba314c8f5ec2800000000000000000000000000000000000000000000000000000000000000000029c85f7bb779ce5316821a6efd3bdc843e32f1438bad24a6b269871ade0c166d2142574b78d7a75d3cb855e51caf9cb36e1d598b43c881989251bd2c450624a2600000000000000000000000000000000000000000000000000000000000000021c1e5c393a57a24f2c698e23c59e6a954a9ef006b63c7f58eeabdd0bbeff21e5353ddef53e9beb93f628cd23e2cc28750533f444559622640bcf7dcc935d66af"

    // Convert the hex string to a byte slice
    fullReportPayload, err := hex.DecodeString(fullReportHex)
    if err != nil {
        log.Fatalf("Failed to decode hex string: %v", err)
    }

    // Decode the full report
    decodedReport, err := internal.DecodeFullReportAndReportData(fullReportPayload)
    if err != nil {
        log.Fatalf("Failed to decode report: %v", err)
    }

    fmt.Printf("Decoded Report: %+v\n", decodedReport)

    if decodedReport.FeedVersion == 3 && decodedReport.V3Report != nil {
        fmt.Printf("Valid From Timestamp: %d\n", decodedReport.V3Report.ValidFromTimestamp)
        fmt.Printf("Observations Timestamp: %d\n", decodedReport.V3Report.ObservationsTimestamp)
        fmt.Printf("Native Fee: %s\n", decodedReport.V3Report.NativeFee.String())
        fmt.Printf("Link Fee: %s\n", decodedReport.V3Report.LinkFee.String())
        fmt.Printf("Expires At: %d\n", decodedReport.V3Report.ExpiresAt)
        fmt.Printf("Benchmark Price: %s\n", decodedReport.V3Report.BenchmarkPrice.String())
        fmt.Printf("Bid: %s\n", decodedReport.V3Report.Bid.String())
        fmt.Printf("Ask: %s\n", decodedReport.V3Report.Ask.String())
    } 
}

Run the decoding script by executing the following command in your terminal:

go run decodeFullReportHex.go

The decodedReport contains structured data about the report, including FeedId and FeedVersion.

Expect output similar to the following in your terminal:

Decoded Report: &{FeedId:0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782 FeedVersion:3 V2Report:<nil> V3Report:0x140001b4c60 Round:11 Epoch:1919476 Digest:[0 6 249 181 83 227 147 206 211 17 85 30 253 48 209 222 206 219 99 215 106 212 23 55 70 46 44 219 189 255 21 120]}
Valid From Timestamp: 1712557606
Observations Timestamp: 1712557606
Native Fee: 29191712479600
Link Fee: 5655191262392300
Expires At: 1712644006
Benchmark Price: 3425629793731062400000
Bid: 3425455500000000000000
Ask: 3425747100000000000000

Decoded report details

The decoded report details include:

Attribute	Value	Description
Feed ID	`0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782`	The unique identifier for the Data Streams feed. In this example, the feed is ETH/USD on Arbitrum Sepolia with the Basic Report schema.
Feed Version	`3`	The feed report schema version.
Valid From Timestamp	`1712557606`	The start validity timestamp for the report, indicating when the data becomes relevant.
Observations Timestamp	`1712557606`	The timestamp indicating when the data was captured.
Native Fee	`29191712479600`	The fee to pay in the native blockchain token (e.g., testnet ETH on Arbitrum Sepolia) for the onchain verification of the report data. With 18 decimals. Note: This example fee is not indicative of actual fees.
Link Fee	`5655191262392300`	The fee to pay in LINK tokens for the onchain verification of the report data. With 18 decimals. For readability: `0.005655191262` LINK. Note: This example fee is not indicative of actual fees.
Expires At	`1712644006`	The expiration timestamp of the report, indicating the point at which the data becomes outdated.
Benchmark Price	`3425629793731062400000`	The observed price in the report, with 18 decimals. For readability: `3,425.6297937310624` ETH per USD.
Bid	`3425455500000000000000`	The simulated price impact of a buy order up to the X% depth of liquidity usage. For readability: `3,425.4555` ETH per USD.
Ask	`3425747100000000000000`	The simulated price impact of a sell order up to the X% depth of liquidity usage. For readability: `3,425.7471` ETH per USD.

Verify report data onchain

In this tutorial, you log and decode the FullReport payloads to extract the reports data. In a production environment, you should verify the data onchain to ensure its integrity and authenticity. Refer to the Verify report data onchain guide.

Explanation

Establishing a WebSocket connection and listening for message reports

The ConnectAndListen function in the client package initializes the connection and enables the application to listen for messages from the Data Streams Aggregation Network via WebSocket.

Authentication and security: Uses HMAC headers, created with CLIENT_ID and CLIENT_SECRET, for secure, authenticated communication.
Continuous listening: After authentication, continuously listens for real-time report messages, which include the feedId and the encoded FullReport.

Decoding a report

The DecodeFullReportAndReportData function in the internal package unpacks the FullReport payload, decodes the FullReport's ReportBlob, and decodes the report data into a structured format:

It unpacks the FullReport payload that includes the report context, the report blob and raw signatures:
- Report context: The report context contains metadata such as the feedId and the report version.
- Cryptographic signatures: These signatures validate the data's integrity and authenticity.
Report data decoding: The function proceeds to decode the report data into a structured format.

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.

Stream and decode reports via WebSocket

What's next

Stay updated on the latest Chainlink news