Websocket

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.

Examples

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())

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())

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())

PreviousHandling the Notification NextgRPC