From 840df6a63d1d854b450ccdf91501c1337e65bcfb Mon Sep 17 00:00:00 2001 From: toki Date: Sat, 11 Apr 2026 19:05:17 +0900 Subject: [PATCH] refactor: implement BaseClient pattern for Dart - Dart: Add BaseClient to eliminate duplicate code - Dart: Replace dispose() with close() (Future return) - Dart: Make isAlive and nonce read-only with @protected setter - Dart: Add meta package dependency for @protected - Dart: Implement BaseClient and BaseClient - 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 --- dart/crosstest/go_dart_client.dart | 4 +-- dart/lib/src/base_client.dart | 49 ++++++++++++++++++++++++++++ dart/lib/src/communicator.dart | 13 ++++++-- dart/lib/src/heartbeat_mixin.dart | 8 ++--- dart/lib/src/protobuf_client.dart | 42 ++++-------------------- dart/lib/src/protobuf_server.dart | 4 +-- dart/lib/src/ws_protobuf_client.dart | 42 +++--------------------- dart/lib/src/ws_protobuf_server.dart | 4 +-- dart/lib/toki_socket.dart | 1 + dart/pubspec.yaml | 1 + dart/test/socket_test.dart | 44 ++++++++++++------------- 11 files changed, 105 insertions(+), 107 deletions(-) create mode 100644 dart/lib/src/base_client.dart diff --git a/dart/crosstest/go_dart_client.dart b/dart/crosstest/go_dart_client.dart index 493694b..b3b7cf4 100644 --- a/dart/crosstest/go_dart_client.dart +++ b/dart/crosstest/go_dart_client.dart @@ -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"'); } diff --git a/dart/lib/src/base_client.dart b/dart/lib/src/base_client.dart new file mode 100644 index 0000000..5f2632d --- /dev/null +++ b/dart/lib/src/base_client.dart @@ -0,0 +1,49 @@ +import 'package:meta/meta.dart'; + +import 'communicator.dart'; +import 'heartbeat_mixin.dart'; + +abstract class BaseClient> extends Communicator + with HeartbeatMixin { + late final Self _self; + final List _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 close() async { + if (!isAlive) { + return; + } + isAlive = false; + stopHeartbeat(); + + final listeners = List.from(_disconnectListeners); + _disconnectListeners.clear(); + for (final fn in listeners) { + fn(_self); + } + + await closeTransport(); + } + + @protected + Future closeTransport(); +} diff --git a/dart/lib/src/communicator.dart b/dart/lib/src/communicator.dart index e623b39..d251b68 100644 --- a/dart/lib/src/communicator.dart +++ b/dart/lib/src/communicator.dart @@ -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 _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(); diff --git a/dart/lib/src/heartbeat_mixin.dart b/dart/lib/src/heartbeat_mixin.dart index e3cdaaf..d252873 100644 --- a/dart/lib/src/heartbeat_mixin.dart +++ b/dart/lib/src/heartbeat_mixin.dart @@ -10,8 +10,7 @@ mixin HeartbeatMixin on Communicator { ResponseChecker? _heartbeatChecker; bool _waitingHeartbeatResponse = false; - void onDisconnected(dynamic data); - void dispose(); + Future 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(); } } } diff --git a/dart/lib/src/protobuf_client.dart b/dart/lib/src/protobuf_client.dart index 074e66b..6d7ff4a 100644 --- a/dart/lib/src/protobuf_client.dart +++ b/dart/lib/src/protobuf_client.dart @@ -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 { final int _headerSize = 4; final Socket _socket; @@ -25,46 +24,23 @@ abstract class ProtobufClient extends Communicator with HeartbeatMixin { int? _length = null; late List _arrivedData = []; - List _onDisconnectListenerList = []; ProtobufClient(this._socket, int heartbeatIntervalTime, int heartbeatWaitTime, Map)> 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().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 closeTransport() async { try { await _socket.close(); _socket.destroy(); diff --git a/dart/lib/src/protobuf_server.dart b/dart/lib/src/protobuf_server.dart index c3168d8..2070763 100644 --- a/dart/lib/src/protobuf_server.dart +++ b/dart/lib/src/protobuf_server.dart @@ -65,7 +65,7 @@ abstract class ProtobufServer { void onDisconnectedClient(ProtobufClient client) { _clientList.remove(client); - client.dispose(); + client.close(); } Future broadcast(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(); diff --git a/dart/lib/src/ws_protobuf_client.dart b/dart/lib/src/ws_protobuf_client.dart index 9185108..1877e9a 100644 --- a/dart/lib/src/ws_protobuf_client.dart +++ b/dart/lib/src/ws_protobuf_client.dart @@ -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 { final WebSocket _ws; /// Plain WebSocket connection. @@ -24,45 +23,18 @@ abstract class WsProtobufClient extends Communicator with HeartbeatMixin { customClient: HttpClient(context: context ?? SecurityContext.defaultContext)); - List _onDisconnectListenerList = []; - WsProtobufClient(this._ws, int heartbeatIntervalTime, int heartbeatWaitTime, Map)> 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 closeTransport() async { try { await _ws.close(); } catch (_) { diff --git a/dart/lib/src/ws_protobuf_server.dart b/dart/lib/src/ws_protobuf_server.dart index a38afd9..ff19c76 100644 --- a/dart/lib/src/ws_protobuf_server.dart +++ b/dart/lib/src/ws_protobuf_server.dart @@ -66,7 +66,7 @@ abstract class WsProtobufServer { void onDisconnectedClient(WsProtobufClient client) { _clientList.remove(client); - client.dispose(); + client.close(); } Future broadcast(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(); diff --git a/dart/lib/toki_socket.dart b/dart/lib/toki_socket.dart index e811709..cab695f 100644 --- a/dart/lib/toki_socket.dart +++ b/dart/lib/toki_socket.dart @@ -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'; diff --git a/dart/pubspec.yaml b/dart/pubspec.yaml index aba365e..fcc8030 100644 --- a/dart/pubspec.yaml +++ b/dart/pubspec.yaml @@ -7,6 +7,7 @@ environment: sdk: '>=3.2.3 <4.0.0' dependencies: + meta: ^1.9.0 protobuf: ^3.1.0 dev_dependencies: diff --git a/dart/test/socket_test.dart b/dart/test/socket_test.dart index 259a799..c18b4bc 100644 --- a/dart/test/socket_test.dart +++ b/dart/test/socket_test.dart @@ -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(); });