From 661c28a9acf66887755e8daa3ad887c8626af73a Mon Sep 17 00:00:00 2001 From: toki Date: Sun, 5 Apr 2026 22:46:46 +0900 Subject: [PATCH] Update protocol docs and fix communicator implementation --- PROTOCOL.md | 51 ++++++- README.md | 10 ++ dart/lib/src/communicator.dart | 129 +++++++++++++++++- dart/lib/src/packets/message_common.pb.dart | 14 ++ .../src/packets/message_common.pbjson.dart | 6 +- dart/lib/src/packets/message_common.proto | 1 + dart/lib/src/protobuf_client.dart | 68 ++++----- dart/lib/src/ws_protobuf_client.dart | 56 +++----- dart/test/communicator_test.dart | 91 ++++++++++++ dart/test/socket_test.dart | 119 ++++++++++++++++ 10 files changed, 458 insertions(+), 87 deletions(-) create mode 100644 dart/test/communicator_test.dart diff --git a/PROTOCOL.md b/PROTOCOL.md index cd0586c..57b7d0a 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -5,6 +5,16 @@ Designed for bidirectional, heterogeneous communication across languages and pla --- +## Scope + +This protocol intentionally defines only the transport-level contract shared across implementations. + +- In scope: framing, protobuf payload transport, `typeName` routing, `nonce`, `responseNonce`, and heartbeat +- Out of scope: authentication, session lifecycle, agent semantics, chat semantics, game rules, room logic, and other application-specific behavior +- Higher-level concerns should be implemented as messages and conventions on top of this protocol, not inside the protocol core + +--- + ## Transport별 Wire Format ### TCP / SSL+TCP @@ -37,17 +47,19 @@ Designed for bidirectional, heterogeneous communication across languages and pla ### PacketBase ```protobuf message PacketBase { - string typeName = 1; // fully-qualified protobuf message name - int32 nonce = 2; // monotonically increasing per-connection counter - bytes data = 3; // serialized inner message bytes + string typeName = 1; // fully-qualified protobuf message name + int32 nonce = 2; // monotonically increasing per-connection counter + bytes data = 3; // serialized inner message bytes + int32 responseNonce = 4; // > 0: this packet is a response to request with that nonce } ``` | Field | Description | |-------|-------------| | `typeName` | `GeneratedMessage.info_.qualifiedMessageName` — language-agnostic routing key | -| `nonce` | Starts at 1, increments by 1 per `send()` call per connection | +| `nonce` | Starts at 1, increments by 1 per send call per connection | | `data` | Protobuf-serialized bytes of the inner message | +| `responseNonce` | `0` for regular messages. Set to the request's `nonce` when sending a response | --- @@ -105,8 +117,35 @@ On the receive side, use the same value as the registration key. ## nonce - Starts at `1` per connection -- Increments by `1` on every `send()` call -- Currently used as a message ID; request-response correlation can be built on top +- Increments by `1` on every send call (including requests and responses) +- Used for request-response correlation via `responseNonce` + +--- + +## Request-Response Pattern + +Fire-and-forget (`send`) and request-response (`sendRequest`) coexist on the same connection. + +``` +Requester Responder + │ │ + │── PacketBase { nonce=5, responseNonce=0, │ + │ typeName="GetUser", │ + │ data=... } ─────────────▶│ + │ │── addRequestListener + │ │ handler called → returns UserData + │◀─ PacketBase { nonce=12, responseNonce=5,│ + │ typeName="UserData", │ + │ data=... } ──────────────│ + │ sendRequest Future completes │ +``` + +**Rules:** +- `responseNonce == 0`: regular message or outgoing request → routed to `addListener` or `addRequestListener` +- `responseNonce > 0`: response → matched to the pending `sendRequest` by `responseNonce` +- A response must also have the expected `typeName` for the waiting `sendRequest`; mismatches are protocol errors +- Both sides can be requester and responder simultaneously on the same connection +- For a given `typeName` on one connection, register it with either `addListener` or `addRequestListener`, not both. Mixed registration is ambiguous and rejected by the Dart implementation. --- diff --git a/README.md b/README.md index f1733a8..c42e73e 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,16 @@ Built on Protocol Buffers with a fixed 4-byte length-prefixed framing, type-base --- +## Design Principles + +- Keep the core transport layer thin and stable +- Provide only the minimum common foundation for cross-language communication +- Standardize framing, serialization, routing, request-response correlation, and heartbeat +- Do not embed application semantics into the protocol core +- Domain-specific concerns such as auth, session, agent workflow, chat features, and game logic belong in upper-layer implementations + +--- + ## Protocol See [PROTOCOL.md](PROTOCOL.md) for the full wire format specification. diff --git a/dart/lib/src/communicator.dart b/dart/lib/src/communicator.dart index 4d2c3f8..8453ae7 100644 --- a/dart/lib/src/communicator.dart +++ b/dart/lib/src/communicator.dart @@ -1,27 +1,134 @@ // ignore_for_file: prefer_final_fields +import 'dart:async'; import 'package:protobuf/protobuf.dart'; +import 'packets/message_common.pb.dart'; abstract class Communicator { Map _handlerDic = {}; + Map Function(List, int)> _requestHandlerDic = {}; + Map _pendingRequests = {}; late Map)> _instanceGenerator; + Future _outboundWrite = Future.value(); + + /// Monotonically increasing nonce. Shared by send / sendRequest / response. + int nonce = 0; + + /// Whether the connection is alive. Set by each transport implementation. + bool isAlive = false; Communicator(); - void initialize(Map)> instanceGenerator) { + void initialize( + Map)> instanceGenerator) { _instanceGenerator = instanceGenerator; } T Function(List) getGenerator(String type) { if (!_instanceGenerator.containsKey(type)) { - throw Exception('Must set protobuf packet creator before use it. Type: ${(T).toString()}'); + throw Exception( + 'Must set protobuf packet creator before use it. Type: ${(T).toString()}'); } return _instanceGenerator[type] as T Function(List); } + /// Transport-specific framing and write. Implemented by each client subclass. + Future transmitPacket(PacketBase base); + + /// Serializes writes so stream transports do not interleave packets. + Future queuePacket(PacketBase base) { + final write = _outboundWrite.then((_) => transmitPacket(base)); + _outboundWrite = write.catchError((_) {}); + return write; + } + Future send(T data); - void onReceivedData(String typeName, List data) { + /// Sends [data] as a request and waits for a typed response. + /// + /// The remote side must have registered an [addRequestListener] for [Req]. + /// [Res] must be registered in the parser map. + /// + /// ```dart + /// final res = await client.sendRequest(GetUser()..id = 1); + /// ``` + Future + sendRequest( + Req data) async { + if (!isAlive) return Future.error(StateError('not connected')); + final requestNonce = ++nonce; + final completer = Completer(); + final expectedResponseType = Res.toString(); + _pendingRequests[requestNonce] = _PendingRequest( + expectedTypeName: expectedResponseType, + complete: (bytes) { + completer.complete(getGenerator(expectedResponseType)(bytes)); + }, + completeError: (error, stackTrace) { + completer.completeError(error, stackTrace); + }, + ); + await queuePacket(PacketBase() + ..typeName = data.info_.qualifiedMessageName + ..nonce = requestNonce + ..data = data.writeToBuffer()) + .catchError((error, stackTrace) { + _pendingRequests.remove(requestNonce); + completer.completeError(error, stackTrace); + }); + return completer.future; + } + + /// Registers a request handler for [Req] that returns [Res]. + /// + /// When a [Req] packet arrives the handler is called and the returned [Res] + /// is automatically sent back to the caller. + /// + /// ```dart + /// server.addRequestListener((req) async { + /// return UserData()..name = db.getUser(req.id).name; + /// }); + /// ``` + void addRequestListener(Future Function(Req) handler) { + final reqType = Req.toString(); + if (_handlerDic.containsKey(reqType)) { + throw StateError( + 'Type $reqType is already registered with addListener and cannot also use addRequestListener.'); + } + _requestHandlerDic[reqType] = (List bytes, int requestNonce) async { + final req = getGenerator(reqType)(bytes); + final res = await handler(req); + if (isAlive) { + await queuePacket(PacketBase() + ..typeName = res.info_.qualifiedMessageName + ..nonce = ++nonce + ..responseNonce = requestNonce + ..data = res.writeToBuffer()); + } + }; + } + + void onReceivedData(String typeName, List data, + {int incomingNonce = 0, int responseNonce = 0}) { + if (responseNonce > 0) { + final pending = _pendingRequests.remove(responseNonce); + if (pending == null) return; + if (typeName != pending.expectedTypeName) { + pending.completeError( + StateError( + 'Response type mismatch for nonce $responseNonce: expected ${pending.expectedTypeName}, got $typeName'), + StackTrace.current, + ); + return; + } + pending.complete(data); + return; + } + if (_requestHandlerDic.containsKey(typeName)) { + _requestHandlerDic[typeName]!(data, incomingNonce); + return; + } if (_handlerDic.containsKey(typeName)) { _handlerDic[typeName]?.onMessage(data); } @@ -29,6 +136,10 @@ abstract class Communicator { void addListener(void Function(T) listener) { var type = T.toString(); + if (_requestHandlerDic.containsKey(type)) { + throw StateError( + 'Type $type is already registered with addRequestListener and cannot also use addListener.'); + } if (!_handlerDic.containsKey(type)) { _handlerDic[type] = DataHandler(getGenerator(type)); } @@ -71,3 +182,15 @@ class DataHandler implements IDataHandler { _listeners.remove(handler); } } + +class _PendingRequest { + final String expectedTypeName; + final void Function(List) complete; + final void Function(Object, StackTrace) completeError; + + _PendingRequest({ + required this.expectedTypeName, + required this.complete, + required this.completeError, + }); +} diff --git a/dart/lib/src/packets/message_common.pb.dart b/dart/lib/src/packets/message_common.pb.dart index 059eb0c..a28c38f 100644 --- a/dart/lib/src/packets/message_common.pb.dart +++ b/dart/lib/src/packets/message_common.pb.dart @@ -18,6 +18,7 @@ class PacketBase extends $pb.GeneratedMessage { $core.String? typeName, $core.int? nonce, $core.List<$core.int>? data, + $core.int? responseNonce, }) { final $result = create(); if (typeName != null) { @@ -29,6 +30,9 @@ class PacketBase extends $pb.GeneratedMessage { if (data != null) { $result.data = data; } + if (responseNonce != null) { + $result.responseNonce = responseNonce; + } return $result; } PacketBase._() : super(); @@ -39,6 +43,7 @@ class PacketBase extends $pb.GeneratedMessage { ..aOS(1, _omitFieldNames ? '' : 'typeName', protoName: 'typeName') ..a<$core.int>(2, _omitFieldNames ? '' : 'nonce', $pb.PbFieldType.O3) ..a<$core.List<$core.int>>(3, _omitFieldNames ? '' : 'data', $pb.PbFieldType.OY) + ..a<$core.int>(4, _omitFieldNames ? '' : 'responseNonce', $pb.PbFieldType.O3, protoName: 'responseNonce') ..hasRequiredFields = false ; @@ -89,6 +94,15 @@ class PacketBase extends $pb.GeneratedMessage { $core.bool hasData() => $_has(2); @$pb.TagNumber(3) void clearData() => clearField(3); + + @$pb.TagNumber(4) + $core.int get responseNonce => $_getIZ(3); + @$pb.TagNumber(4) + set responseNonce($core.int v) { $_setSignedInt32(3, v); } + @$pb.TagNumber(4) + $core.bool hasResponseNonce() => $_has(3); + @$pb.TagNumber(4) + void clearResponseNonce() => clearField(4); } class HeartBeat extends $pb.GeneratedMessage { diff --git a/dart/lib/src/packets/message_common.pbjson.dart b/dart/lib/src/packets/message_common.pbjson.dart index 6687246..61fcc84 100644 --- a/dart/lib/src/packets/message_common.pbjson.dart +++ b/dart/lib/src/packets/message_common.pbjson.dart @@ -20,13 +20,15 @@ const PacketBase$json = { {'1': 'typeName', '3': 1, '4': 1, '5': 9, '10': 'typeName'}, {'1': 'nonce', '3': 2, '4': 1, '5': 5, '10': 'nonce'}, {'1': 'data', '3': 3, '4': 1, '5': 12, '10': 'data'}, + {'1': 'responseNonce', '3': 4, '4': 1, '5': 5, '10': 'responseNonce'}, ], }; /// Descriptor for `PacketBase`. Decode as a `google.protobuf.DescriptorProto`. final $typed_data.Uint8List packetBaseDescriptor = $convert.base64Decode( - 'CgpQYWNrZXRCYXNlEhoKCHR5cGVOYW1lGAEgASgJUgh0eXBlTmFtZRIUCgVub25jZRgCIAEoBV' - 'IFbm9uY2USEgoEZGF0YRgDIAEoDFIEZGF0YQ=='); + 'CgpQYWNrZXRCYXNlEhoKCHR5cGVOYW1lGAEgASgJUgh0eXBlTmFtZRIUCgVub25jZRgCIAEoBVIF' + 'bm9uY2USEgoEZGF0YRgDIAEoDFIEZGF0YRIkCg1yZXNwb25zZU5vbmNlGAQgASgFUg1yZXNwb25z' + 'ZU5vbmNl'); @$core.Deprecated('Use heartBeatDescriptor instead') const HeartBeat$json = { diff --git a/dart/lib/src/packets/message_common.proto b/dart/lib/src/packets/message_common.proto index 4efcae2..02ffe92 100644 --- a/dart/lib/src/packets/message_common.proto +++ b/dart/lib/src/packets/message_common.proto @@ -4,6 +4,7 @@ message PacketBase { string typeName = 1; int32 nonce = 2; bytes data = 3; + int32 responseNonce = 4; } message HeartBeat {} diff --git a/dart/lib/src/protobuf_client.dart b/dart/lib/src/protobuf_client.dart index 9e45395..16b8c83 100644 --- a/dart/lib/src/protobuf_client.dart +++ b/dart/lib/src/protobuf_client.dart @@ -16,35 +16,18 @@ abstract class ProtobufClient extends Communicator { final Socket _socket; /// Plain TCP connection. - /// - /// ```dart - /// final socket = await ProtobufClient.connect('localhost', 9090); - /// final client = MyClient(socket); - /// ``` static Future connect(String host, int port) => Socket.connect(host, port); /// SSL/TLS connection. - /// - /// ```dart - /// final socket = await ProtobufClient.connectSecure('example.com', 9090); - /// final client = MyClient(socket); - /// ``` - /// - /// For self-signed or custom certificates, provide a [SecurityContext]: - /// ```dart - /// final ctx = SecurityContext()..setTrustedCertificates('ca.crt'); - /// final socket = await ProtobufClient.connectSecure('host', 9090, context: ctx); - /// ``` static Future connectSecure(String host, int port, {SecurityContext? context}) => SecureSocket.connect(host, port, context: context ?? SecurityContext.defaultContext); + late ResponseChecker? _heartbeatChecker = null; int? _length = null; - bool _isAlive = false; bool _waitingHeartbeatResponse = false; - int _nonce = 0; late List _arrivedData = []; List _onDisconnectListenerList = []; @@ -54,7 +37,7 @@ abstract class ProtobufClient extends Communicator { this._heartbeatWaitTime, Map)> parserMap) { print('Connected New Client'); - _isAlive = true; + isAlive = true; parserMap.addAll({ (HeartBeat).toString(): HeartBeat.fromBuffer, }); @@ -65,16 +48,16 @@ abstract class ProtobufClient extends Communicator { } void sendHeartBeat() { - if (_isAlive) { + if (isAlive) { _heartbeatChecker?.responded(); _heartbeatChecker = ResponseChecker.second(this, _heartbeatIntervalTime, (client) { - if (_isAlive) { + if (isAlive) { _waitingHeartbeatResponse = true; send(HeartBeat()); _heartbeatChecker = ResponseChecker.second(this, _heartbeatWaitTime, (client) { - if (_isAlive) { + if (isAlive) { onDisconnected(null); dispose(); } @@ -105,7 +88,7 @@ abstract class ProtobufClient extends Communicator { } void onDisconnected(dynamic data) { - if (_isAlive) { + if (isAlive) { for (var item in _onDisconnectListenerList) { item.call(this); } @@ -147,30 +130,31 @@ abstract class ProtobufClient extends Communicator { _arrivedData = _arrivedData.sublist(_headerSize + _length!); _length = null; final common = PacketBase.fromBuffer(packetBytes); - onReceivedData(common.typeName, common.data); + onReceivedData(common.typeName, common.data, + incomingNonce: common.nonce, responseNonce: common.responseNonce); sendHeartBeat(); _parsing(); } } } + @override + Future transmitPacket(PacketBase base) async { + final baseBytes = base.writeToBuffer(); + final header = Uint8List(_headerSize) + ..buffer.asByteData().setInt32(0, baseBytes.length); + _socket.add([...header, ...baseBytes]); + await _socket.flush(); + } + @override Future send(T data) async { - if (_isAlive) { + if (isAlive) { try { - List packet = []; - var base = PacketBase(); - base.typeName = data.info_.qualifiedMessageName; - base.nonce = ++_nonce; - base.data = data.writeToBuffer(); - var baseBytes = base.writeToBuffer(); - int length = baseBytes.length; - var header = Uint8List(_headerSize) - ..buffer.asByteData().setInt32(0, length); - packet.addAll(header); - packet.addAll(baseBytes); - _socket.add(packet); - await _socket.flush(); + await queuePacket(PacketBase() + ..typeName = data.info_.qualifiedMessageName + ..nonce = ++nonce + ..data = data.writeToBuffer()); } catch (e) { onDisconnected(null); } @@ -179,13 +163,15 @@ abstract class ProtobufClient extends Communicator { } void dispose() async { - if (_isAlive) { - _isAlive = false; + if (isAlive) { + isAlive = false; _heartbeatChecker?.responded(); try { await _socket.close(); _socket.destroy(); - } catch (e) {} + } catch (_) { + // already closed + } } } } diff --git a/dart/lib/src/ws_protobuf_client.dart b/dart/lib/src/ws_protobuf_client.dart index 85a9f1f..149a715 100644 --- a/dart/lib/src/ws_protobuf_client.dart +++ b/dart/lib/src/ws_protobuf_client.dart @@ -15,27 +15,11 @@ abstract class WsProtobufClient extends Communicator { 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', @@ -43,9 +27,7 @@ abstract class WsProtobufClient extends Communicator { HttpClient(context: context ?? SecurityContext.defaultContext)); late ResponseChecker? _heartbeatChecker = null; - bool _isAlive = false; bool _waitingHeartbeatResponse = false; - int _nonce = 0; List _onDisconnectListenerList = []; WsProtobufClient( @@ -54,25 +36,26 @@ abstract class WsProtobufClient extends Communicator { this._heartbeatWaitTime, Map)> parserMap) { print('Connected New WS Client'); - _isAlive = true; + isAlive = true; parserMap.addAll({(HeartBeat).toString(): HeartBeat.fromBuffer}); super.initialize(parserMap); - _ws.listen(_onMessage, onError: onError, onDone: () => onDisconnected(null)); + _ws.listen(_onMessage, + onError: onError, onDone: () => onDisconnected(null)); addListener(onHeartBeat); sendHeartBeat(); } void sendHeartBeat() { - if (_isAlive) { + if (isAlive) { _heartbeatChecker?.responded(); _heartbeatChecker = ResponseChecker.second(this, _heartbeatIntervalTime, (client) { - if (_isAlive) { + if (isAlive) { _waitingHeartbeatResponse = true; send(HeartBeat()); _heartbeatChecker = ResponseChecker.second(this, _heartbeatWaitTime, (client) { - if (_isAlive) { + if (isAlive) { onDisconnected(null); dispose(); } @@ -103,7 +86,7 @@ abstract class WsProtobufClient extends Communicator { } void onDisconnected(dynamic data) { - if (_isAlive) { + if (isAlive) { for (var item in _onDisconnectListenerList) { item.call(this); } @@ -118,21 +101,24 @@ abstract class WsProtobufClient extends Communicator { void _onMessage(dynamic data) { final bytes = data is List ? data : (data as Uint8List).toList(); final common = PacketBase.fromBuffer(bytes); - onReceivedData(common.typeName, common.data); + onReceivedData(common.typeName, common.data, + incomingNonce: common.nonce, responseNonce: common.responseNonce); sendHeartBeat(); } - /// Sends [data] as a binary WebSocket frame — no length header. - /// The WebSocket protocol guarantees message boundaries. + @override + Future transmitPacket(PacketBase base) async { + _ws.add(base.writeToBuffer()); + } + @override Future send(T data) async { - if (_isAlive) { + if (isAlive) { try { - var base = PacketBase(); - base.typeName = data.info_.qualifiedMessageName; - base.nonce = ++_nonce; - base.data = data.writeToBuffer(); - _ws.add(base.writeToBuffer()); + await queuePacket(PacketBase() + ..typeName = data.info_.qualifiedMessageName + ..nonce = ++nonce + ..data = data.writeToBuffer()); } catch (e) { onDisconnected(null); } @@ -141,8 +127,8 @@ abstract class WsProtobufClient extends Communicator { } void dispose() async { - if (_isAlive) { - _isAlive = false; + if (isAlive) { + isAlive = false; _heartbeatChecker?.responded(); try { await _ws.close(); diff --git a/dart/test/communicator_test.dart b/dart/test/communicator_test.dart new file mode 100644 index 0000000..8346145 --- /dev/null +++ b/dart/test/communicator_test.dart @@ -0,0 +1,91 @@ +import 'package:protobuf/protobuf.dart'; +import 'package:test/test.dart'; +import 'package:toki_socket/toki_socket.dart'; + +class _FakeCommunicator extends Communicator { + final sentPackets = []; + + _FakeCommunicator() { + isAlive = true; + initialize({ + TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer, + HeartBeat.getDefault().info_.qualifiedMessageName: HeartBeat.fromBuffer, + }); + } + + @override + Future transmitPacket(PacketBase base) async { + sentPackets.add(base); + } + + @override + Future send(T data) async { + if (isAlive) { + await queuePacket(PacketBase() + ..typeName = data.info_.qualifiedMessageName + ..nonce = ++nonce + ..data = data.writeToBuffer()); + } + return Future.value(); + } +} + +void main() { + group('Communicator protocol guards', () { + test('response typeName mismatch completes sendRequest with error', + () async { + final communicator = _FakeCommunicator(); + final future = communicator.sendRequest( + TestData() + ..index = 1 + ..message = 'hello', + ); + + await Future.delayed(Duration.zero); + final requestNonce = communicator.sentPackets.single.nonce; + communicator.onReceivedData( + HeartBeat.getDefault().info_.qualifiedMessageName, + HeartBeat().writeToBuffer(), + responseNonce: requestNonce, + ); + + await expectLater( + future, + throwsA( + isA().having((error) => error.toString(), 'message', + contains('Response type mismatch')), + ), + ); + }); + + test( + 'cannot register addRequestListener for a type already using addListener', + () { + final communicator = _FakeCommunicator(); + + communicator.addListener((_) {}); + + expect( + () => communicator.addRequestListener( + (req) async => TestData()..index = req.index, + ), + throwsA(isA()), + ); + }); + + test( + 'cannot register addListener for a type already using addRequestListener', + () { + final communicator = _FakeCommunicator(); + + communicator.addRequestListener( + (req) async => TestData()..index = req.index, + ); + + expect( + () => communicator.addListener((_) {}), + throwsA(isA()), + ); + }); + }); +} diff --git a/dart/test/socket_test.dart b/dart/test/socket_test.dart index 9ecf047..9c83e35 100644 --- a/dart/test/socket_test.dart +++ b/dart/test/socket_test.dart @@ -82,6 +82,32 @@ class _TestWsServerSsl extends WsProtobufServer { } } +class _TestReqServer extends ProtobufServer { + _TestReqServer() : super(_host, _testPort, (socket) => _TestClient(socket)); + + @override + void onClientConnected(ProtobufClient client) { + client.addRequestListener((req) async { + return TestData() + ..index = req.index * 2 + ..message = 'echo: ${req.message}'; + }); + } +} + +class _TestWsReqServer extends WsProtobufServer { + _TestWsReqServer() : super(_host, _testPortWs, (ws) => _TestWsClient(ws)); + + @override + void onClientConnected(WsProtobufClient client) { + client.addRequestListener((req) async { + return TestData() + ..index = req.index * 2 + ..message = 'echo: ${req.message}'; + }); + } +} + // ── 테스트 ────────────────────────────────────────────────────── void main() { @@ -440,4 +466,97 @@ void main() { expect(completer.isCompleted, isTrue); }); }); + + // ── Request-Response (TCP) ──────────────────────────────────── + + group('Request-Response (TCP plain)', () { + late _TestReqServer server; + late _TestClient client; + + setUp(() async { + server = _TestReqServer(); + await server.start(); + final socket = await ProtobufClient.connect(_host, _testPort); + client = _TestClient(socket); + await Future.delayed(const Duration(milliseconds: 100)); + }); + + tearDown(() async { + client.dispose(); + await Future.delayed(const Duration(milliseconds: 100)); + await server.stop(); + }); + + test('sendRequest로 서버 응답을 받는다', () async { + final response = await client + .sendRequest(TestData() + ..index = 21 + ..message = 'hello') + .timeout(const Duration(seconds: 2)); + + expect(response.index, equals(42)); + expect(response.message, equals('echo: hello')); + }); + + test('여러 sendRequest가 각각 올바른 응답을 받는다', () async { + final futures = [ + client.sendRequest(TestData()..index = 1..message = 'a'), + client.sendRequest(TestData()..index = 2..message = 'b'), + client.sendRequest(TestData()..index = 3..message = 'c'), + ]; + final results = await Future.wait(futures).timeout(const Duration(seconds: 3)); + + expect(results[0].index, equals(2)); + expect(results[1].index, equals(4)); + expect(results[2].index, equals(6)); + expect(results[0].message, equals('echo: a')); + expect(results[1].message, equals('echo: b')); + expect(results[2].message, equals('echo: c')); + }); + }); + + // ── Request-Response (WS) ───────────────────────────────────── + + group('Request-Response (WS plain)', () { + late _TestWsReqServer server; + late _TestWsClient client; + + setUp(() async { + server = _TestWsReqServer(); + 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 sendRequest로 서버 응답을 받는다', () async { + final response = await client + .sendRequest(TestData() + ..index = 21 + ..message = 'hello ws') + .timeout(const Duration(seconds: 2)); + + expect(response.index, equals(42)); + expect(response.message, equals('echo: hello ws')); + }); + + test('WS 여러 sendRequest가 각각 올바른 응답을 받는다', () async { + final futures = [ + client.sendRequest(TestData()..index = 1..message = 'x'), + client.sendRequest(TestData()..index = 2..message = 'y'), + client.sendRequest(TestData()..index = 3..message = 'z'), + ]; + final results = await Future.wait(futures).timeout(const Duration(seconds: 3)); + + expect(results[0].index, equals(2)); + expect(results[1].index, equals(4)); + expect(results[2].index, equals(6)); + }); + }); }