toki/proto-socket
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Update protocol docs and fix communicator implementation

Browse source
This commit is contained in:
toki 2026-04-05 22:46:46 +09:00
parent f18f50ce5b
commit 661c28a9ac
10 changed files with 458 additions and 87 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
PROTOCOL.md
View file

@ -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<GetUser, UserData>
│ │ 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.
---

10
README.md
View file

@ -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.

129
dart/lib/src/communicator.dart
View file

@ -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<String, IDataHandler> _handlerDic = {};
Map<String, Future<void> Function(List<int>, int)> _requestHandlerDic = {};
Map<int, _PendingRequest> _pendingRequests = {};
late Map<String, GeneratedMessage Function(List<int>)> _instanceGenerator;
Future<void> _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<String, GeneratedMessage Function(List<int>)> instanceGenerator) {
void initialize(
Map<String, GeneratedMessage Function(List<int>)> instanceGenerator) {
_instanceGenerator = instanceGenerator;
}
T Function(List<int>) getGenerator<T extends GeneratedMessage>(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<int>);
}
/// Transport-specific framing and write. Implemented by each client subclass.
Future<void> transmitPacket(PacketBase base);
/// Serializes writes so stream transports do not interleave packets.
Future<void> queuePacket(PacketBase base) {
final write = _outboundWrite.then((_) => transmitPacket(base));
_outboundWrite = write.catchError((_) {});
return write;
}
Future send<T extends GeneratedMessage>(T data);
void onReceivedData(String typeName, List<int> 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, UserData>(GetUser()..id = 1);
/// ```
Future<Res>
sendRequest<Req extends GeneratedMessage, Res extends GeneratedMessage>(
Req data) async {
if (!isAlive) return Future.error(StateError('not connected'));
final requestNonce = ++nonce;
final completer = Completer<Res>();
final expectedResponseType = Res.toString();
_pendingRequests[requestNonce] = _PendingRequest(
expectedTypeName: expectedResponseType,
complete: (bytes) {
completer.complete(getGenerator<Res>(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<GetUser, UserData>((req) async {
/// return UserData()..name = db.getUser(req.id).name;
/// });
/// ```
void addRequestListener<Req extends GeneratedMessage,
Res extends GeneratedMessage>(Future<Res> 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<int> bytes, int requestNonce) async {
final req = getGenerator<Req>(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<int> 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<T extends GeneratedMessage>(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<T>(getGenerator(type));
}
@ -71,3 +182,15 @@ class DataHandler<T extends GeneratedMessage> implements IDataHandler {
_listeners.remove(handler);
}
}
class _PendingRequest {
final String expectedTypeName;
final void Function(List<int>) complete;
final void Function(Object, StackTrace) completeError;
_PendingRequest({
required this.expectedTypeName,
required this.complete,
required this.completeError,
});
}

14
dart/lib/src/packets/message_common.pb.dart
View file

@ -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 {

6
dart/lib/src/packets/message_common.pbjson.dart
View file

@ -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 = {

1
dart/lib/src/packets/message_common.proto
View file

@ -4,6 +4,7 @@ message PacketBase {
string typeName = 1;
int32 nonce = 2;
bytes data = 3;
int32 responseNonce = 4;
}
message HeartBeat {}

68
dart/lib/src/protobuf_client.dart
View file

@ -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<Socket> 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<Socket> 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<int> _arrivedData = [];
List<void Function(ProtobufClient)> _onDisconnectListenerList = [];
@ -54,7 +37,7 @@ abstract class ProtobufClient extends Communicator {
this._heartbeatWaitTime,
Map<String, GeneratedMessage Function(List<int>)> 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<void> 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 extends GeneratedMessage>(T data) async {
if (_isAlive) {
if (isAlive) {
try {
List<int> 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
}
}
}
}

56
dart/lib/src/ws_protobuf_client.dart
View file

@ -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<WebSocket> 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<WebSocket> 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<void Function(WsProtobufClient)> _onDisconnectListenerList = [];
WsProtobufClient(
@ -54,25 +36,26 @@ abstract class WsProtobufClient extends Communicator {
this._heartbeatWaitTime,
Map<String, GeneratedMessage Function(List<int>)> 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<int> ? 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<void> transmitPacket(PacketBase base) async {
_ws.add(base.writeToBuffer());
}
@override
Future send<T extends GeneratedMessage>(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();

91
dart/test/communicator_test.dart Normal file
View file

@ -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 = <PacketBase>[];
_FakeCommunicator() {
isAlive = true;
initialize({
TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
HeartBeat.getDefault().info_.qualifiedMessageName: HeartBeat.fromBuffer,
});
}
@override
Future<void> transmitPacket(PacketBase base) async {
sentPackets.add(base);
}
@override
Future send<T extends GeneratedMessage>(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, TestData>(
TestData()
..index = 1
..message = 'hello',
);
await Future<void>.delayed(Duration.zero);
final requestNonce = communicator.sentPackets.single.nonce;
communicator.onReceivedData(
HeartBeat.getDefault().info_.qualifiedMessageName,
HeartBeat().writeToBuffer(),
responseNonce: requestNonce,
);
await expectLater(
future,
throwsA(
isA<StateError>().having((error) => error.toString(), 'message',
contains('Response type mismatch')),
),
);
});
test(
'cannot register addRequestListener for a type already using addListener',
() {
final communicator = _FakeCommunicator();
communicator.addListener<TestData>((_) {});
expect(
() => communicator.addRequestListener<TestData, TestData>(
(req) async => TestData()..index = req.index,
),
throwsA(isA<StateError>()),
);
});
test(
'cannot register addListener for a type already using addRequestListener',
() {
final communicator = _FakeCommunicator();
communicator.addRequestListener<TestData, TestData>(
(req) async => TestData()..index = req.index,
);
expect(
() => communicator.addListener<TestData>((_) {}),
throwsA(isA<StateError>()),
);
});
});
}

119
dart/test/socket_test.dart
View file

@ -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<TestData, TestData>((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<TestData, TestData>((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, TestData>(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, TestData>(TestData()..index = 1..message = 'a'),
client.sendRequest<TestData, TestData>(TestData()..index = 2..message = 'b'),
client.sendRequest<TestData, TestData>(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, TestData>(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, TestData>(TestData()..index = 1..message = 'x'),
client.sendRequest<TestData, TestData>(TestData()..index = 2..message = 'y'),
client.sendRequest<TestData, TestData>(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));
});
});
}