コンテンツにスキップ

新しいブロックのリスニング⚓︎

初級

block WS と finalizedBlock WS WebSocket チャネルは、新しい ブロック が生成されたとき、または ファイナライズ されたときにリアルタイムの通知を送信します。 /chain/info GET エンドポイントをポーリングする場合と比較して、WebSocket は API 呼び出しを繰り返すオーバーヘッドなしに、更新が発生した瞬間にプッシュします。

このチュートリアルでは、両方のチャネルをサブスクライブし、到着した各更新を表示する方法を説明します。

代替手段: ポーリング

ポーリングベースのアプローチについては、チェーンとファイナライズの最新高の照会 チュートリアルを参照してください。

前提条件⚓︎

websockets ライブラリをインストールします。

pip install websockets

このチュートリアルでは、Node.js 22 以降で利用可能なネイティブの WebSocket API を使用します。 追加のパッケージは必要ありません。

このチュートリアルでは Tyrus WebSocket クライアントを使用します。これはスニペット内の JBang 依存関係の行によって読み込まれます。

接続プロトコルの詳細については、WebSocket リファレンス を参照してください。

完全なコード⚓︎

このチュートリアルの完全なコード一覧を以下に示します。 詳細な手順ごとの説明は次のセクションで行います。

 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
import asyncio
import json
import os

from websockets import connect

NODE_URL = os.getenv('NODE_URL', 'https://reference.symboltest.net:3001')
WS_URL = NODE_URL.replace('http', 'ws', 1) + '/ws'
print(f'Using node {NODE_URL}')


async def main():
    async with connect(WS_URL) as websocket:
        # Connect to websocket endpoint
        response = json.loads(await websocket.recv())
        uid = response['uid']
        print(f'Connected to {WS_URL} with uid {uid}')

        # Subscribe to block channel
        await websocket.send(json.dumps(
            {'uid': uid, 'subscribe': 'block'}))
        print('Subscribed to block channel')

        # Subscribe to finalizedBlock channel
        await websocket.send(json.dumps(
            {'uid': uid, 'subscribe': 'finalizedBlock'}))
        print('Subscribed to finalizedBlock channel')

        # Handle incoming messages
        try:
            async for raw_message in websocket:
                message = json.loads(raw_message)
                topic = message['topic']

                if topic == 'block':
                    block = message['data']['block']
                    block_meta = message['data']['meta']
                    print(
                        f'New block: height={int(block["height"]):,}'
                        f' hash={block_meta["hash"][:16]}...'
                    )

                if topic == 'finalizedBlock':
                    finalized = message['data']
                    print(
                        f'Finalized: height={int(finalized["height"]):,}'
                        f' hash={finalized["hash"][:16]}...'
                    )

        # Unsubscribe on exit
        finally:
            await websocket.send(json.dumps(
                {'uid': uid, 'unsubscribe': 'block'}))
            await websocket.send(json.dumps(
                {'uid': uid, 'unsubscribe': 'finalizedBlock'}))
            print('Unsubscribed from all channels')


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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
const NODE_URL = process.env.NODE_URL ||
    'https://reference.symboltest.net:3001';
const WS_URL = `${NODE_URL.replace('http', 'ws')}/ws`;
console.log(`Using node ${NODE_URL}`);

try {
    const websocket = new WebSocket(WS_URL);

    // Connect to websocket endpoint
    const uid = await new Promise(resolve => {
        websocket.addEventListener('message', event => {
            const message = JSON.parse(event.data);
            resolve(message.uid);
        }, { once: true });
    });
    console.log(`Connected to ${WS_URL} with uid ${uid}`);

    // Subscribe to block channel
    websocket.send(JSON.stringify({ uid, subscribe: 'block' }));
    console.log('Subscribed to block channel');

    // Subscribe to finalizedBlock channel
    websocket.send(JSON.stringify({ uid, subscribe: 'finalizedBlock' }));
    console.log('Subscribed to finalizedBlock channel');

    // Handle incoming messages
    websocket.addEventListener('message', event => {
        const message = JSON.parse(event.data);
        const topic = message.topic;

        if ('block' === topic) {
            const block = message.data.block;
            const blockMeta = message.data.meta;
            console.log(
                `New block: height=${
                    BigInt(block.height).toLocaleString()}` +
                ` hash=${blockMeta.hash.substring(0, 16)}...`);
        }

        if ('finalizedBlock' === topic) {
            const finalized = message.data;
            console.log(
                `Finalized: height=${
                    BigInt(finalized.height).toLocaleString()}` +
                ` hash=${finalized.hash.substring(0, 16)}...`);
        }
    });

    // Unsubscribe on exit
    process.on('SIGINT', () => {
        websocket.send(JSON.stringify({ uid, unsubscribe: 'block' }));
        websocket.send(
            JSON.stringify({ uid, unsubscribe: 'finalizedBlock' }));
        console.log('Unsubscribed from all channels');
        websocket.close();
        process.exit(0);
    });
} catch (error) {
    console.error(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
 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
//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.concurrent.CompletableFuture;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
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<String> uidFuture =
        new CompletableFuture<>();

    private final String nodeUrl = System.getenv().getOrDefault(
        "NODE_URL", "https://reference.symboltest.net:3001");

    private final String wsUrl = nodeUrl.replaceFirst("http", "ws")
        + "/ws";

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


        final WebSocketContainer container =
            ContainerProvider.getWebSocketContainer();
        final Session session = container.connectToServer(
            this, URI.create(wsUrl));
        final RemoteEndpoint.Basic remote = session.getBasicRemote();

        // Connect to websocket endpoint
        final String uid = uidFuture.join();
        System.out.printf("Connected to %s with uid %s%n", wsUrl, uid);

        // Subscribe to block channel
        remote.sendText(JSON_MAPPER.createObjectNode()
            .put("uid", uid).put("subscribe", "block")
            .toString());
        System.out.println("Subscribed to block channel");

        // Subscribe to finalizedBlock channel
        remote.sendText(JSON_MAPPER.createObjectNode()
            .put("uid", uid).put("subscribe", "finalizedBlock")
            .toString());
        System.out.println("Subscribed to finalizedBlock channel");

        // Unsubscribe on exit
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                remote.sendText(JSON_MAPPER.createObjectNode()
                    .put("uid", uid).put("unsubscribe", "block")
                    .toString());
                remote.sendText(JSON_MAPPER.createObjectNode()
                    .put("uid", uid).put("unsubscribe", "finalizedBlock")
                    .toString());
                System.out.println("Unsubscribed from all channels");
                session.close();
            } catch (final IOException ex) {
                throw new IllegalStateException(ex);
            }
        }));

        // Wait forever
        Thread.currentThread().join();
    }

    // Handle incoming messages
    @OnMessage
    public void onMessage(final String payload) throws IOException {
        final JsonNode message = JSON_MAPPER.readTree(payload);

        // Special case for the initial handshake message
        if (message.has("uid")) {
            uidFuture.complete(message.get("uid").asText());
            return;
        }

        final String topic = message.get("topic").asText();

        if ("block".equals(topic)) {
            final JsonNode block = message.get("data").get("block");
            final JsonNode blockMeta = message.get("data").get("meta");
            System.out.printf("New block: height=%,d hash=%s...%n",
                block.get("height").asLong(),
                blockMeta.get("hash").asText().substring(0, 16));
        }

        if ("finalizedBlock".equals(topic)) {
            final JsonNode finalized = message.get("data");
            System.out.printf("Finalized: height=%,d hash=%s...%n",
                finalized.get("height").asLong(),
                finalized.get("hash").asText().substring(0, 16));
        }
    }

}

Download source

このスニペットでは、 NODE_URL 環境変数を使用して Symbol API ノード を設定します。 値が指定されない場合は、デフォルト値が使用されます。 WebSocket URL は、HTTP プロトコルを WebSocket プロトコルに置き換え、 /ws を追加することで NODE_URL から派生します。

プログラムは Ctrl+C で中断されるまで実行され、接続を閉じる前にサブスクライブ解除のステップをトリガーします。

コード解説⚓︎

WebSocket への接続⚓︎

13
14
15
16
17
    async with connect(WS_URL) as websocket:
        # Connect to websocket endpoint
        response = json.loads(await websocket.recv())
        uid = response['uid']
        print(f'Connected to {WS_URL} with uid {uid}')

 7
 8
 9
10
11
12
13
14
15
16
    const websocket = new WebSocket(WS_URL);

    // Connect to websocket endpoint
    const uid = await new Promise(resolve => {
        websocket.addEventListener('message', event => {
            const message = JSON.parse(event.data);
            resolve(message.uid);
        }, { once: true });
    });
    console.log(`Connected to ${WS_URL} with uid ${uid}`);

45
46
47
48
49
50
51
52
53
        final WebSocketContainer container =
            ContainerProvider.getWebSocketContainer();
        final Session session = container.connectToServer(
            this, URI.create(wsUrl));
        final RemoteEndpoint.Basic remote = session.getBasicRemote();

        // Connect to websocket endpoint
        final String uid = uidFuture.join();
        System.out.printf("Connected to %s with uid %s%n", wsUrl, uid);

最初のステップは、ノードの /ws エンドポイントへの WebSocket 接続を開くことです。 接続すると、サーバーは以降のすべてのサブスクリプションリクエストに含める必要がある一意の識別子（ uid ）を含むメッセージを送信します。

チャネルのサブスクライブ⚓︎

19
20
21
22
23
24
25
26
27
        # Subscribe to block channel
        await websocket.send(json.dumps(
            {'uid': uid, 'subscribe': 'block'}))
        print('Subscribed to block channel')

        # Subscribe to finalizedBlock channel
        await websocket.send(json.dumps(
            {'uid': uid, 'subscribe': 'finalizedBlock'}))
        print('Subscribed to finalizedBlock channel')

18
19
20
21
22
23
24
    // Subscribe to block channel
    websocket.send(JSON.stringify({ uid, subscribe: 'block' }));
    console.log('Subscribed to block channel');

    // Subscribe to finalizedBlock channel
    websocket.send(JSON.stringify({ uid, subscribe: 'finalizedBlock' }));
    console.log('Subscribed to finalizedBlock channel');

55
56
57
58
59
60
61
62
63
64
65
        // Subscribe to block channel
        remote.sendText(JSON_MAPPER.createObjectNode()
            .put("uid", uid).put("subscribe", "block")
            .toString());
        System.out.println("Subscribed to block channel");

        // Subscribe to finalizedBlock channel
        remote.sendText(JSON_MAPPER.createObjectNode()
            .put("uid", uid).put("subscribe", "finalizedBlock")
            .toString());
        System.out.println("Subscribed to finalizedBlock channel");

コードは2つのチャネルをサブスクライブします。

  • block WS: 新しいブロックが生成されるたびに（約30秒ごとに）通知します。
  • finalizedBlock WS: ファイナライズラウンドが完了するたびに（約10〜20分ごとに）通知します。

各サブスクリプションメッセージには、接続ステップで受信した uid とチャネル名が含まれます。

メッセージの処理⚓︎

29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
        # Handle incoming messages
        try:
            async for raw_message in websocket:
                message = json.loads(raw_message)
                topic = message['topic']

                if topic == 'block':
                    block = message['data']['block']
                    block_meta = message['data']['meta']
                    print(
                        f'New block: height={int(block["height"]):,}'
                        f' hash={block_meta["hash"][:16]}...'
                    )

                if topic == 'finalizedBlock':
                    finalized = message['data']
                    print(
                        f'Finalized: height={int(finalized["height"]):,}'
                        f' hash={finalized["hash"][:16]}...'
                    )

26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
    // Handle incoming messages
    websocket.addEventListener('message', event => {
        const message = JSON.parse(event.data);
        const topic = message.topic;

        if ('block' === topic) {
            const block = message.data.block;
            const blockMeta = message.data.meta;
            console.log(
                `New block: height=${
                    BigInt(block.height).toLocaleString()}` +
                ` hash=${blockMeta.hash.substring(0, 16)}...`);
        }

        if ('finalizedBlock' === topic) {
            const finalized = message.data;
            console.log(
                `Finalized: height=${
                    BigInt(finalized.height).toLocaleString()}` +
                ` hash=${finalized.hash.substring(0, 16)}...`);
        }
    });

 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
    // Handle incoming messages
    @OnMessage
    public void onMessage(final String payload) throws IOException {
        final JsonNode message = JSON_MAPPER.readTree(payload);

        // Special case for the initial handshake message
        if (message.has("uid")) {
            uidFuture.complete(message.get("uid").asText());
            return;
        }

        final String topic = message.get("topic").asText();

        if ("block".equals(topic)) {
            final JsonNode block = message.get("data").get("block");
            final JsonNode blockMeta = message.get("data").get("meta");
            System.out.printf("New block: height=%,d hash=%s...%n",
                block.get("height").asLong(),
                blockMeta.get("hash").asText().substring(0, 16));
        }

        if ("finalizedBlock".equals(topic)) {
            final JsonNode finalized = message.get("data");
            System.out.printf("Finalized: height=%,d hash=%s...%n",
                finalized.get("height").asLong(),
                finalized.get("hash").asText().substring(0, 16));
        }
    }

コードは、プログラムが中断されるまで受信メッセージをリスニングします。 各メッセージには、チャネルを識別する topic フィールドと、イベントペイロードを含む data オブジェクトが含まれます。

block メッセージの場合、ペイロードは BlockInfoDTO スキーマに従います。 このチュートリアルでは、各ブロックを識別するためにそのうちの2つを使用します。

  • data.block.height: 新しいブロックの高さ。
  • data.meta.hash: 新しいブロックのハッシュ。

finalizedBlock メッセージの場合、ペイロードは FinalizedBlockDTO スキーマに従います。 このチュートリアルでは以下を使用します。

  • data.height: ファイナライズされたブロックの高さ。
  • data.hash: ファイナライズされたブロックのハッシュ。

チェーンの高さは、新しいブロックが生成されるたびに増加します。 ファイナライズは通常、ブロック生成の10〜20分後に行われるため、ファイナライズされた高さはチェーンの先端（最新高）より遅れます。

投票ノード がこのプロセスをどのように推進するかについての詳細は、テキストブックの コンセンサス セクションを参照してください。

終了時のサブスクライブ解除⚓︎

50
51
52
53
54
55
56
        # Unsubscribe on exit
        finally:
            await websocket.send(json.dumps(
                {'uid': uid, 'unsubscribe': 'block'}))
            await websocket.send(json.dumps(
                {'uid': uid, 'unsubscribe': 'finalizedBlock'}))
            print('Unsubscribed from all channels')

49
50
51
52
53
54
55
56
57
    // Unsubscribe on exit
    process.on('SIGINT', () => {
        websocket.send(JSON.stringify({ uid, unsubscribe: 'block' }));
        websocket.send(
            JSON.stringify({ uid, unsubscribe: 'finalizedBlock' }));
        console.log('Unsubscribed from all channels');
        websocket.close();
        process.exit(0);
    });

67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
        // Unsubscribe on exit
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                remote.sendText(JSON_MAPPER.createObjectNode()
                    .put("uid", uid).put("unsubscribe", "block")
                    .toString());
                remote.sendText(JSON_MAPPER.createObjectNode()
                    .put("uid", uid).put("unsubscribe", "finalizedBlock")
                    .toString());
                System.out.println("Unsubscribed from all channels");
                session.close();
            } catch (final IOException ex) {
                throw new IllegalStateException(ex);
            }
        }));

プログラムが中断されたとき（ Ctrl+C ）、コードは接続を閉じる前に両方のチャネルのサブスクライブ解除メッセージを送信します。 これにより、ノードからのクリーンな切断が保証されます。

出力⚓︎

以下の出力は、新しいブロックとファイナライズイベントをリスニングする典型的な実行例を示しています。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
Using node https://reference.symboltest.net:3001
Connected to wss://reference.symboltest.net:3001/ws with uid 9AQEv+DFuCuddfrJlNh7ERf8Zlg=
Subscribed to block channel
Subscribed to finalizedBlock channel
New block: height=3,176,948 hash=EDD1ED4C92E29655...
New block: height=3,176,949 hash=18DBF7E75B2CC003...
New block: height=3,176,950 hash=A3409E1951E8FC8C...
Finalized: height=3,176,948 hash=EDD1ED4C92E29655...
New block: height=3,176,951 hash=A7D80AFD75C4C918...
New block: height=3,176,952 hash=027D939F6E9D0DAE...
Unsubscribed from all channels

出力の主なポイント:

  • 接続 (2行目): wss:// URL への WebSocket 接続が確立され、サーバーは一意の uid を返します。
  • サブスクリプション (3-4行目): block チャネルと finalizedBlock チャネルの両方がサブスクライブされます。
  • 新しいブロック (5-7行目、9-10行目): 新しいブロックの通知が約30秒ごとに届きます。
  • ファイナライズ (8行目): ファイナライズラウンドが完了すると、一度に複数のブロックをカバーするファイナライズ通知が届きます。
  • サブスクライブ解除 (11行目): Ctrl+C を押すと、コードは両方のチャネルのサブスクライブを解除します。

結論⚓︎

このチュートリアルでは、以下の方法を説明しました。

ステップ 関連ドキュメント
block チャネルのサブスクライブ block WS
finalized チャネルのサブスクライブ finalizedBlock WS
ブロックメッセージの処理 BlockInfoDTO
ファイナライズメッセージの処理 FinalizedBlockDTO