コンテンツにスキップ

トランザクションフローをリッスンする⚓︎

中級

NEM は、特定の アカウント の トランザクション が承認プロセスを進むとき、リアルタイム通知を送信する WebSocket チャネル を提供します。 /transaction/get GET エンドポイントをポーリングする場合と比べ、WebSocket は API を繰り返し呼び出すオーバーヘッドなしに、発生した更新をプッシュします。

このチュートリアルでは、トランザクションチャネルをサブスクライブし、最小限の 転送トランザクション をアナウンスして、WebSocket で承認を待つ方法を説明します。

別の方法: ポーリング

ポーリングを使う方法については、トランザクションのステータスを監視する チュートリアルを参照してください。

前提条件⚓︎

始める前に、次の準備をしてください。

さらに、NEM は SockJS 上で STOMP メッセージングプロトコルを使って WebSocket を提供するため、STOMP クライアントと WebSocket トランスポートが必要です。

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

pip install stomper websockets

@stomp/stompjs と sockjs-client ライブラリをインストールします。

npm install @stomp/stompjs sockjs-client

このチュートリアルでは 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
 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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
import asyncio
import json
import os
import random
import urllib.request
import uuid

import stomper
from symbolchain.CryptoTypes import PrivateKey
from symbolchain.facade.NemFacade import NemFacade
from symbolchain.nc import Amount
from symbolchain.nem.FeeCalculator import calculate_transaction_fee
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}')


def announce_transaction(payload, endpoint, label):
    request = urllib.request.Request(
        f'{NODE_URL}{endpoint}',
        data=payload.encode(),
        headers={'Content-Type': 'application/json'},
        method='POST'
    )
    with urllib.request.urlopen(request) as response:
        result = json.loads(response.read().decode())
    if 'SUCCESS' == result['message']:
        print(label)
    return result['message']


# SockJS has no Python client library.
# These helpers wrap the raw WebSocket transport to mirror a STOMP client.
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('', '', NODE_URL, heartbeats=(0, 0)))


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


async def stomp_send(websocket, destination, body):
    await send_frame(websocket, stomper.send(destination, body))


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 each STOMP MESSAGE frame 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 frame


async def stomp_frames(websocket):
    # Yield each STOMP MESSAGE frame as it arrives
    async for raw_frame in websocket:
        for frame in stomp_messages(raw_frame):
            yield frame


# Set up the monitored address and signer
MONITOR_ADDRESS = os.getenv(
    'MONITOR_ADDRESS',
    'TBULEAUG2CZQISUR442HWA6UAKGWIXHDABJVIPS4'
)
print(f'Monitoring address: {MONITOR_ADDRESS}')

SIGNER_PRIVATE_KEY = os.getenv(
    'SIGNER_PRIVATE_KEY',
    '0000000000000000000000000000000000000000000000000000000000000000'
)
facade = NemFacade('testnet')
signer_key_pair = NemFacade.KeyPair(PrivateKey(SIGNER_PRIVATE_KEY))


async def main():
    # Build and sign a transfer to the monitored address
    transaction = facade.create_transaction_from_descriptor({
        'type': 'transfer_transaction_v2',
        'recipient_address': MONITOR_ADDRESS,
        'amount': 0,
    }, signer_key_pair.public_key, 0, 2 * 60 * 60)
    transaction.fee = Amount(calculate_transaction_fee(transaction))
    signature = facade.sign_transaction(signer_key_pair, transaction)
    json_payload = facade.transaction_factory.attach_signature(
        transaction, signature)
    transaction_hash = str(
        facade.hash_transaction(transaction)).upper()

    # Connect to the WebSocket
    endpoint = f'{WS_URL}/w/messages'
    async with connect(sockjs_url(endpoint)) as websocket:
        await stomp_connect(websocket)
        print(f'Connected to {WS_URL}')
        frames = stomp_frames(websocket)

        # Subscribe to the account and transaction channels
        account_channel = f'/account/{MONITOR_ADDRESS}'
        channels = {
            account_channel: 'id-0',
            f'/unconfirmed/{MONITOR_ADDRESS}': 'id-1',
            f'/transactions/{MONITOR_ADDRESS}': 'id-2',
        }
        for channel, sub_id in channels.items():
            await stomp_subscribe(websocket, channel, sub_id)
            print(f'Subscribed to {channel} channel')

        # Register the account and confirm it is active
        await stomp_send(websocket, '/w/api/account/get',
            json.dumps({'account': MONITOR_ADDRESS}))
        async for frame in frames:
            if account_channel == frame['headers']['destination']:
                balance = json.loads(
                    frame['body'])['account']['balance']
                print(f'Account update: balance={balance}')
                break
        print('Account registered')

        # Announce the transaction
        announce_result = announce_transaction(
            json_payload, '/transaction/announce',
            f'Announcing transaction {transaction_hash[:16]}...')

        # Wait for the transaction to confirm
        if 'SUCCESS' == announce_result:
            confirmed = False
            async for frame in frames:
                destination = frame['headers']['destination']
                body = json.loads(frame['body'])
                if account_channel == destination:
                    balance = body['account']['balance']
                    print(f'Account update: balance={balance}')
                    if confirmed:
                        break
                elif '/transactions/' in destination:
                    message_hash = body['meta']['hash']['data']
                    print(f'confirmed: hash={message_hash[:16]}...')
                    if message_hash.upper() == transaction_hash:
                        short_hash = transaction_hash[:16]
                        print(f'Transaction {short_hash}... confirmed')
                        confirmed = True
                else:
                    message_hash = body['meta']['hash']['data']
                    if message_hash.upper() == transaction_hash:
                        print(f'unconfirmed: hash={message_hash[:16]}...')
        else:
            print(f'Transaction rejected: {announce_result}')

        # Unsubscribe before closing
        for sub_id in channels.values():
            await stomp_unsubscribe(websocket, sub_id)
        print('Unsubscribed from all channels')
        await stomp_disconnect(websocket)


try:
    asyncio.run(main())
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
 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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
import { Client } from '@stomp/stompjs';
import SockJS from 'sockjs-client';
import { PrivateKey } from 'symbol-sdk';
import {
    Address, NemFacade, calculateTransactionFee, descriptors, models
} from 'symbol-sdk/nem';

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

async function announceTransaction(payload, endpoint, label) {
    const response = await fetch(`${NODE_URL}${endpoint}`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: payload
    });
    if (!response.ok)
        throw new Error(`HTTP ${response.status}`);
    const result = await response.json();
    if ('SUCCESS' === result.message)
        console.log(label);
    return result.message;
}

// Set up the monitored address and signer
const MONITOR_ADDRESS = process.env.MONITOR_ADDRESS ||
    'TBULEAUG2CZQISUR442HWA6UAKGWIXHDABJVIPS4';
console.log(`Monitoring address: ${MONITOR_ADDRESS}`);

const SIGNER_PRIVATE_KEY = process.env.SIGNER_PRIVATE_KEY ||
    '0000000000000000000000000000000000000000000000000000000000000000';
const facade = new NemFacade('testnet');
const signerKeyPair = new NemFacade.KeyPair(
    new PrivateKey(SIGNER_PRIVATE_KEY));

try {
    // Build and sign a transfer to the monitored address
    const transaction = facade.createTransactionFromTypedDescriptor(
        new descriptors.TransferTransactionV2Descriptor(
            new Address(MONITOR_ADDRESS),
            new models.Amount(0n)),
        signerKeyPair.publicKey,
        0n,
        2 * 60 * 60);
    transaction.fee = new models.Amount(
        calculateTransactionFee(transaction));
    const signature = facade.signTransaction(signerKeyPair, transaction);
    const jsonPayload = facade.transactionFactory.static.attachSignature(
        transaction, signature);
    const transactionHash =
        facade.hashTransaction(transaction).toString().toUpperCase();
    const shortHash = transactionHash.substring(0, 16);

    // Connect to the WebSocket
    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}`);

    // Wait for the transaction to confirm
    let confirmed = false;
    let resolveRegistered;
    let resolveDone;
    const registered = new Promise(resolve => {
        resolveRegistered = resolve;
    });
    const done = new Promise(resolve => {
        resolveDone = resolve;
    });
    const onUnconfirmed = message => {
        const messageHash = JSON.parse(message.body).meta.hash.data;
        if (messageHash.toUpperCase() === transactionHash) {
            console.log(
                `unconfirmed: hash=${messageHash.substring(0, 16)}...`);
        }
    };
    const onConfirmed = message => {
        const messageHash = JSON.parse(message.body).meta.hash.data;
        console.log(`confirmed: hash=${messageHash.substring(0, 16)}...`);
        if (messageHash.toUpperCase() === transactionHash) {
            console.log(`Transaction ${shortHash}... confirmed`);
            confirmed = true;
        }
    };
    const onAccountUpdate = message => {
        const { balance } = JSON.parse(message.body).account;
        console.log(`Account update: balance=${balance}`);
        resolveRegistered();
        if (confirmed)
            resolveDone();
    };

    // Subscribe to the account and transaction channels
    const accountChannel = `/account/${MONITOR_ADDRESS}`;
    const subscriptions = [
        { channel: accountChannel, handler: onAccountUpdate, id: 'id-0' },
        {
            channel: `/unconfirmed/${MONITOR_ADDRESS}`,
            handler: onUnconfirmed,
            id: 'id-1'
        },
        {
            channel: `/transactions/${MONITOR_ADDRESS}`,
            handler: onConfirmed,
            id: 'id-2'
        }
    ];
    for (const { channel, handler, id } of subscriptions) {
        client.subscribe(channel, handler, { id });
        console.log(`Subscribed to ${channel} channel`);
    }

    // Register the account and confirm it is active
    client.publish({
        destination: '/w/api/account/get',
        body: JSON.stringify({ account: MONITOR_ADDRESS })
    });
    await registered;
    console.log('Account registered');

    // Announce the transaction
    const announceResult = await announceTransaction(
        jsonPayload, '/transaction/announce',
        `Announcing transaction ${shortHash}...`);

    // Wait for the transaction to confirm
    if ('SUCCESS' === announceResult)
        await done;
    else
        console.log(`Transaction rejected: ${announceResult}`);
    // Unsubscribe before closing
    for (const { id } of subscriptions)
        client.unsubscribe(id);
    console.log('Unsubscribed from all channels');
    client.deactivate();
} 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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
//JAVA 21+
//DEPS org.symbol:symbol-sdk:3.3.3
//DEPS org.glassfish.tyrus.bundles:tyrus-standalone-client:2.2.0

import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandlers;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
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 org.symbol.sdk.CryptoTypes;
import org.symbol.sdk.facade.NemFacade;
import org.symbol.sdk.nem.Address;
import org.symbol.sdk.nem.FeeCalculator;
import org.symbol.sdk.nem.KeyPair;
import org.symbol.sdk.nem.NemTransactionFactory;
import org.symbol.sdk.nem.descriptors.TransferTransactionV2Descriptor;
import org.symbol.sdk.nem.models.Amount;
import org.symbol.sdk.nem.models.Transaction;

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

    private static final HttpClient HTTP_CLIENT =
        HttpClient.newHttpClient();

    private final BlockingQueue<String> rawFrames =
        new LinkedBlockingQueue<>();

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

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

    private final NemFacade facade = new NemFacade("testnet");

    private String announceTransaction(
        final String payload,
        final String endpoint,
        final String label
    ) throws IOException, InterruptedException {
        final HttpRequest request = HttpRequest.newBuilder(
            URI.create(nodeUrl + endpoint))
            .header("Content-Type", "application/json")
            .POST(HttpRequest.BodyPublishers.ofString(payload))
            .build();
        final HttpResponse<String> response = HTTP_CLIENT.send(
            request, BodyHandlers.ofString());
        if (2 != response.statusCode() / 100)
            throw new IOException("HTTP " + response.statusCode());
        final String result = JSON_MAPPER.readTree(response.body())
            .get("message").asText();
        if ("SUCCESS".equals(result))
            System.out.println(label);
        return result;
    }

    public static void main(final String[] args) {
        try {
            new ListenTransactionFlow().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);

        // Set up the monitored address and signer
        final String monitorAddress = System.getenv().getOrDefault(
            "MONITOR_ADDRESS",
            "TBULEAUG2CZQISUR442HWA6UAKGWIXHDABJVIPS4");
        System.out.printf("Monitoring address: %s%n", monitorAddress);
        final String privateKey = System.getenv().getOrDefault(
            "SIGNER_PRIVATE_KEY", "0".repeat(64));
        final KeyPair signerKeyPair = new KeyPair(
            new CryptoTypes.PrivateKey(privateKey));

        // Build and sign a transfer to the monitored address
        final Transaction transaction =
            facade.createTransactionFromTypedDescriptor(
                new TransferTransactionV2Descriptor(
                    new Address(monitorAddress), new Amount(0),
                    null, List.of()),
                signerKeyPair.getPublicKey(), 0, 2 * 60 * 60);
        transaction.setFee(new Amount(
            FeeCalculator.calculateTransactionFee(transaction)));
        final String payload = NemTransactionFactory.attachSignature(
            transaction,
            facade.signTransaction(signerKeyPair, transaction));
        final String transactionHash = facade.hashTransaction(transaction)
            .toString().toUpperCase();

        // Connect to the WebSocket
        final Session session = ContainerProvider.getWebSocketContainer()
            .connectToServer(this,
                URI.create(sockJsUrl(wsUrl + "/w/messages")));
        final RemoteEndpoint.Basic remote = session.getBasicRemote();
        rawFrames.take(); // consume the SockJS open frame
        sendFrame(remote, "CONNECT\naccept-version:1.1,1.0\n"
            + "heart-beat:0,0\nhost:" + nodeUrl + "\n\n\0");
        System.out.printf("Connected to %s%n", wsUrl);

        // Subscribe to the account and transaction channels
        final String accountChannel = "/account/" + monitorAddress;
        final Map<String, String> channels = new HashMap<>();
        channels.put(accountChannel, "id-0");
        channels.put("/unconfirmed/" + monitorAddress, "id-1");
        channels.put("/transactions/" + monitorAddress, "id-2");
        for (final Map.Entry<String, String> channel
            : channels.entrySet()) {
            subscribe(remote, channel.getKey(), channel.getValue());
            System.out.printf("Subscribed to %s channel%n",
                channel.getKey());
        }

        // Register the account and confirm it is active
        send(remote, "/w/api/account/get", JSON_MAPPER.createObjectNode()
            .put("account", monitorAddress).toString());
        while (true) {
            final StompMessage message = nextMessage();
            if (accountChannel.equals(message.destination())) {
                final String balance = JSON_MAPPER.readTree(message.body())
                    .get("account").get("balance").asText();
                System.out.printf("Account update: balance=%s%n", balance);
                break;
            }
        }
        System.out.println("Account registered");

        // Announce the transaction
        final String result = announceTransaction(payload,
            "/transaction/announce", "Announcing transaction "
                + transactionHash.substring(0, 16) + "...");

        // Wait for the transaction to confirm
        if ("SUCCESS".equals(result)) {
            boolean confirmed = false;
            while (true) {
                final StompMessage message = nextMessage();
                final JsonNode body = JSON_MAPPER.readTree(message.body());
                if (accountChannel.equals(message.destination())) {
                    System.out.printf("Account update: balance=%s%n",
                        body.get("account").get("balance").asText());
                    if (confirmed)
                        break;
                } else {
                    final String hash = body.get("meta").get("hash")
                        .get("data").asText();
                    if (message.destination().contains("/transactions/")) {
                        System.out.printf("confirmed: hash=%s...%n",
                            hash.substring(0, 16));
                        if (hash.equalsIgnoreCase(transactionHash)) {
                            System.out.printf(
                                "Transaction %s... confirmed%n",
                                transactionHash.substring(0, 16));
                            confirmed = true;
                        }
                    } else if (hash.equalsIgnoreCase(transactionHash)) {
                        System.out.printf("unconfirmed: hash=%s...%n",
                            hash.substring(0, 16));
                    }
                }
            }
        } else {
            System.out.printf("Transaction rejected: %s%n", result);
        }

        // Unsubscribe before closing
        for (final String id : channels.values())
            sendFrame(remote, "UNSUBSCRIBE\nid:" + id + "\n\n\0");
        System.out.println("Unsubscribed from all channels");
        sendFrame(remote, "DISCONNECT\n\n\0");
        session.close();
    }

    @OnMessage
    public void onMessage(final String frame) {
        rawFrames.add(frame);
    }

    private StompMessage nextMessage() throws Exception {
        while (true) {
            final String raw = rawFrames.take();
            if (!raw.startsWith("a"))
                continue;
            for (final JsonNode item
                : JSON_MAPPER.readTree(raw.substring(1))) {
                final String frame = item.asText();
                if (!frame.startsWith("MESSAGE\n"))
                    continue;
                final int separator = frame.indexOf("\n\n");
                final String headers = frame.substring(0, separator);
                String destination = "";
                for (final String line : headers.split("\n"))
                    if (line.startsWith("destination:"))
                        destination = line.substring(12);
                final String body = frame.substring(separator + 2)
                    .replace("\0", "");
                return new StompMessage(destination, body);
            }
        }
    }

    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 subscribe(
        final RemoteEndpoint.Basic remote,
        final String destination,
        final String id
    ) throws IOException {
        sendFrame(remote, "SUBSCRIBE\nid:" + id + "\ndestination:"
            + destination + "\nack:auto\n\n\0");
    }

    private static void send(
        final RemoteEndpoint.Basic remote,
        final String destination,
        final String body
    ) throws IOException {
        sendFrame(remote, "SEND\ndestination:" + destination
            + "\ncontent-type:application/json\ncontent-length:"
            + body.length() + "\n\n" + body + "\0");
    }

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

    private record StompMessage(String destination, String body) {}
}

Download source

スニペットでは、NODE_URL 環境変数を使って NEM ノード を指定します。 値が指定されていない場合は、デフォルト値を使用します。

WS_URL は同じノードの WebSocket エンドポイントを定義します。 NODE_URL のポート 7890（デフォルトの HTTP API ポート）を 7778（デフォルトの NIS WebSocket ポート）に置き換えて導出します。

Python の SockJS ヘルパー

Python 用の SockJS クライアントライブラリはないため、便宜上、ファイルの先頭に小さなヘルパーメソッドをいくつか定義しています。

コードの説明⚓︎

監視対象アドレスと署名者をセットアップする⚓︎

 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
# Set up the monitored address and signer
MONITOR_ADDRESS = os.getenv(
    'MONITOR_ADDRESS',
    'TBULEAUG2CZQISUR442HWA6UAKGWIXHDABJVIPS4'
)
print(f'Monitoring address: {MONITOR_ADDRESS}')

SIGNER_PRIVATE_KEY = os.getenv(
    'SIGNER_PRIVATE_KEY',
    '0000000000000000000000000000000000000000000000000000000000000000'
)
facade = NemFacade('testnet')
signer_key_pair = NemFacade.KeyPair(PrivateKey(SIGNER_PRIVATE_KEY))

27
28
29
30
31
32
33
34
35
36
// Set up the monitored address and signer
const MONITOR_ADDRESS = process.env.MONITOR_ADDRESS ||
    'TBULEAUG2CZQISUR442HWA6UAKGWIXHDABJVIPS4';
console.log(`Monitoring address: ${MONITOR_ADDRESS}`);

const SIGNER_PRIVATE_KEY = process.env.SIGNER_PRIVATE_KEY ||
    '0000000000000000000000000000000000000000000000000000000000000000';
const facade = new NemFacade('testnet');
const signerKeyPair = new NemFacade.KeyPair(
    new PrivateKey(SIGNER_PRIVATE_KEY));

89
90
91
92
93
94
95
96
97
        // Set up the monitored address and signer
        final String monitorAddress = System.getenv().getOrDefault(
            "MONITOR_ADDRESS",
            "TBULEAUG2CZQISUR442HWA6UAKGWIXHDABJVIPS4");
        System.out.printf("Monitoring address: %s%n", monitorAddress);
        final String privateKey = System.getenv().getOrDefault(
            "SIGNER_PRIVATE_KEY", "0".repeat(64));
        final KeyPair signerKeyPair = new KeyPair(
            new CryptoTypes.PrivateKey(privateKey));

この手順では、監視するアドレスと、そこへ送金するアカウントをセットアップします。

MONITOR_ADDRESS は監視するアドレスです。 このチュートリアルがサブスクライブするチャネルはこのアドレスに対応付けられ、送金の送信者や受取人など、トランザクションに関係するたびに通知します。 WebSocket API では、アドレスを大文字かつハイフンなしで指定します。

SIGNER_PRIVATE_KEY は送金を送信するアカウントの秘密鍵で、通知を発生させます。

これらの環境変数が設定されていない場合は、チュートリアルがデフォルト値を用意します。

転送トランザクションを構築して署名する⚓︎

104
105
106
107
108
109
110
111
112
113
114
115
    # Build and sign a transfer to the monitored address
    transaction = facade.create_transaction_from_descriptor({
        'type': 'transfer_transaction_v2',
        'recipient_address': MONITOR_ADDRESS,
        'amount': 0,
    }, signer_key_pair.public_key, 0, 2 * 60 * 60)
    transaction.fee = Amount(calculate_transaction_fee(transaction))
    signature = facade.sign_transaction(signer_key_pair, transaction)
    json_payload = facade.transaction_factory.attach_signature(
        transaction, signature)
    transaction_hash = str(
        facade.hash_transaction(transaction)).upper()

39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
    // Build and sign a transfer to the monitored address
    const transaction = facade.createTransactionFromTypedDescriptor(
        new descriptors.TransferTransactionV2Descriptor(
            new Address(MONITOR_ADDRESS),
            new models.Amount(0n)),
        signerKeyPair.publicKey,
        0n,
        2 * 60 * 60);
    transaction.fee = new models.Amount(
        calculateTransactionFee(transaction));
    const signature = facade.signTransaction(signerKeyPair, transaction);
    const jsonPayload = facade.transactionFactory.static.attachSignature(
        transaction, signature);
    const transactionHash =
        facade.hashTransaction(transaction).toString().toUpperCase();
    const shortHash = transactionHash.substring(0, 16);

 99
100
101
102
103
104
105
106
107
108
109
110
111
112
        // Build and sign a transfer to the monitored address
        final Transaction transaction =
            facade.createTransactionFromTypedDescriptor(
                new TransferTransactionV2Descriptor(
                    new Address(monitorAddress), new Amount(0),
                    null, List.of()),
                signerKeyPair.getPublicKey(), 0, 2 * 60 * 60);
        transaction.setFee(new Amount(
            FeeCalculator.calculateTransactionFee(transaction)));
        final String payload = NemTransactionFactory.attachSignature(
            transaction,
            facade.signTransaction(signerKeyPair, transaction));
        final String transactionHash = facade.hashTransaction(transaction)
            .toString().toUpperCase();

このチュートリアルでは、監視対象アドレスへ、金額 0、モザイクなし、メッセージなしの最小限の 転送トランザクション を構築します。 簡単にするため送金を使いますが、どのトランザクションタイプでも同じ WebSocket 通知を発生させます。

トランザクションは XEM を送信する チュートリアルと同じ方法で作成して署名します。 ファサードは署名者を追加し、2 時間のデッドライン期間からタイムスタンプとデッドラインを導出します。

トランザクションに署名するとハッシュが生成され、一意に識別できるようになります。 コードはこのハッシュを保存します。トランザクションチャネルの通知にはトランザクションハッシュが含まれるためです。 後で、受信した各ハッシュを保存値と比較して、このトランザクションの通知を特定します。

トランザクションは準備されますが、まだ アナウンス されません。 チャネルのサブスクリプションを確立した後でアナウンスするため、結果の通知を取り逃しません。

WebSocket に接続する⚓︎

117
118
119
120
121
122
    # Connect to the WebSocket
    endpoint = f'{WS_URL}/w/messages'
    async with connect(sockjs_url(endpoint)) as websocket:
        await stomp_connect(websocket)
        print(f'Connected to {WS_URL}')
        frames = stomp_frames(websocket)

56
57
58
59
60
61
62
63
64
    // Connect to the WebSocket
    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}`);

114
115
116
117
118
119
120
121
122
        // Connect to the WebSocket
        final Session session = ContainerProvider.getWebSocketContainer()
            .connectToServer(this,
                URI.create(sockJsUrl(wsUrl + "/w/messages")));
        final RemoteEndpoint.Basic remote = session.getBasicRemote();
        rawFrames.take(); // consume the SockJS open frame
        sendFrame(remote, "CONNECT\naccept-version:1.1,1.0\n"
            + "heart-beat:0,0\nhost:" + nodeUrl + "\n\n\0");
        System.out.printf("Connected to %s%n", wsUrl);

コードは WS_URL の /w/messages エンドポイントへ SockJS 接続を開き、その上で STOMP セッション を開始します。

チャネルをサブスクライブする⚓︎

124
125
126
127
128
129
130
131
132
133
        # Subscribe to the account and transaction channels
        account_channel = f'/account/{MONITOR_ADDRESS}'
        channels = {
            account_channel: 'id-0',
            f'/unconfirmed/{MONITOR_ADDRESS}': 'id-1',
            f'/transactions/{MONITOR_ADDRESS}': 'id-2',
        }
        for channel, sub_id in channels.items():
            await stomp_subscribe(websocket, channel, sub_id)
            print(f'Subscribed to {channel} channel')

 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
    // Subscribe to the account and transaction channels
    const accountChannel = `/account/${MONITOR_ADDRESS}`;
    const subscriptions = [
        { channel: accountChannel, handler: onAccountUpdate, id: 'id-0' },
        {
            channel: `/unconfirmed/${MONITOR_ADDRESS}`,
            handler: onUnconfirmed,
            id: 'id-1'
        },
        {
            channel: `/transactions/${MONITOR_ADDRESS}`,
            handler: onConfirmed,
            id: 'id-2'
        }
    ];
    for (const { channel, handler, id } of subscriptions) {
        client.subscribe(channel, handler, { id });
        console.log(`Subscribed to ${channel} channel`);
    }

124
125
126
127
128
129
130
131
132
133
134
135
        // Subscribe to the account and transaction channels
        final String accountChannel = "/account/" + monitorAddress;
        final Map<String, String> channels = new HashMap<>();
        channels.put(accountChannel, "id-0");
        channels.put("/unconfirmed/" + monitorAddress, "id-1");
        channels.put("/transactions/" + monitorAddress, "id-2");
        for (final Map.Entry<String, String> channel
            : channels.entrySet()) {
            subscribe(remote, channel.getKey(), channel.getValue());
            System.out.printf("Subscribed to %s channel%n",
                channel.getKey());
        }

コードは、アドレスに対応付けられた次の 3 チャネルをサブスクライブします。

サブスクリプションには id-0、id-1、id-2 を使います。プログラムが最後にサブスクライブを解除するときに、それぞれを識別します。

メッセージ処理の違い

JavaScript では、各チャネルを専用のハンドラー関数でサブスクライブします。関数は下の 承認を待つ 手順で定義します。 Python では、接続から到着したメッセージを順番に読み取ります。

3 つのチャネルは、次の手順でアドレスを登録するまで何も送信しません。

アカウントを登録する⚓︎

135
136
137
138
139
140
141
142
143
144
        # Register the account and confirm it is active
        await stomp_send(websocket, '/w/api/account/get',
            json.dumps({'account': MONITOR_ADDRESS}))
        async for frame in frames:
            if account_channel == frame['headers']['destination']:
                balance = json.loads(
                    frame['body'])['account']['balance']
                print(f'Account update: balance={balance}')
                break
        print('Account registered')

119
120
121
122
123
124
125
    // Register the account and confirm it is active
    client.publish({
        destination: '/w/api/account/get',
        body: JSON.stringify({ account: MONITOR_ADDRESS })
    });
    await registered;
    console.log('Account registered');

137
138
139
140
141
142
143
144
145
146
147
148
149
        // Register the account and confirm it is active
        send(remote, "/w/api/account/get", JSON_MAPPER.createObjectNode()
            .put("account", monitorAddress).toString());
        while (true) {
            final StompMessage message = nextMessage();
            if (accountChannel.equals(message.destination())) {
                final String balance = JSON_MAPPER.readTree(message.body())
                    .get("account").get("balance").asText();
                System.out.printf("Account update: balance=%s%n", balance);
                break;
            }
        }
        System.out.println("Account registered");

アカウントのチャネルから通知を受け取るには、まずアドレスをノードに 登録 する必要があります。

コードは w/api/account/get REQ にリクエストを送信します。アドレスを登録するとともに、account/{address} WS チャネルでアカウントの現在の状態を送信するようノードに要求します。

コードは最初のアカウント通知を待ち、登録が有効になったことを確認します。 通知は AccountMetaDataPair スキーマに従います。

アカウントチャネルのサブスクリプションは実行中ずっと開いたままなので、トランザクションの承認で発生するアカウント通知も出力に表示されます。

トランザクションをアナウンスする⚓︎

146
147
148
149
        # Announce the transaction
        announce_result = announce_transaction(
            json_payload, '/transaction/announce',
            f'Announcing transaction {transaction_hash[:16]}...')

127
128
129
130
    // Announce the transaction
    const announceResult = await announceTransaction(
        jsonPayload, '/transaction/announce',
        `Announcing transaction ${shortHash}...`);

151
152
153
154
        // Announce the transaction
        final String result = announceTransaction(payload,
            "/transaction/announce", "Announcing transaction "
                + transactionHash.substring(0, 16) + "...");

チャネルをサブスクライブしてからアナウンスしてください

リスナーの準備ができていることを確実にするため、トランザクションは必ず WebSocket チャネルをサブスクライブした後にアナウンスしてください。 そうしないと、WebSocket がリッスンする前に通知が届く可能性があります。

コードは /transaction/announce POST エンドポイントへトランザクションをアナウンスし、結果を確認します。 ノードが拒否した場合は、拒否理由を表示して停止します。

承認を待つ⚓︎

151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
        # Wait for the transaction to confirm
        if 'SUCCESS' == announce_result:
            confirmed = False
            async for frame in frames:
                destination = frame['headers']['destination']
                body = json.loads(frame['body'])
                if account_channel == destination:
                    balance = body['account']['balance']
                    print(f'Account update: balance={balance}')
                    if confirmed:
                        break
                elif '/transactions/' in destination:
                    message_hash = body['meta']['hash']['data']
                    print(f'confirmed: hash={message_hash[:16]}...')
                    if message_hash.upper() == transaction_hash:
                        short_hash = transaction_hash[:16]
                        print(f'Transaction {short_hash}... confirmed')
                        confirmed = True
                else:
                    message_hash = body['meta']['hash']['data']
                    if message_hash.upper() == transaction_hash:
                        print(f'unconfirmed: hash={message_hash[:16]}...')
        else:
            print(f'Transaction rejected: {announce_result}')

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
    // Wait for the transaction to confirm
    let confirmed = false;
    let resolveRegistered;
    let resolveDone;
    const registered = new Promise(resolve => {
        resolveRegistered = resolve;
    });
    const done = new Promise(resolve => {
        resolveDone = resolve;
    });
    const onUnconfirmed = message => {
        const messageHash = JSON.parse(message.body).meta.hash.data;
        if (messageHash.toUpperCase() === transactionHash) {
            console.log(
                `unconfirmed: hash=${messageHash.substring(0, 16)}...`);
        }
    };
    const onConfirmed = message => {
        const messageHash = JSON.parse(message.body).meta.hash.data;
        console.log(`confirmed: hash=${messageHash.substring(0, 16)}...`);
        if (messageHash.toUpperCase() === transactionHash) {
            console.log(`Transaction ${shortHash}... confirmed`);
            confirmed = true;
        }
    };
    const onAccountUpdate = message => {
        const { balance } = JSON.parse(message.body).account;
        console.log(`Account update: balance=${balance}`);
        resolveRegistered();
        if (confirmed)
            resolveDone();
    };

156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
        // Wait for the transaction to confirm
        if ("SUCCESS".equals(result)) {
            boolean confirmed = false;
            while (true) {
                final StompMessage message = nextMessage();
                final JsonNode body = JSON_MAPPER.readTree(message.body());
                if (accountChannel.equals(message.destination())) {
                    System.out.printf("Account update: balance=%s%n",
                        body.get("account").get("balance").asText());
                    if (confirmed)
                        break;
                } else {
                    final String hash = body.get("meta").get("hash")
                        .get("data").asText();
                    if (message.destination().contains("/transactions/")) {
                        System.out.printf("confirmed: hash=%s...%n",
                            hash.substring(0, 16));
                        if (hash.equalsIgnoreCase(transactionHash)) {
                            System.out.printf(
                                "Transaction %s... confirmed%n",
                                transactionHash.substring(0, 16));
                            confirmed = true;
                        }
                    } else if (hash.equalsIgnoreCase(transactionHash)) {
                        System.out.printf("unconfirmed: hash=%s...%n",
                            hash.substring(0, 16));
                    }
                }
            }
        } else {
            System.out.printf("Transaction rejected: %s%n", result);
        }

受け付けられた場合、コードは承認を待ち、サブスクライブしたチャネルからの各メッセージを表示します。

トランザクションチャネルからのメッセージは TransactionMetaDataPair スキーマに従い、meta.hash.data フィールドにトランザクションハッシュを持ちます。 メッセージが届くたびに、コードはそのハッシュを保存値と比較して、チャネル通知の中からこのトランザクションを認識します。

成功したトランザクションで想定される順序は、トランザクションのライフサイクル に説明されています。

  1. unconfirmed: トランザクションが 未承認トランザクションプール に入る。
  2. confirmed: トランザクションが ブロック に含まれる。

トランザクションを含むブロックは、account/{address} WS チャネルでも最後の通知を発生させます。 トランザクションチャネルとは異なり、この通知にはトランザクションハッシュではなくアカウントの更新後の状態が含まれるため、特定のトランザクションとは照合できません。

この最後の通知が届くと、プログラムは後片付けの手順へ進みます。

チャネルのサブスクライブを解除する⚓︎

176
177
178
179
180
        # Unsubscribe before closing
        for sub_id in channels.values():
            await stomp_unsubscribe(websocket, sub_id)
        print('Unsubscribed from all channels')
        await stomp_disconnect(websocket)

137
138
139
140
141
    // Unsubscribe before closing
    for (const { id } of subscriptions)
        client.unsubscribe(id);
    console.log('Unsubscribed from all channels');
    client.deactivate();

189
190
191
192
193
194
        // Unsubscribe before closing
        for (final String id : channels.values())
            sendFrame(remote, "UNSUBSCRIBE\nid:" + id + "\n\n\0");
        System.out.println("Unsubscribed from all channels");
        sendFrame(remote, "DISCONNECT\n\n\0");
        session.close();

承認後、コードは 3 つのチャネルのサブスクライブを解除し、接続を閉じる前に STOMP セッションを終了します。

出力⚓︎

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
Using node http://libertalia.nemtest.net:7890
Monitoring address: TBULEAUG2CZQISUR442HWA6UAKGWIXHDABJVIPS4
Connected to http://libertalia.nemtest.net:7778
Subscribed to /account/TBULEAUG2CZQISUR442HWA6UAKGWIXHDABJVIPS4 channel
Subscribed to /unconfirmed/TBULEAUG2CZQISUR442HWA6UAKGWIXHDABJVIPS4 channel
Subscribed to /transactions/TBULEAUG2CZQISUR442HWA6UAKGWIXHDABJVIPS4 channel
Account update: balance=10035200000
Account registered
Announcing transaction 2928C2D9554AE127...
unconfirmed: hash=2928c2d9554ae127...
confirmed: hash=2928c2d9554ae127...
Transaction 2928C2D9554AE127... confirmed
Account update: balance=10035200000
Unsubscribed from all channels

出力には次の内容が表示されます。

  • アドレス（2 行目）: 監視対象のアドレス。
  • 接続（3 行目）: ノードのポート 7778 の WebSocket エンドポイント上で STOMP セッションが確立されます。
  • サブスクリプション（4～6 行目）: アカウントチャネルと 2 つのトランザクションチャネルがサブスクライブされます。
  • 登録（7～8 行目）: アカウントの現在の状態がアカウントチャネルに届き、登録を確認します。
  • アナウンス（9 行目）: トランザクションがアナウンスされ、そのハッシュが表示されます。
  • トランザクションフロー（10～11 行目）: トランザクションが unconfirmed から confirmed へ進み、承認のライフサイクルが表示されます。
  • 承認（12 行目）: transactions/{address} WS チャネルのハッシュが、アナウンスしたトランザクションと一致します。
  • アカウント更新（13 行目）: トランザクションを含むブロックが最後のアカウント通知を発生させます。送金額が 0 なので残高は変わりません。
  • サブスクライブ解除（14 行目）: コードが 3 つのチャネルのサブスクライブを解除します。

まとめ⚓︎

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

手順 関連ドキュメント
アカウントチャネルをサブスクライブする account/{address} WS
未承認チャネルをサブスクライブする unconfirmed/{address} WS
トランザクションチャネルをサブスクライブする transactions/{address} WS
アカウントを登録する w/api/account/get REQ
トランザクションメッセージを処理する TransactionMetaDataPair