Creating a Subscription

This section explains how to create stream subscriptions using the two supported protocols: WebSocket and gRPC. Both approaches are covered below.

WebSocket

Subscriptions are created using the subscribe RPC call, which takes the stream (feed) name and optional subscription parameters. Each stream has its own subscription name and supported options; refer to the relevant stream page for details.

The subscribe call returns a subscription ID, which is included in all related notifications.

Before subscribing, you must establish a WebSocket connection to either the Gateway or the Cloud API.

Notes

• The examples below assume the Gateway is running on 127.0.0.1 with the default WebSocket port 28333:

Gateway

Notes:

• We assume that the Gateway IP is 127.0.0.1 with default ws port 28333 in the example below.

ws://127.0.0.1:28333/ws

Cloud-API

Authentication method: Use authorization header.

wss://<region>.eth.blxrbdn.com 

Examples

Gateway

Subscribing to Gateway Stream in Python (version 3.7 or higher):

import asyncio, websockets, json

async def main():
    try:
        auth_key = "YOUR_AUTHORIZATION_HEADER"
        
        async with websockets.connect(
            'ws://127.0.0.1:28333/ws',
            additional_headers={"Authorization" : auth_key},
        ) as websocket:
            request = json.dumps({"id": 1, "method": "subscribe", "params": ["newTxs", {"include": ["tx_hash"]}]})
            websocket.send(request)
            while True:
                response = json.loads(websocket.recv())
                print(response) # or process it generally
    except Exception as e:
        print(f'Connection failed, Reason: {e}')

if __name__ == '__main__':
    asyncio.run(main())

Cloud-API

Subscribing to Cloud-API Feed in Python (version 3.7 or higher):

import asyncio, json, websockets

async def main():
    try:
        auth_key = "YOUR_AUTHORIZATION_HEADER"
        
        async with websockets.connect(
            'wss://virginia.eth.blxrbdn.com/ws',
             additional_headers={"Authorization" : auth_key},    
        ) as websocket:
            subscription_request = json.dumps({"id": 1, "method": "subscribe", "params": ["newTxs", {"include": ["tx_hash"]}]})
            await websocket.send(subscription_request)
            while True:
                response = await websocket.recv()
                print(response)
    except Exception as e:
        print(f'Connection failed, Reason: {e}')

if __name__ == '__main__':
    asyncio.run(main())

gRPC

Streams are available over gRPC via both Gateway API and Cloud API. To enable gRPC, start the Gateway with the grpc flag.

To create a subscription, first open a gRPC connection to the Gateway using:

• grpc-host (default: 127.0.0.1)

• grpc-port (default: 5001)

Each stream has a dedicated gRPC method and request type. Transaction streams support a Filters field, and all streams support an Includes field. Stream-specific options are documented on the corresponding stream page.

All gRPC requests must include the Authorization header. Requests without valid authentication will return an error.

Some gRPC response fields are deprecated and may not be populated. Refer to the proto files for the current response schema.

Example

Gateway-API

Subscribing to Gateway Stream in Go using gRPC (ex: NewTxs stream):

package main

import (
	"context"
	"fmt"
	"time"

	pb "github.com/bloXroute-Labs/gateway/v2/protobuf"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

type blxrCredentials struct {
	authorization string
}

func (bc blxrCredentials) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
	return map[string]string{
		"authorization": bc.authorization,
	}, nil
}

func (bc blxrCredentials) RequireTransportSecurity() bool {
	return false
}

func main() {
	// gRPC server default values
	gatewayHostIP := "localhost"
	gatewayGRPCPort := 5001
	
	// this will use localhost CA to verify the certificate
	creds := credentials.NewClientTLSFromCert(nil, "")

	// Open gRPC connection to Gateway.
	conn, _ := grpc.Dial(
		fmt.Sprintf("%v:%v", gatewayHostIP, gatewayGRPCPort),
		grpc.WithTransportCredentials(creds),
		grpc.WithPerRPCCredentials(blxrCredentials{authorization: "<YOUR-AUTHORIZATION-HEADER>"}),
		// for the best networking performance between gateway and a client
		// we recommend to use following dial configuration:
		grpc.WithWriteBufferSize(0),
		grpc.WithInitialConnWindowSize(128*1024),
	)

	// Use the Gateway client connection interface.
	client := pb.NewGatewayClient(conn)

	// create context and defer cancel of context
	callContext, cancel := context.WithTimeout(context.Background(), 24*time.Hour)
	defer cancel()

	// Create a subscription using the stream-specific method and request.
	stream, _ := client.NewTxs(callContext, &pb.TxsRequest{Filters: "", Includes: ""})

	for {
		subscriptionNotification, err := stream.Recv()
		if err == nil {
			fmt.Println(subscriptionNotification) // or process it generally
		}
	}
}