From de309ca2f12e58b74347b863fd0a1426b4299837 Mon Sep 17 00:00:00 2001 From: toki Date: Sun, 5 Apr 2026 20:44:37 +0900 Subject: [PATCH] Update protocol and add WebSocket protobuf client/server implementations --- .codex | 0 PROTOCOL.md | 18 ++- dart/lib/src/ws_protobuf_client.dart | 154 +++++++++++++++++++++ dart/lib/src/ws_protobuf_server.dart | 85 ++++++++++++ dart/lib/toki_socket.dart | 2 + dart/test/socket_test.dart | 191 +++++++++++++++++++++++++++ 6 files changed, 448 insertions(+), 2 deletions(-) create mode 100644 .codex create mode 100644 dart/lib/src/ws_protobuf_client.dart create mode 100644 dart/lib/src/ws_protobuf_server.dart diff --git a/.codex b/.codex new file mode 100644 index 0000000..e69de29 diff --git a/PROTOCOL.md b/PROTOCOL.md index 946ebea..cd0586c 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -5,7 +5,9 @@ Designed for bidirectional, heterogeneous communication across languages and pla --- -## Wire Format +## Transport별 Wire Format + +### TCP / SSL+TCP ``` ┌──────────────────────────────────────────┐ @@ -15,11 +17,23 @@ Designed for bidirectional, heterogeneous communication across languages and pla └──────────────────────────────────────────┘ ``` -### Header +#### Header - 4 bytes, big-endian signed int32 - Value: byte length of the following `PacketBase` protobuf payload - Value of `0`: reserved / no-op, receiver clears buffer +### WebSocket / WSS + +``` +┌──────────────────────────────────────────┐ +│ PacketBase (protobuf, N bytes) │ ← typeName + nonce + data +└──────────────────────────────────────────┘ +``` + +- Length 헤더 없음 — WebSocket 프로토콜이 메시지 경계를 보장한다 +- Binary frame 사용 (text frame 아님) +- 연결: `ws://host:port/path` (plain) / `wss://host:port/path` (TLS) + ### PacketBase ```protobuf message PacketBase { diff --git a/dart/lib/src/ws_protobuf_client.dart b/dart/lib/src/ws_protobuf_client.dart new file mode 100644 index 0000000..85a9f1f --- /dev/null +++ b/dart/lib/src/ws_protobuf_client.dart @@ -0,0 +1,154 @@ +// ignore_for_file: avoid_init_to_null, prefer_final_fields, avoid_print + +import 'dart:io'; +import 'dart:async'; +import 'dart:typed_data'; + +import 'package:protobuf/protobuf.dart'; +import 'communicator.dart'; +import 'response_checker.dart'; +import 'packets/message_common.pb.dart'; + +abstract class WsProtobufClient extends Communicator { + late int _heartbeatIntervalTime; + late int _heartbeatWaitTime; + final WebSocket _ws; + + /// Plain WebSocket connection. + /// + /// ```dart + /// final ws = await WsProtobufClient.connect('localhost', 9090); + /// final client = MyClient(ws); + /// ``` + static Future connect(String host, int port, + {String path = '/'}) => + WebSocket.connect('ws://$host:$port$path'); + + /// Secure WebSocket (WSS) connection. + /// + /// ```dart + /// final ws = await WsProtobufClient.connectSecure('example.com', 443); + /// final client = MyClient(ws); + /// ``` + /// + /// For self-signed or custom certificates, provide a [SecurityContext]: + /// ```dart + /// final ctx = SecurityContext()..setTrustedCertificates('ca.crt'); + /// final ws = await WsProtobufClient.connectSecure('host', 443, context: ctx); + /// ``` + static Future connectSecure(String host, int port, + {String path = '/', SecurityContext? context}) => + WebSocket.connect('wss://$host:$port$path', + customClient: + HttpClient(context: context ?? SecurityContext.defaultContext)); + + late ResponseChecker? _heartbeatChecker = null; + bool _isAlive = false; + bool _waitingHeartbeatResponse = false; + int _nonce = 0; + List _onDisconnectListenerList = []; + + WsProtobufClient( + this._ws, + this._heartbeatIntervalTime, + this._heartbeatWaitTime, + Map)> parserMap) { + print('Connected New WS Client'); + _isAlive = true; + parserMap.addAll({(HeartBeat).toString(): HeartBeat.fromBuffer}); + super.initialize(parserMap); + _ws.listen(_onMessage, onError: onError, onDone: () => onDisconnected(null)); + addListener(onHeartBeat); + sendHeartBeat(); + } + + void sendHeartBeat() { + if (_isAlive) { + _heartbeatChecker?.responded(); + _heartbeatChecker = + ResponseChecker.second(this, _heartbeatIntervalTime, (client) { + if (_isAlive) { + _waitingHeartbeatResponse = true; + send(HeartBeat()); + _heartbeatChecker = + ResponseChecker.second(this, _heartbeatWaitTime, (client) { + if (_isAlive) { + onDisconnected(null); + dispose(); + } + }); + } + }); + } + } + + void onHeartBeat(HeartBeat data) { + if (_waitingHeartbeatResponse) { + _waitingHeartbeatResponse = false; + } else { + send(HeartBeat()); + } + } + + void addDisconnectListener(void Function(WsProtobufClient) handler) { + if (!_onDisconnectListenerList.contains(handler)) { + _onDisconnectListenerList.add(handler); + } + } + + void removeDisconnectListener(void Function(WsProtobufClient) handler) { + if (_onDisconnectListenerList.contains(handler)) { + _onDisconnectListenerList.remove(handler); + } + } + + void onDisconnected(dynamic data) { + if (_isAlive) { + for (var item in _onDisconnectListenerList) { + item.call(this); + } + _onDisconnectListenerList.clear(); + } + } + + void onError(dynamic e) { + print('=========> WS onError: $e'); + } + + void _onMessage(dynamic data) { + final bytes = data is List ? data : (data as Uint8List).toList(); + final common = PacketBase.fromBuffer(bytes); + onReceivedData(common.typeName, common.data); + sendHeartBeat(); + } + + /// Sends [data] as a binary WebSocket frame — no length header. + /// The WebSocket protocol guarantees message boundaries. + @override + Future send(T data) async { + if (_isAlive) { + try { + var base = PacketBase(); + base.typeName = data.info_.qualifiedMessageName; + base.nonce = ++_nonce; + base.data = data.writeToBuffer(); + _ws.add(base.writeToBuffer()); + } catch (e) { + onDisconnected(null); + } + } + return Future.value(); + } + + void dispose() async { + if (_isAlive) { + _isAlive = false; + _heartbeatChecker?.responded(); + try { + await _ws.close(); + } catch (_) { + // already closed by peer + } + } + } +} diff --git a/dart/lib/src/ws_protobuf_server.dart b/dart/lib/src/ws_protobuf_server.dart new file mode 100644 index 0000000..bdf9f94 --- /dev/null +++ b/dart/lib/src/ws_protobuf_server.dart @@ -0,0 +1,85 @@ +// ignore_for_file: avoid_print, prefer_final_fields + +import 'dart:io'; +import 'package:protobuf/protobuf.dart'; +import 'packets/message_common.pb.dart'; +import 'ws_protobuf_client.dart'; + +abstract class WsProtobufServer { + List _clientList = []; + bool _started = false; + HttpServer? _server; + final String _host; + final int _port; + final SecurityContext? _securityContext; + final WsProtobufClient Function(WebSocket) _createNewClient; + + bool get started => _started; + bool get isSecure => _securityContext != null; + + /// Plain WebSocket server (ws://). + WsProtobufServer(this._host, this._port, this._createNewClient) + : _securityContext = null; + + /// Secure WebSocket server (wss://). + /// + /// [securityContext] must have a certificate and private key loaded: + /// ```dart + /// final ctx = SecurityContext() + /// ..useCertificateChain('server.crt') + /// ..usePrivateKey('server.key'); + /// final server = MyServer.secure('0.0.0.0', 443, ctx, (ws) => MyClient(ws)); + /// ``` + WsProtobufServer.secure( + this._host, this._port, SecurityContext securityContext, this._createNewClient) + : _securityContext = securityContext; + + Future start() async { + final ctx = _securityContext; + if (ctx != null) { + _server = await HttpServer.bindSecure(_host, _port, ctx); + } else { + _server = await HttpServer.bind(_host, _port); + } + _started = true; + _server!.listen((HttpRequest request) async { + if (WebSocketTransformer.isUpgradeRequest(request)) { + final ws = await WebSocketTransformer.upgrade(request); + _onClientWebSocket(ws); + } + }); + return Future.value(); + } + + void _onClientWebSocket(WebSocket ws) { + var client = _createNewClient(ws); + client.addDisconnectListener(onDisconnectedClient); + client.addListener((_) => onClientHeartBeat(client)); + _clientList.add(client); + onClientConnected(client); + } + + // implement in child + void onClientConnected(WsProtobufClient client); + + void onClientHeartBeat(WsProtobufClient client) {} + + void onDisconnectedClient(WsProtobufClient client) { + _clientList.remove(client); + client.dispose(); + print('WS Client disconnected'); + } + + Future broadcast(T data) async { + for (var client in _clientList) { + await client.send(data); + } + return Future.value(); + } + + Future stop() async { + await _server?.close(force: true); + _started = false; + return Future.value(); + } +} diff --git a/dart/lib/toki_socket.dart b/dart/lib/toki_socket.dart index 54d9bbb..e811709 100644 --- a/dart/lib/toki_socket.dart +++ b/dart/lib/toki_socket.dart @@ -4,4 +4,6 @@ export 'src/communicator.dart'; export 'src/protobuf_client.dart'; export 'src/protobuf_server.dart'; export 'src/response_checker.dart'; +export 'src/ws_protobuf_client.dart'; +export 'src/ws_protobuf_server.dart'; export 'src/packets/message_common.pb.dart'; diff --git a/dart/test/socket_test.dart b/dart/test/socket_test.dart index fa7f6f7..9ecf047 100644 --- a/dart/test/socket_test.dart +++ b/dart/test/socket_test.dart @@ -6,6 +6,8 @@ import 'package:toki_socket/toki_socket.dart'; const _testPort = 19090; const _testPortSsl = 19091; +const _testPortWs = 19092; +const _testPortWss = 19093; const _host = 'localhost'; const _certPath = 'test/certs/server.crt'; const _keyPath = 'test/certs/server.key'; @@ -46,6 +48,40 @@ class _TestServerSsl extends ProtobufServer { } } +class _TestWsClient extends WsProtobufClient { + _TestWsClient(WebSocket ws) + : super(ws, 5, 3, { + TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer, + }); +} + +class _TestWsServer extends WsProtobufServer { + final receivedMessages = []; + final connectedClients = []; + + _TestWsServer() : super(_host, _testPortWs, (ws) => _TestWsClient(ws)); + + @override + void onClientConnected(WsProtobufClient client) { + connectedClients.add(client); + client.addListener((data) => receivedMessages.add(data)); + } +} + +class _TestWsServerSsl extends WsProtobufServer { + final receivedMessages = []; + final connectedClients = []; + + _TestWsServerSsl(SecurityContext ctx) + : super.secure(_host, _testPortWss, ctx, (ws) => _TestWsClient(ws)); + + @override + void onClientConnected(WsProtobufClient client) { + connectedClients.add(client); + client.addListener((data) => receivedMessages.add(data)); + } +} + // ── 테스트 ────────────────────────────────────────────────────── void main() { @@ -249,4 +285,159 @@ void main() { expect(completer.isCompleted, isTrue); }); }); + + // ── WebSocket ──────────────────────────────────────────────── + + group('WsProtobufServer (plain)', () { + late _TestWsServer server; + + setUp(() async { + server = _TestWsServer(); + await server.start(); + }); + + tearDown(() async { + await server.stop(); + }); + + test('WS 서버가 정상 시작된다', () { + expect(server.started, isTrue); + expect(server.isSecure, isFalse); + }); + }); + + group('WsProtobufClient (plain)', () { + late _TestWsServer server; + late _TestWsClient client; + + setUp(() async { + server = _TestWsServer(); + await server.start(); + final ws = await WsProtobufClient.connect(_host, _testPortWs); + client = _TestWsClient(ws); + await Future.delayed(const Duration(milliseconds: 100)); + }); + + tearDown(() async { + client.dispose(); + await Future.delayed(const Duration(milliseconds: 100)); + await server.stop(); + }); + + test('WS 클라이언트가 서버에 연결된다', () { + expect(server.connectedClients, isNotEmpty); + }); + + test('WS TestData 메시지를 서버가 수신한다', () async { + await client.send(TestData() + ..index = 42 + ..message = 'hello ws'); + await Future.delayed(const Duration(milliseconds: 200)); + + expect(server.receivedMessages, hasLength(1)); + expect(server.receivedMessages.first.index, equals(42)); + expect(server.receivedMessages.first.message, equals('hello ws')); + }); + + test('WS 서버에서 클라이언트로 메시지를 전송한다', () async { + final completer = Completer(); + client.addListener((data) => completer.complete(data)); + + await Future.delayed(const Duration(milliseconds: 100)); + await server.connectedClients.first.send(TestData() + ..index = 99 + ..message = 'from ws server'); + + final received = await completer.future.timeout(const Duration(seconds: 2)); + expect(received.index, equals(99)); + expect(received.message, equals('from ws server')); + }); + + test('WS 클라이언트 disconnect 시 서버 콜백이 호출된다', () async { + final completer = Completer(); + server.connectedClients.first.addDisconnectListener((_) { + if (!completer.isCompleted) completer.complete(); + }); + + client.dispose(); + await completer.future.timeout(const Duration(seconds: 2)); + expect(completer.isCompleted, isTrue); + }); + }); + + // ── WebSocket SSL ───────────────────────────────────────────── + + group('WsProtobufServer (SSL)', () { + late _TestWsServerSsl server; + + setUp(() async { + final ctx = SecurityContext() + ..useCertificateChain(_certPath) + ..usePrivateKey(_keyPath); + server = _TestWsServerSsl(ctx); + await server.start(); + }); + + tearDown(() async { + await server.stop(); + }); + + test('WSS 서버가 정상 시작된다', () { + expect(server.started, isTrue); + expect(server.isSecure, isTrue); + }); + }); + + group('WsProtobufClient (SSL)', () { + late _TestWsServerSsl server; + late _TestWsClient client; + + setUp(() async { + final serverCtx = SecurityContext() + ..useCertificateChain(_certPath) + ..usePrivateKey(_keyPath); + server = _TestWsServerSsl(serverCtx); + await server.start(); + + final clientCtx = SecurityContext()..setTrustedCertificates(_certPath); + final ws = await WsProtobufClient.connectSecure( + _host, _testPortWss, + context: clientCtx, + ); + client = _TestWsClient(ws); + await Future.delayed(const Duration(milliseconds: 100)); + }); + + tearDown(() async { + client.dispose(); + await Future.delayed(const Duration(milliseconds: 100)); + await server.stop(); + }); + + test('WSS 클라이언트가 서버에 연결된다', () { + expect(server.connectedClients, isNotEmpty); + }); + + test('WSS TestData 메시지를 서버가 수신한다', () async { + await client.send(TestData() + ..index = 7 + ..message = 'hello over wss'); + await Future.delayed(const Duration(milliseconds: 200)); + + expect(server.receivedMessages, hasLength(1)); + expect(server.receivedMessages.first.index, equals(7)); + expect(server.receivedMessages.first.message, equals('hello over wss')); + }); + + test('WSS disconnect 시 서버 콜백이 호출된다', () async { + final completer = Completer(); + server.connectedClients.first.addDisconnectListener((_) { + if (!completer.isCompleted) completer.complete(); + }); + + client.dispose(); + await completer.future.timeout(const Duration(seconds: 2)); + expect(completer.isCompleted, isTrue); + }); + }); }