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

Update protocol and add WebSocket protobuf client/server implementations

Browse source
This commit is contained in:
toki 2026-04-05 20:44:37 +09:00
parent d89b25948a
commit de309ca2f1
6 changed files with 448 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
.codex Normal file
View file

18
PROTOCOL.md
View file

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

154
dart/lib/src/ws_protobuf_client.dart Normal file
View file

@ -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<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',
customClient:
HttpClient(context: context ?? SecurityContext.defaultContext));
late ResponseChecker? _heartbeatChecker = null;
bool _isAlive = false;
bool _waitingHeartbeatResponse = false;
int _nonce = 0;
List<void Function(WsProtobufClient)> _onDisconnectListenerList = [];
WsProtobufClient(
this._ws,
this._heartbeatIntervalTime,
this._heartbeatWaitTime,
Map<String, GeneratedMessage Function(List<int>)> 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<int> ? 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 extends GeneratedMessage>(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
}
}
}
}

85
dart/lib/src/ws_protobuf_server.dart Normal file
View file

@ -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<WsProtobufClient> _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<HeartBeat>((_) => 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 extends GeneratedMessage>(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();
}
}

2
dart/lib/toki_socket.dart
View file

@ -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';

191
dart/test/socket_test.dart
View file

@ -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 = <TestData>[];
final connectedClients = <WsProtobufClient>[];
_TestWsServer() : super(_host, _testPortWs, (ws) => _TestWsClient(ws));
@override
void onClientConnected(WsProtobufClient client) {
connectedClients.add(client);
client.addListener<TestData>((data) => receivedMessages.add(data));
}
}
class _TestWsServerSsl extends WsProtobufServer {
final receivedMessages = <TestData>[];
final connectedClients = <WsProtobufClient>[];
_TestWsServerSsl(SecurityContext ctx)
: super.secure(_host, _testPortWss, ctx, (ws) => _TestWsClient(ws));
@override
void onClientConnected(WsProtobufClient client) {
connectedClients.add(client);
client.addListener<TestData>((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<TestData>();
client.addListener<TestData>((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<void>();
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<void>();
server.connectedClients.first.addDisconnectListener((_) {
if (!completer.isCompleted) completer.complete();
});
client.dispose();
await completer.future.timeout(const Duration(seconds: 2));
expect(completer.isCompleted, isTrue);
});
});
}