Handling the Notification

To process a new notification, call the function get_next_subscription_notification_by_id, passing in the subscription_id returned from the subscribe command as a parameter.

After subscribing to a stream, your application receives a notification for each published event (for example, a new transaction or block). This section explains how to handle stream notifications using the supported protocols: WebSocket and gRPC. Both approaches are covered below.

WebSocket

For WebSocket subscriptions, notifications are delivered over the established WebSocket connection. Each notification includes the subscription_id returned by the initial subscribe call.

To process a notification, listen for incoming messages and match the message id with the corresponding subscription_id.

Examples

Gateway-API

Handling the transaction notification from a Gateway Stream in Python (line 8 handles the notification):

import asyncio, json, ssl, websockets

async def main():
    try:
        ws_uri = "ws://127.0.0.1:28333/ws",
        auth_key = "YOUR_AUTHORIZATION_HEADER"

        async with websockets.connect(
            ws_uri,
            header=["Authorization:{}".format(auth_key)],
        ) as websocket:
            subscription_request = json.dumps({"id": 1, "method": "subscribe", "params": ["newTxs", {"include": ["tx_hash"]}]})
            await websocket.send(subscription_request)
            response = await websocket.recv()
            subscription_id = json.loads(response)['result']

            while True:
                response = await websocket.recv()
                response = json.loads(response)
                if response['id'] == subscription_id:
                    print(response) # or process it generally
    except Exception as e:
        print(f'Connection failed, Reason: {e}')

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

Cloud-API (Enterprise)

Handling the transaction notification from a Cloud-API Stream in Python (version 3.7 or higher):

import asyncio, json, ssl, websockets

async def main():
    try:
        auth_key = "YOUR_AUTHORIZATION_HEADER"
        
        async with websockets.connect(
            'wss://virginia.eth.blxrbdn.com/ws',
            header=["Authorization:{}".format(auth_key)],
            sslopt={"cert_reqs": ssl.CERT_NONE},
        ) as websocket:
            subscription_request = json.dumps({"id": 1, "method": "subscribe", "params": ["newTxs", {"include": ["tx_hash"]}]})
            await websocket.send(subscription_request)
            subscription_id = None
            while True:
                response = await websocket.recv()
                response_json = json.loads(response)
                if response_json.get('id') == 1:
                    subscription_id = response_json.get('result')
                else:
                    print(response_json)  # or process it generally
    except Exception as e:
        print(f'Connection failed, Reason: {e}')
    finally:
        if subscription_id is not None:
            try:
                unsubscription_request = json.dumps({"id": 2, "method": "unsubscribe", "params": [subscription_id]})
                await websocket.send(unsubscription_request)
            except Exception as e:
                print(f'Unsubscription failed, Reason: {e}')

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

Cloud-API (non-Enterprise)

Handling the transaction notification from a Cloud-API Stream in Python (version 3.7 or higher):

import asyncio, json, ssl, websockets

async def main():
    try:
        auth_key = "YOUR_AUTHORIZATION_HEADER"

        async with websockets.connect(
            'wss://api.blxrbdn.com/ws',
            header=["Authorization:{}".format(auth_key)],
            sslopt={"cert_reqs": ssl.CERT_NONE},
        ) as websocket:
            subscription_request = json.dumps({"id": 1, "method": "subscribe", "params": ["newTxs", {"include": ["tx_hash"]}]})
            await websocket.send(subscription_request)
            subscription_id = None
            while True:
                response = await websocket.recv()
                response_json = json.loads(response)
                if response_json.get('id') == 1:
                    subscription_id = response_json.get('result')
                else:
                    print(response_json)  # or process it generally
    except Exception as e:
        print(f'Connection failed, Reason: {e}')
    finally:
        if subscription_id is not None:
            try:
                unsubscription_request = json.dumps({"id": 2, "method": "unsubscribe", "params": [subscription_id]})
                await websocket.send(unsubscription_request)
            except Exception as e:
                print(f'Unsubscription failed, Reason: {e}')

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

gRPC

For gRPC subscriptions, notifications are received by calling Recv() on the stream object returned by the subscription request. Each call to Recv() blocks until a new event is available or an error occurs.

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
		}
	}
}