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

refactor: implement BaseClient pattern for Dart

Browse source 
- Dart: Add BaseClient<Self> to eliminate duplicate code
- Dart: Replace dispose() with close() (Future<void> return)
- Dart: Make isAlive and nonce read-only with @protected setter
- Dart: Add meta package dependency for @protected
- Dart: Implement BaseClient<ProtobufClient> and BaseClient<WsProtobufClient>
- Dart: Update communicator.dart with private fields and protected setters
- Dart: Update heartbeat_mixin.dart for new close() pattern
- Dart: Update all client/server implementations
- Dart: Update test files with new API
- Dart: Export base_client.dart in toki_socket.dart
This commit is contained in:
toki 2026-04-11 19:05:17 +09:00
parent 0736a87928
commit 840df6a63d
11 changed files with 105 additions and 107 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
dart/crosstest/go_dart_client.dart
View file

@ -86,12 +86,12 @@ Future<_ClientHandle> _connectWithRetry(String mode, int port) async {
final socket = await ProtobufClient.connect(_host, port)
.timeout(const Duration(milliseconds: 300));
final client = _TcpClient(socket);
return _ClientHandle(client, () async => client.dispose());
return _ClientHandle(client, client.close);
case 'ws':
final ws = await WsProtobufClient.connect(_host, port, path: _wsPath)
.timeout(const Duration(milliseconds: 300));
final client = _WsClient(ws);
return _ClientHandle(client, () async => client.dispose());
return _ClientHandle(client, client.close);
default:
throw ArgumentError('unknown mode "$mode"');
}

49
dart/lib/src/base_client.dart Normal file
View file

@ -0,0 +1,49 @@
import 'package:meta/meta.dart';
import 'communicator.dart';
import 'heartbeat_mixin.dart';
abstract class BaseClient<Self extends BaseClient<Self>> extends Communicator
with HeartbeatMixin {
late final Self _self;
final List<void Function(Self)> _disconnectListeners = [];
@protected
void initSelf(Self self) {
_self = self;
}
void addDisconnectListener(void Function(Self) fn) {
if (!_disconnectListeners.contains(fn)) {
_disconnectListeners.add(fn);
}
}
void removeDisconnectListener(void Function(Self) fn) {
_disconnectListeners.remove(fn);
}
void onDisconnected(dynamic _) {
close();
}
@override
Future<void> close() async {
if (!isAlive) {
return;
}
isAlive = false;
stopHeartbeat();
final listeners = List<void Function(Self)>.from(_disconnectListeners);
_disconnectListeners.clear();
for (final fn in listeners) {
fn(_self);
}
await closeTransport();
}
@protected
Future<void> closeTransport();
}

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

@ -1,6 +1,7 @@
// ignore_for_file: prefer_final_fields
import 'dart:async';
import 'package:meta/meta.dart';
import 'package:protobuf/protobuf.dart';
import 'packets/message_common.pb.dart';
@ -12,10 +13,18 @@ abstract class Communicator {
Future<void> _outboundWrite = Future.value();
/// Monotonically increasing nonce. Shared by send / sendRequest / response.
int nonce = 0;
int _nonce = 0;
int get nonce => _nonce;
@protected
set nonce(int value) => _nonce = value;
/// Whether the connection is alive. Set by each transport implementation.
bool isAlive = false;
bool _isAlive = false;
bool get isAlive => _isAlive;
@protected
set isAlive(bool value) => _isAlive = value;
Communicator();

8
dart/lib/src/heartbeat_mixin.dart
View file

@ -10,8 +10,7 @@ mixin HeartbeatMixin on Communicator {
ResponseChecker<Communicator>? _heartbeatChecker;
bool _waitingHeartbeatResponse = false;
void onDisconnected(dynamic data);
void dispose();
Future<void> close();
void initHeartbeat(int intervalSec, int waitSec) {
_heartbeatIntervalTime = intervalSec;
@ -29,8 +28,7 @@ mixin HeartbeatMixin on Communicator {
_heartbeatChecker =
ResponseChecker.second(this, _heartbeatWaitTime, (client) {
if (isAlive) {
onDisconnected(null);
dispose();
close();
}
});
}
@ -55,7 +53,7 @@ mixin HeartbeatMixin on Communicator {
..nonce = ++nonce
..data = data.writeToBuffer());
} catch (e) {
onDisconnected(null);
close();
}
}
}

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

@ -5,11 +5,10 @@ import 'dart:async';
import 'dart:typed_data';
import 'package:protobuf/protobuf.dart';
import 'communicator.dart';
import 'heartbeat_mixin.dart';
import 'base_client.dart';
import 'packets/message_common.pb.dart';
abstract class ProtobufClient extends Communicator with HeartbeatMixin {
abstract class ProtobufClient extends BaseClient<ProtobufClient> {
final int _headerSize = 4;
final Socket _socket;
@ -25,46 +24,23 @@ abstract class ProtobufClient extends Communicator with HeartbeatMixin {
int? _length = null;
late List<int> _arrivedData = [];
List<void Function(ProtobufClient)> _onDisconnectListenerList = [];
ProtobufClient(this._socket, int heartbeatIntervalTime, int heartbeatWaitTime,
Map<String, GeneratedMessage Function(List<int>)> parserMap) {
initSelf(this);
initHeartbeat(heartbeatIntervalTime, heartbeatWaitTime);
isAlive = true;
parserMap.addAll({
(HeartBeat).toString(): HeartBeat.fromBuffer,
});
super.initialize(parserMap);
_socket.listen(onData, onError: onError).asFuture().then(onDisconnected);
_socket.listen(onData, onError: onError).asFuture<void>().then((_) {
close();
});
addListener(onHeartBeat);
sendHeartBeat();
}
void addDisconnectListener(void Function(ProtobufClient) handler) {
if (!_onDisconnectListenerList.contains(handler)) {
_onDisconnectListenerList.add(handler);
}
}
void removeDisconnectListener(void Function(ProtobufClient) handler) {
if (_onDisconnectListenerList.contains(handler)) {
_onDisconnectListenerList.remove(handler);
}
}
@override
void onDisconnected(dynamic data) {
if (!isAlive) {
return;
}
isAlive = false;
stopHeartbeat();
for (var item in _onDisconnectListenerList) {
item.call(this);
}
_onDisconnectListenerList.clear();
}
void onError(dynamic e) {}
void onData(Uint8List data) {
@ -119,11 +95,7 @@ abstract class ProtobufClient extends Communicator with HeartbeatMixin {
}
@override
void dispose() async {
if (isAlive) {
isAlive = false;
stopHeartbeat();
}
Future<void> closeTransport() async {
try {
await _socket.close();
_socket.destroy();

4
dart/lib/src/protobuf_server.dart
View file

@ -65,7 +65,7 @@ abstract class ProtobufServer {
void onDisconnectedClient(ProtobufClient client) {
_clientList.remove(client);
client.dispose();
client.close();
}
Future broadcast<T extends GeneratedMessage>(T data) async {
@ -81,7 +81,7 @@ abstract class ProtobufServer {
await _server?.close();
await _secureServer?.close();
for (final client in clients) {
client.dispose();
await client.close();
}
_started = false;
return Future.value();

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

@ -5,11 +5,10 @@ import 'dart:async';
import 'dart:typed_data';
import 'package:protobuf/protobuf.dart';
import 'communicator.dart';
import 'heartbeat_mixin.dart';
import 'base_client.dart';
import 'packets/message_common.pb.dart';
abstract class WsProtobufClient extends Communicator with HeartbeatMixin {
abstract class WsProtobufClient extends BaseClient<WsProtobufClient> {
final WebSocket _ws;
/// Plain WebSocket connection.
@ -24,45 +23,18 @@ abstract class WsProtobufClient extends Communicator with HeartbeatMixin {
customClient:
HttpClient(context: context ?? SecurityContext.defaultContext));
List<void Function(WsProtobufClient)> _onDisconnectListenerList = [];
WsProtobufClient(this._ws, int heartbeatIntervalTime, int heartbeatWaitTime,
Map<String, GeneratedMessage Function(List<int>)> parserMap) {
initSelf(this);
initHeartbeat(heartbeatIntervalTime, heartbeatWaitTime);
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: close);
addListener(onHeartBeat);
sendHeartBeat();
}
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);
}
}
@override
void onDisconnected(dynamic data) {
if (!isAlive) {
return;
}
isAlive = false;
stopHeartbeat();
for (var item in _onDisconnectListenerList) {
item.call(this);
}
_onDisconnectListenerList.clear();
}
void onError(dynamic e) {}
void _onMessage(dynamic data) {
@ -79,11 +51,7 @@ abstract class WsProtobufClient extends Communicator with HeartbeatMixin {
}
@override
void dispose() async {
if (isAlive) {
isAlive = false;
stopHeartbeat();
}
Future<void> closeTransport() async {
try {
await _ws.close();
} catch (_) {

4
dart/lib/src/ws_protobuf_server.dart
View file

@ -66,7 +66,7 @@ abstract class WsProtobufServer {
void onDisconnectedClient(WsProtobufClient client) {
_clientList.remove(client);
client.dispose();
client.close();
}
Future broadcast<T extends GeneratedMessage>(T data) async {
@ -81,7 +81,7 @@ abstract class WsProtobufServer {
_clientList.clear();
await _server?.close(force: true);
for (final client in clients) {
client.dispose();
await client.close();
}
_started = false;
return Future.value();

1
dart/lib/toki_socket.dart
View file

@ -1,6 +1,7 @@
library toki_socket;
export 'src/communicator.dart';
export 'src/base_client.dart';
export 'src/protobuf_client.dart';
export 'src/protobuf_server.dart';
export 'src/response_checker.dart';

1
dart/pubspec.yaml
View file

@ -7,6 +7,7 @@ environment:
sdk: '>=3.2.3 <4.0.0'
dependencies:
meta: ^1.9.0
protobuf: ^3.1.0
dev_dependencies:

44
dart/test/socket_test.dart
View file

@ -194,7 +194,7 @@ void main() {
});
tearDown(() async {
client.dispose();
await client.close();
await Future.delayed(const Duration(milliseconds: 100));
await server.stop();
});
@ -258,7 +258,7 @@ void main() {
if (!completer.isCompleted) completer.complete();
});
client.dispose();
await client.close();
await completer.future.timeout(const Duration(seconds: 2));
expect(completer.isCompleted, isTrue);
});
@ -283,8 +283,8 @@ void main() {
expect(server.receivedMessages, hasLength(1));
});
test('dispose 후 isAlive가 false다', () async {
client.dispose();
test('close 후 isAlive가 false다', () async {
await client.close();
await Future.delayed(const Duration(milliseconds: 100));
expect(client.isAlive, isFalse);
});
@ -314,12 +314,12 @@ void main() {
expect(first.message, equals('broadcast'));
expect(second.message, equals('broadcast'));
} finally {
secondClient.dispose();
await secondClient.close();
}
});
test('dispose 후 send는 무시된다', () async {
client.dispose();
test('close 후 send는 무시된다', () async {
await client.close();
await Future.delayed(const Duration(milliseconds: 100));
await expectLater(
@ -347,7 +347,7 @@ void main() {
await completer.future.timeout(const Duration(seconds: 4));
expect(client.isAlive, isFalse);
} finally {
client.dispose();
await client.close();
await server.stop();
}
});
@ -366,7 +366,7 @@ void main() {
await completer.future.timeout(const Duration(seconds: 4));
expect(client.isAlive, isFalse);
} finally {
client.dispose();
await client.close();
await server.stop();
}
});
@ -417,7 +417,7 @@ void main() {
});
tearDown(() async {
client.dispose();
await client.close();
await Future.delayed(const Duration(milliseconds: 100));
await server.stop();
});
@ -458,7 +458,7 @@ void main() {
if (!completer.isCompleted) completer.complete();
});
client.dispose();
await client.close();
await completer.future.timeout(const Duration(seconds: 2));
expect(completer.isCompleted, isTrue);
});
@ -497,7 +497,7 @@ void main() {
});
tearDown(() async {
client.dispose();
await client.close();
await Future.delayed(const Duration(milliseconds: 100));
await server.stop();
});
@ -538,7 +538,7 @@ void main() {
if (!completer.isCompleted) completer.complete();
});
client.dispose();
await client.close();
await completer.future.timeout(const Duration(seconds: 2));
expect(completer.isCompleted, isTrue);
});
@ -554,8 +554,8 @@ void main() {
expect(client.isAlive, isFalse);
});
test('WS dispose 후 isAlive가 false다', () async {
client.dispose();
test('WS close 후 isAlive가 false다', () async {
await client.close();
await Future.delayed(const Duration(milliseconds: 100));
expect(client.isAlive, isFalse);
});
@ -585,12 +585,12 @@ void main() {
expect(first.message, equals('ws broadcast'));
expect(second.message, equals('ws broadcast'));
} finally {
secondClient.dispose();
await secondClient.close();
}
});
test('WS dispose 후 send는 무시된다', () async {
client.dispose();
test('WS close 후 send는 무시된다', () async {
await client.close();
await Future.delayed(const Duration(milliseconds: 100));
await expectLater(
@ -648,7 +648,7 @@ void main() {
});
tearDown(() async {
client.dispose();
await client.close();
await Future.delayed(const Duration(milliseconds: 100));
await server.stop();
});
@ -674,7 +674,7 @@ void main() {
if (!completer.isCompleted) completer.complete();
});
client.dispose();
await client.close();
await completer.future.timeout(const Duration(seconds: 2));
expect(completer.isCompleted, isTrue);
});
@ -695,7 +695,7 @@ void main() {
});
tearDown(() async {
client.dispose();
await client.close();
await Future.delayed(const Duration(milliseconds: 100));
await server.stop();
});
@ -750,7 +750,7 @@ void main() {
});
tearDown(() async {
client.dispose();
await client.close();
await Future.delayed(const Duration(milliseconds: 100));
await server.stop();
});