Skip to content

Listening to New Blocks⚓︎

BEGINNER

The blocks WS WebSocket Channel sends a real-time notification every time a new block is added to the chain. Compared to polling the /chain/height GET endpoint, WebSockets push updates as they happen without the overhead of repeated API calls.

This tutorial shows how to subscribe to the channel and display each update as it arrives.

Polling alternative

For a polling-based approach, see the Querying Chain and Irreversible Height tutorial.

Prerequisites⚓︎

NEM serves WebSockets using the STOMP messaging protocol over SockJS, so a STOMP client and a WebSocket transport are required.

Install the stomper and websockets libraries:

pip install stomper websockets

Install the @stomp/stompjs and sockjs-client libraries:

npm install @stomp/stompjs sockjs-client

This tutorial uses the Tyrus WebSocket client, which is loaded by the JBang dependency line in the snippet.

See the WebSocket reference for details on the connection protocol.

Full Code⚓︎

The following is the complete code listing for this tutorial. A detailed, step-by-step explanation follows in the next section.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
import asyncio
import json
import os
import random
import uuid

import stomper
from websockets import connect

NODE_URL = os.getenv('NODE_URL', 'http://libertalia.nemtest.net:7890')
WS_URL = NODE_URL.replace(':7890', ':7778')
print(f'Using node {NODE_URL}')


# SockJS has no Python client library.
# These helpers wrap the raw WebSocket transport.
def sockjs_url(endpoint_url):
    # SockJS raw WebSocket transport adds a random server and session id
    server = random.randint(100, 999)
    session = uuid.uuid4().hex
    ws_base = endpoint_url.replace('http', 'ws', 1)
    return f'{ws_base}/{server}/{session}/websocket'


async def send_frame(websocket, frame):
    # SockJS wraps each client payload as a JSON array of frame strings
    await websocket.send(json.dumps([frame]))


async def stomp_connect(websocket):
    await websocket.recv()  # consume the SockJS open frame
    await send_frame(
        websocket, stomper.connect('', '', WS_URL, heartbeats=(0, 0)))


async def stomp_subscribe(websocket, destination, sub_id):
    await send_frame(websocket, stomper.subscribe(destination, sub_id))


async def stomp_unsubscribe(websocket, sub_id):
    await send_frame(websocket, stomper.unsubscribe(sub_id))


async def stomp_disconnect(websocket):
    await send_frame(websocket, stomper.disconnect())


def stomp_messages(raw_frame):
    # Yield the JSON body of each STOMP MESSAGE in a SockJS data frame
    if 'a' != raw_frame[0]:  # skip 'o' open, 'h' heartbeat, 'c' close
        return
    for payload in json.loads(raw_frame[1:]):
        frame = stomper.unpack_frame(payload)
        if 'MESSAGE' == frame['cmd']:
            yield json.loads(frame['body'])


async def main():
    # Open connection
    async with connect(sockjs_url(f'{WS_URL}/w/messages')) as websocket:
        await stomp_connect(websocket)
        print(f'Connected to {WS_URL}')

        # Subscribe to the new block channel
        destination = '/blocks'
        await stomp_subscribe(websocket, destination, 'id-0')
        print(f'Subscribed to {destination} channel')

        # Read and format each new block
        try:
            async for raw_frame in websocket:
                for block in stomp_messages(raw_frame):
                    print(
                        f'New block: height={block["height"]:,}'
                        f' harvester={block["signer"][:16].upper()}...'
                    )

        # Unsubscribe on exit
        finally:
            await stomp_unsubscribe(websocket, 'id-0')
            await stomp_disconnect(websocket)
            print('Unsubscribed and disconnected')


try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass
except Exception as error:
    print(error)

Download source

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
import { Client } from '@stomp/stompjs';
import SockJS from 'sockjs-client';

const NODE_URL = process.env.NODE_URL ||
    'http://libertalia.nemtest.net:7890';
const WS_URL = NODE_URL.replace(':7890', ':7778');
console.log(`Using node ${NODE_URL}`);

// Open connection
const client = new Client({
    webSocketFactory: () => new SockJS(`${WS_URL}/w/messages`)
});
await new Promise(resolve => {
    client.onConnect = resolve;
    client.activate();
});
console.log(`Connected to ${WS_URL}`);

// Read and format each new block
function formatBlock(message) {
    const block = JSON.parse(message.body);
    console.log(
        `New block: height=${block.height.toLocaleString()}` +
        ` harvester=${block.signer.substring(0, 16).toUpperCase()}...`
    );
}

// Subscribe to the new block channel
const destination = '/blocks';
const subscription = client.subscribe(destination, formatBlock, {
    id: 'id-0'
});
console.log(`Subscribed to ${destination} channel`);

// Unsubscribe on exit
process.on('SIGINT', () => {
    subscription.unsubscribe();
    client.deactivate();
    console.log('Unsubscribed and disconnected');
    process.exit(0);
});

Download source

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
//JAVA 21+
//DEPS com.fasterxml.jackson.core:jackson-databind:2.17.1
//DEPS org.glassfish.tyrus.bundles:tyrus-standalone-client:2.2.0

import java.io.IOException;
import java.net.URI;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import jakarta.websocket.ClientEndpoint;
import jakarta.websocket.ContainerProvider;
import jakarta.websocket.OnMessage;
import jakarta.websocket.RemoteEndpoint;
import jakarta.websocket.Session;
import jakarta.websocket.WebSocketContainer;

@ClientEndpoint
public final class ListenNewBlocks {
    private static final ObjectMapper JSON_MAPPER = new ObjectMapper();

    private final CompletableFuture<Void> openFuture =
        new CompletableFuture<>();

    private final String nodeUrl = System.getenv().getOrDefault(
        "NODE_URL", "http://libertalia.nemtest.net:7890");

    private final String wsUrl = nodeUrl.replace(":7890", ":7778");

    public static void main(final String[] args) {
        try {
            new ListenNewBlocks().run();
        } catch (final Exception ex) {
            System.out.println(null == ex.getMessage()
                ? ex.toString()
                : ex.getMessage());
        }
    }

    private void run() throws Exception {
        System.out.printf("Using node %s%n", nodeUrl);

        // Open connection
        final WebSocketContainer container =
            ContainerProvider.getWebSocketContainer();
        final Session session = container.connectToServer(
            this, URI.create(sockJsUrl(wsUrl + "/w/messages")));
        final RemoteEndpoint.Basic remote = session.getBasicRemote();
        openFuture.join();
        sendFrame(remote, "CONNECT\naccept-version:1.1,1.0\n"
            + "heart-beat:0,0\nhost:" + wsUrl + "\n\n\0");
        System.out.printf("Connected to %s%n", wsUrl);

        // Subscribe to the new block channel
        final String destination = "/blocks";
        sendFrame(remote, "SUBSCRIBE\nid:id-0\ndestination:"
            + destination + "\nack:auto\n\n\0");
        System.out.printf("Subscribed to %s channel%n", destination);

        // Unsubscribe on exit
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                sendFrame(remote, "UNSUBSCRIBE\nid:id-0\n\n\0");
                sendFrame(remote, "DISCONNECT\n\n\0");
                session.close();
                System.out.println("Unsubscribed and disconnected");
            } catch (final IOException ex) {
                throw new IllegalStateException(ex);
            }
        }));

        new CompletableFuture<Void>().join();
    }

    // Read and format each new block
    @OnMessage
    public void onMessage(final String rawFrame) throws IOException {
        if ("o".equals(rawFrame)) {
            openFuture.complete(null);
            return;
        }
        if (!rawFrame.startsWith("a"))
            return;
        for (final JsonNode payload
            : JSON_MAPPER.readTree(rawFrame.substring(1))) {
            final String frame = payload.asText();
            if (!frame.startsWith("MESSAGE\n"))
                continue;
            final int bodyStart = frame.indexOf("\n\n") + 2;
            final String body = frame.substring(bodyStart)
                .replace("\0", "");
            final JsonNode block = JSON_MAPPER.readTree(body);
            System.out.printf(
                "New block: height=%,d harvester=%s...%n",
                block.get("height").asLong(),
                block.get("signer").asText().substring(0, 16)
                    .toUpperCase());
        }
    }

    private static String sockJsUrl(final String endpoint) {
        final int server = ThreadLocalRandom.current().nextInt(100, 1000);
        return endpoint.replaceFirst("http", "ws") + "/" + server
            + "/" + UUID.randomUUID().toString().replace("-", "")
            + "/websocket";
    }

    private static void sendFrame(
        final RemoteEndpoint.Basic remote,
        final String frame
    ) throws IOException {
        final ArrayNode payload = JSON_MAPPER.createArrayNode();
        payload.add(frame);
        remote.sendText(payload.toString());
    }
}

Download source

The snippet uses the NODE_URL environment variable to set the NEM node. If no value is provided, a default one is used.

WS_URL defines the WebSocket endpoint for the same node. It is derived from NODE_URL by replacing port 7890, the default HTTP API port, with 7778, the default NIS WebSocket port.

The program runs until interrupted with Ctrl+C, which triggers the unsubscribe step before closing the connection.

Python SockJS helpers

There is no SockJS client library for Python, so a few small helper methods are defined at the top of the file for convenience.

Code Explanation⚓︎

Connecting to the WebSocket⚓︎

59
60
61
62
    # Open connection
    async with connect(sockjs_url(f'{WS_URL}/w/messages')) as websocket:
        await stomp_connect(websocket)
        print(f'Connected to {WS_URL}')

 9
10
11
12
13
14
15
16
17
// Open connection
const client = new Client({
    webSocketFactory: () => new SockJS(`${WS_URL}/w/messages`)
});
await new Promise(resolve => {
    client.onConnect = resolve;
    client.activate();
});
console.log(`Connected to ${WS_URL}`);

46
47
48
49
50
51
52
53
54
55
        // Open connection
        final WebSocketContainer container =
            ContainerProvider.getWebSocketContainer();
        final Session session = container.connectToServer(
            this, URI.create(sockJsUrl(wsUrl + "/w/messages")));
        final RemoteEndpoint.Basic remote = session.getBasicRemote();
        openFuture.join();
        sendFrame(remote, "CONNECT\naccept-version:1.1,1.0\n"
            + "heart-beat:0,0\nhost:" + wsUrl + "\n\n\0");
        System.out.printf("Connected to %s%n", wsUrl);

The first step is to open a connection to the node's /w/messages endpoint and start a STOMP session over it.

Subscribing to the Channel⚓︎

64
65
66
67
        # Subscribe to the new block channel
        destination = '/blocks'
        await stomp_subscribe(websocket, destination, 'id-0')
        print(f'Subscribed to {destination} channel')

28
29
30
31
32
33
// Subscribe to the new block channel
const destination = '/blocks';
const subscription = client.subscribe(destination, formatBlock, {
    id: 'id-0'
});
console.log(`Subscribed to ${destination} channel`);

57
58
59
60
61
        // Subscribe to the new block channel
        final String destination = "/blocks";
        sendFrame(remote, "SUBSCRIBE\nid:id-0\ndestination:"
            + destination + "\nack:auto\n\n\0");
        System.out.printf("Subscribed to %s channel%n", destination);

The code subscribes to the blocks WS channel. The node then notifies subscribers every time a new block is added to the chain (approximately every minute).

Notifications are not always evenly spaced. When the node adds several blocks at once, for example while catching up with its peers, the channel delivers them in a quick burst, still one notification per block.

The subscription is given an id (id-0) which is used to unsubscribe on exit.

Each incoming message is then passed to the formatting logic below.

Formatting the Message⚓︎

69
70
71
72
73
74
75
76
        # Read and format each new block
        try:
            async for raw_frame in websocket:
                for block in stomp_messages(raw_frame):
                    print(
                        f'New block: height={block["height"]:,}'
                        f' harvester={block["signer"][:16].upper()}...'
                    )

19
20
21
22
23
24
25
26
// Read and format each new block
function formatBlock(message) {
    const block = JSON.parse(message.body);
    console.log(
        `New block: height=${block.height.toLocaleString()}` +
        ` harvester=${block.signer.substring(0, 16).toUpperCase()}...`
    );
}

 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
    // Read and format each new block
    @OnMessage
    public void onMessage(final String rawFrame) throws IOException {
        if ("o".equals(rawFrame)) {
            openFuture.complete(null);
            return;
        }
        if (!rawFrame.startsWith("a"))
            return;
        for (final JsonNode payload
            : JSON_MAPPER.readTree(rawFrame.substring(1))) {
            final String frame = payload.asText();
            if (!frame.startsWith("MESSAGE\n"))
                continue;
            final int bodyStart = frame.indexOf("\n\n") + 2;
            final String body = frame.substring(bodyStart)
                .replace("\0", "");
            final JsonNode block = JSON_MAPPER.readTree(body);
            System.out.printf(
                "New block: height=%,d harvester=%s...%n",
                block.get("height").asLong(),
                block.get("signer").asText().substring(0, 16)
                    .toUpperCase());
        }
    }

The body of each incoming message is the new block, following the Block schema.

For each message, the snippet prints two of its fields:

  • height: The height of the new block.
  • signer: The public key of the harvester account that produced the block.

A NEM block does not include its own hash in this payload, so this tutorial identifies each block by its height and also prints the harvester's signer.

New blocks are not yet final

New blocks are already part of the chain but not yet irreversible. Until enough subsequent blocks surpass the rewrite limit, rollbacks are still possible. After a rollback, the channel can even report a block with a lower height than one already received, because the incoming blocks replace blocks that were reported earlier.

The Querying Chain and Irreversible Height tutorial shows how to calculate the irreversible height.

Unsubscribing on Exit⚓︎

78
79
80
81
82
        # Unsubscribe on exit
        finally:
            await stomp_unsubscribe(websocket, 'id-0')
            await stomp_disconnect(websocket)
            print('Unsubscribed and disconnected')

35
36
37
38
39
40
41
// Unsubscribe on exit
process.on('SIGINT', () => {
    subscription.unsubscribe();
    client.deactivate();
    console.log('Unsubscribed and disconnected');
    process.exit(0);
});

63
64
65
66
67
68
69
70
71
72
73
        // Unsubscribe on exit
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                sendFrame(remote, "UNSUBSCRIBE\nid:id-0\n\n\0");
                sendFrame(remote, "DISCONNECT\n\n\0");
                session.close();
                System.out.println("Unsubscribed and disconnected");
            } catch (final IOException ex) {
                throw new IllegalStateException(ex);
            }
        }));

When the program is interrupted (Ctrl+C), the code unsubscribes from the channel and ends the STOMP session before closing the connection. This ensures a clean disconnection from the node.

Output⚓︎

The following output shows a typical run listening to new blocks:

1
2
3
4
5
6
7
8
9
Using node http://libertalia.nemtest.net:7890
Connected to http://libertalia.nemtest.net:7778
Subscribed to /blocks channel
New block: height=682,352 harvester=95BA0E864CD5EDB3...
New block: height=682,353 harvester=95BA0E864CD5EDB3...
New block: height=682,354 harvester=7C814B3B44B1A98B...
New block: height=682,355 harvester=95BA0E864CD5EDB3...
New block: height=682,356 harvester=95BA0E864CD5EDB3...
Unsubscribed and disconnected

The output shows:

  • Connection (line 2): The STOMP session is established over the node's WebSocket endpoint at port 7778.
  • Subscription (line 3): The /blocks channel is subscribed.
  • New blocks (lines 4-8): New block notifications arrive approximately every minute.
  • Unsubscribe (line 9): On Ctrl+C, the code unsubscribes and disconnects.

Conclusion⚓︎

This tutorial showed how to:

Step Related documentation
Subscribe to the block channel blocks WS
Format block messages Block