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

fix dart qualified protobuf type names

Browse source
This commit is contained in:
toki 2026-05-24 20:06:51 +09:00
parent b95e757655
commit e0e70d846d
2 changed files with 146 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -10,7 +10,8 @@ abstract class Communicator {
static const int maxNonce = 2147483647;
Map<String, IDataHandler> _handlerDic = {};
Map<String, Future<void> Function(List<int>, int)> _requestHandlerDic = {};
Map<String, Future<void> Function(String, List<int>, int)>
_requestHandlerDic = {};
Map<int, _PendingRequest> _pendingRequests = {};
late Map<String, GeneratedMessage Function(List<int>)> _instanceGenerator;
Future<void> _outboundWrite = Future.value();
@ -49,11 +50,42 @@ abstract class Communicator {
}
T Function(List<int>) getGenerator<T extends GeneratedMessage>(String type) {
if (!_instanceGenerator.containsKey(type)) {
final generator =
_instanceGenerator[type] ?? _instanceGenerator[_shortTypeName(type)];
if (generator == null) {
throw Exception(
'Must set protobuf packet creator before use it. Type: ${(T).toString()}');
}
return _instanceGenerator[type] as T Function(List<int>);
return generator as T Function(List<int>);
}
String _shortTypeName(String typeName) {
final lastDot = typeName.lastIndexOf('.');
if (lastDot < 0 || lastDot == typeName.length - 1) {
return typeName;
}
return typeName.substring(lastDot + 1);
}
V? _lookupByWireType<V>(Map<String, V> map, String typeName) {
return map[typeName] ?? map[_shortTypeName(typeName)];
}
GeneratedMessage _decodeMessage(
String typeName,
List<int> data, {
String? fallbackTypeName,
}) {
final generator = _instanceGenerator[typeName] ??
(fallbackTypeName == null
? null
: _instanceGenerator[fallbackTypeName]) ??
_instanceGenerator[_shortTypeName(typeName)];
if (generator == null) {
throw Exception(
'Must set protobuf packet creator before use it. Type: $typeName');
}
return generator(data);
}
/// Serializes writes so stream transports do not interleave packets.
@ -100,8 +132,18 @@ abstract class Communicator {
final expectedResponseType = Res.toString();
_pendingRequests[requestNonce] = _PendingRequest(
expectedTypeName: expectedResponseType,
complete: (bytes) {
completer.complete(getGenerator<Res>(expectedResponseType)(bytes));
complete: (typeName, bytes) {
final message = _decodeMessage(typeName, bytes,
fallbackTypeName: expectedResponseType);
if (message is! Res) {
completer.completeError(
StateError(
'Response type mismatch for nonce $requestNonce: expected $expectedResponseType, got $typeName'),
StackTrace.current,
);
return;
}
completer.complete(message);
},
completeError: (error, stackTrace) {
completer.completeError(error, stackTrace);
@ -139,8 +181,15 @@ abstract class Communicator {
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);
_requestHandlerDic[reqType] =
(String typeName, List<int> bytes, int requestNonce) async {
final message =
_decodeMessage(typeName, bytes, fallbackTypeName: reqType);
if (message is! Req) {
throw StateError(
'Request type mismatch for nonce $requestNonce: expected $reqType, got $typeName');
}
final req = message;
final res = await handler(req);
if (isAlive) {
await queuePacket(PacketBase()
@ -157,23 +206,17 @@ abstract class Communicator {
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);
pending.complete(typeName, data);
return;
}
if (_requestHandlerDic.containsKey(typeName)) {
_requestHandlerDic[typeName]!(data, incomingNonce);
final requestHandler = _lookupByWireType(_requestHandlerDic, typeName);
if (requestHandler != null) {
requestHandler(typeName, data, incomingNonce);
return;
}
if (_handlerDic.containsKey(typeName)) {
_handlerDic[typeName]?.onMessage(data);
final handler = _lookupByWireType(_handlerDic, typeName);
if (handler != null) {
handler.onMessage(typeName, data);
}
}
@ -184,7 +227,14 @@ abstract class Communicator {
'Type $type is already registered with addRequestListener and cannot also use addListener.');
}
if (!_handlerDic.containsKey(type)) {
_handlerDic[type] = DataHandler<T>(getGenerator(type));
_handlerDic[type] = DataHandler<T>((typeName, data) {
final message = _decodeMessage(typeName, data, fallbackTypeName: type);
if (message is! T) {
throw StateError(
'Message type mismatch: expected $type, got $typeName');
}
return message;
});
}
var handler = _handlerDic[type] as DataHandler<T>;
handler.addListener(listener);
@ -200,19 +250,19 @@ abstract class Communicator {
}
abstract class IDataHandler {
void onMessage(List<int> data);
void onMessage(String typeName, List<int> data);
}
class DataHandler<T extends GeneratedMessage> implements IDataHandler {
T Function(List<int>) _generator;
T Function(String, List<int>) _generator;
List<void Function(T)> _listeners = [];
DataHandler(this._generator);
@override
void onMessage(List<int> data) {
void onMessage(String typeName, List<int> data) {
for (var listener in _listeners) {
listener.call(_generator(data));
listener.call(_generator(typeName, data));
}
}
@ -228,7 +278,7 @@ class DataHandler<T extends GeneratedMessage> implements IDataHandler {
class _PendingRequest {
final String expectedTypeName;
final void Function(List<int>) complete;
final void Function(String, List<int>) complete;
final void Function(Object, StackTrace) completeError;
_PendingRequest({

72
dart/test/communicator_test.dart
View file

@ -7,10 +7,13 @@ import 'package:proto_socket/proto_socket.dart';
class _FakeCommunicator extends Communicator {
final _FakeTransport transport = _FakeTransport();
_FakeCommunicator() {
_FakeCommunicator({bool packageQualifiedParser = false}) {
isAlive = true;
final testDataType = packageQualifiedParser
? 'example.TestData'
: TestData.getDefault().info_.qualifiedMessageName;
initialize({
TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
testDataType: TestData.fromBuffer,
HeartBeat.getDefault().info_.qualifiedMessageName: HeartBeat.fromBuffer,
}, transport: transport);
}
@ -177,6 +180,71 @@ void main() {
expect((await wrappedFuture).message, 'wrapped response');
});
test('sendRequest accepts package-qualified response typeName', () async {
final communicator = _FakeCommunicator(packageQualifiedParser: true);
final future = communicator.sendRequest<TestData, TestData>(
TestData()
..index = 1
..message = 'request',
);
await Future<void>.delayed(Duration.zero);
final requestNonce = communicator.transport.sentPackets.single.nonce;
communicator.onReceivedData(
'example.TestData',
(TestData()
..index = 2
..message = 'qualified response')
.writeToBuffer(),
responseNonce: requestNonce,
);
expect((await future).message, 'qualified response');
});
test('addListener accepts package-qualified message typeName', () {
final communicator = _FakeCommunicator(packageQualifiedParser: true);
final messages = <TestData>[];
communicator.addListener<TestData>(messages.add);
communicator.onReceivedData(
'example.TestData',
(TestData()
..index = 3
..message = 'qualified event')
.writeToBuffer(),
);
expect(messages, hasLength(1));
expect(messages.single.message, 'qualified event');
});
test('addRequestListener accepts package-qualified request typeName',
() async {
final communicator = _FakeCommunicator(packageQualifiedParser: true);
communicator.addRequestListener<TestData, TestData>((req) async {
return TestData()
..index = req.index + 1
..message = 'echo: ${req.message}';
});
communicator.onReceivedData(
'example.TestData',
(TestData()
..index = 4
..message = 'qualified request')
.writeToBuffer(),
incomingNonce: 12,
);
await Future<void>.delayed(Duration.zero);
final response = communicator.transport.sentPackets.single;
expect(response.responseNonce, 12);
final data = TestData.fromBuffer(response.data);
expect(data.index, 5);
expect(data.message, 'echo: qualified request');
});
test(
'cannot register addRequestListener for a type already using addListener',
() {