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

refactor: implement HeartbeatMixin for heartbeat handling

Browse source 
- Dart: Replace ResponseChecker with HeartbeatMixin
- Dart: Remove heartbeat-related fields from ProtobufClient/Server
- Dart: Add HeartbeatMixin for cleaner heartbeat logic
- Go: Add timeout test in communicator_test.go
- Go: Add heartbeat_test.go with heartbeat functionality tests
- Go: Add TLS test in tls_test.go
- Go: Add message type mismatch tests
- Go: Improve tcp_test.go and ws_test.go with comprehensive tests
- Dart: Add comprehensive socket_test.dart with multiple test scenarios
- Add untracked documentation files
This commit is contained in:
toki 2026-04-11 16:33:40 +09:00
parent 9a0ebd0e22
commit 7f8771b89e
14 changed files with 834 additions and 146 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
CODE_REVIEW_REVIEW_TEST.md Normal file
View file

@ -0,0 +1,58 @@
# 테스트 보강 리뷰 후속 코드리뷰 참고문서
## 작업 배경
`IMPROVEMENT_PLAN_REVIEW_TEST.md`에 정리된 테스트 보강 리뷰 후속 항목을 반영했다. 핵심은 Dart 라이브러리 코드에 남아 있던 `print()` 제거와, Go 서버 종료 테스트에서 테스트 의도를 흐리는 중복 `Stop()` 호출 정리다.
## 변경 요약
### Dart 서버 로그 제거
- `dart/lib/src/protobuf_server.dart`
- 파일 상단 `avoid_print` suppress 제거.
- `onDisconnectedClient`의 `print('Client disconnected')` 제거.
- `dart/lib/src/ws_protobuf_server.dart`
- 파일 상단 `avoid_print` suppress 제거.
- `onDisconnectedClient`의 `print('WS Client disconnected')` 제거.
라이브러리 코드가 disconnect 이벤트마다 stdout에 직접 로그를 남기지 않도록 정리했다.
### Go 서버 종료 테스트 정리
- `go/test/tcp_test.go`
- `TestTcpServerStopDisconnectsClients` 내부의 `defer server.Stop()` 제거.
- `go/test/ws_test.go`
- `TestWsServerStopDisconnectsClients` 내부의 `defer server.Stop()` 제거.
두 테스트는 본문에서 명시적으로 `server.Stop()`을 호출하고 그 결과로 client disconnect 콜백이 호출되는지 검증한다. 따라서 별도 deferred stop을 제거해 테스트 의도를 더 분명하게 만들었다.
## 검증 결과
```bash
cd dart
dart analyze
```
결과: 통과. `No issues found!`
```bash
cd dart
dart test
```
결과: 통과. 총 39개 테스트 통과.
```bash
cd go
PATH=/config/go-sdk/go/bin:/config/go/bin:$PATH GOCACHE=/tmp/go-build GOMODCACHE=/tmp/go-mod go test ./test/... -v
```
결과: 통과.
## 리뷰 포인트
- Dart 서버 코드에서 `print()` 호출은 제거됐고, 관련 `avoid_print` ignore도 함께 제거됐다.
- Go의 다른 테스트에 있는 `defer server.Stop()`은 각 테스트의 일반 teardown 역할이므로 유지했다.
- `TestTcpServerStopDisconnectsClients`, `TestWsServerStopDisconnectsClients`는 명시적 `server.Stop()` 호출 하나만으로 disconnect 동작을 검증한다.

112
IMPROVEMENT_PLAN_REVIEW_TEST.md Normal file
View file

@ -0,0 +1,112 @@
# 테스트 보강 코드리뷰 후속 작업계획서
## 개요
`IMPROVEMENT_PLAN_TEST.md` 기반 테스트 추가 작업에 대한 코드리뷰 결과, 두 가지 후속 수정이 필요하다.
---
## 개선 항목
### [Review-1] Dart 서버 print() 제거 ⭐ 필수
**문제**
이전 Dart 개선 작업([Dart-4])에서 라이브러리 코드의 `print()`를 제거했으나 서버 파일 두 곳이 누락됐다.
```dart
// protobuf_server.dart:69
void onDisconnectedClient(ProtobufClient client) {
_clientList.remove(client);
client.dispose();
print('Client disconnected'); // ← 제거 대상
}
// ws_protobuf_server.dart:70
void onDisconnectedClient(WsProtobufClient client) {
_clientList.remove(client);
client.dispose();
print('WS Client disconnected'); // ← 제거 대상
}
```
파일 상단의 `// ignore_for_file: avoid_print` suppress 주석도 함께 삭제한다.
**수정 파일**
- `dart/lib/src/protobuf_server.dart`
- 1행 `// ignore_for_file: avoid_print, prefer_final_fields` → `// ignore_for_file: prefer_final_fields`
- `print('Client disconnected')` 한 줄 제거
- `dart/lib/src/ws_protobuf_server.dart`
- 1행 `// ignore_for_file: avoid_print, prefer_final_fields` → `// ignore_for_file: prefer_final_fields`
- `print('WS Client disconnected')` 한 줄 제거
**검증**
```bash
cd dart && dart analyze && dart test
```
---
### [Review-2] Go 서버 종료 테스트 — 이중 Stop() 명확화 ⭐ 선택
**문제**
`TestTcpServerStopDisconnectsClients`와 `TestWsServerStopDisconnectsClients`에서 `server.Stop()`이 두 번 호출된다.
```go
defer server.Stop() // ← defer 등록
// ... 테스트 본문 ...
if err := server.Stop(); err != nil { // ← 명시적 호출
t.Fatal(err)
}
```
테스트 목적상 `Stop()` 이후 disconnect 콜백이 울리는지를 검증하는 것이 핵심이므로, `defer server.Stop()`은 불필요하다. 오히려 이중 호출이 `Stop()` 멱등성에 의존하는 것처럼 읽혀 의도를 흐린다.
**해결 방법**
`defer server.Stop()`을 제거한다. 명시적 `server.Stop()` 호출 이후 서버는 이미 중단 상태이므로 teardown 역할은 필요 없다.
```go
// defer server.Stop() ← 제거
// ...
if err := server.Stop(); err != nil {
t.Fatal(err)
}
```
**수정 파일**
- `go/test/tcp_test.go` — `TestTcpServerStopDisconnectsClients` 내 `defer server.Stop()` 제거
- `go/test/ws_test.go` — `TestWsServerStopDisconnectsClients` 내 `defer server.Stop()` 제거
**검증**
```bash
cd go && go test ./test/... -run "TestTcpServerStopDisconnectsClients|TestWsServerStopDisconnectsClients" -v
```
---
## 우선순위
| 우선순위 | 항목 | 이유 |
|---------|------|------|
| 1 | Review-1 (print 제거) | 라이브러리 코드에 로그가 남는 버그. analyze 경고 발생 |
| 2 | Review-2 (이중 Stop 정리) | 기능 문제 아님. 가독성 개선 |
---
## 최종 검증
```bash
# Go
cd go && go test ./test/... -v
# Dart
cd dart && dart analyze && dart test
```

66
dart/lib/src/heartbeat_mixin.dart Normal file
View file

@ -0,0 +1,66 @@
import 'package:protobuf/protobuf.dart';
import 'communicator.dart';
import 'packets/message_common.pb.dart';
import 'response_checker.dart';
mixin HeartbeatMixin on Communicator {
late int _heartbeatIntervalTime;
late int _heartbeatWaitTime;
ResponseChecker<Communicator>? _heartbeatChecker;
bool _waitingHeartbeatResponse = false;
void onDisconnected(dynamic data);
void dispose();
void initHeartbeat(int intervalSec, int waitSec) {
_heartbeatIntervalTime = intervalSec;
_heartbeatWaitTime = waitSec;
}
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());
}
}
@override
Future<void> send<T extends GeneratedMessage>(T data) async {
if (isAlive) {
try {
await queuePacket(PacketBase()
..typeName = data.info_.qualifiedMessageName
..nonce = ++nonce
..data = data.writeToBuffer());
} catch (e) {
onDisconnected(null);
}
}
}
void stopHeartbeat() {
_heartbeatChecker?.responded();
}
}

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

@ -1,4 +1,4 @@
// ignore_for_file: avoid_init_to_null, prefer_final_fields, avoid_print
// ignore_for_file: avoid_init_to_null, prefer_final_fields
import 'dart:io';
import 'dart:async';
@ -6,13 +6,11 @@ import 'dart:typed_data';
import 'package:protobuf/protobuf.dart';
import 'communicator.dart';
import 'response_checker.dart';
import 'heartbeat_mixin.dart';
import 'packets/message_common.pb.dart';
abstract class ProtobufClient extends Communicator {
abstract class ProtobufClient extends Communicator with HeartbeatMixin {
final int _headerSize = 4;
late int _heartbeatIntervalTime;
late int _heartbeatWaitTime;
final Socket _socket;
/// Plain TCP connection.
@ -25,18 +23,13 @@ abstract class ProtobufClient extends Communicator {
SecureSocket.connect(host, port,
context: context ?? SecurityContext.defaultContext);
late ResponseChecker? _heartbeatChecker = null;
int? _length = null;
bool _waitingHeartbeatResponse = false;
late List<int> _arrivedData = [];
List<void Function(ProtobufClient)> _onDisconnectListenerList = [];
ProtobufClient(
this._socket,
this._heartbeatIntervalTime,
this._heartbeatWaitTime,
ProtobufClient(this._socket, int heartbeatIntervalTime, int heartbeatWaitTime,
Map<String, GeneratedMessage Function(List<int>)> parserMap) {
print('Connected New Client');
initHeartbeat(heartbeatIntervalTime, heartbeatWaitTime);
isAlive = true;
parserMap.addAll({
(HeartBeat).toString(): HeartBeat.fromBuffer,
@ -47,34 +40,6 @@ abstract class ProtobufClient extends Communicator {
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(ProtobufClient) handler) {
if (!_onDisconnectListenerList.contains(handler)) {
_onDisconnectListenerList.add(handler);
@ -87,28 +52,27 @@ abstract class ProtobufClient extends Communicator {
}
}
@override
void onDisconnected(dynamic data) {
if (!isAlive) {
return;
}
isAlive = false;
_heartbeatChecker?.responded();
stopHeartbeat();
for (var item in _onDisconnectListenerList) {
item.call(this);
}
_onDisconnectListenerList.clear();
}
void onError(dynamic e) {
print('=========> onError: $e');
}
void onError(dynamic e) {}
void onData(Uint8List data) {
try {
_arrivedData.addAll(data);
_parsing();
} on Exception catch (e) {
print(e);
} on Exception {
// Ignore malformed partial data and keep the socket state unchanged.
}
}
@ -155,24 +119,10 @@ abstract class ProtobufClient extends Communicator {
}
@override
Future<void> send<T extends GeneratedMessage>(T data) async {
if (isAlive) {
try {
await queuePacket(PacketBase()
..typeName = data.info_.qualifiedMessageName
..nonce = ++nonce
..data = data.writeToBuffer());
} catch (e) {
onDisconnected(null);
}
}
return Future.value();
}
void dispose() async {
if (isAlive) {
isAlive = false;
_heartbeatChecker?.responded();
stopHeartbeat();
}
try {
await _socket.close();

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

@ -1,4 +1,4 @@
// ignore_for_file: avoid_print, prefer_final_fields
// ignore_for_file: prefer_final_fields
import 'dart:io';
import 'package:protobuf/protobuf.dart';
@ -66,7 +66,6 @@ abstract class ProtobufServer {
void onDisconnectedClient(ProtobufClient client) {
_clientList.remove(client);
client.dispose();
print('Client disconnected');
}
Future broadcast<T extends GeneratedMessage>(T data) async {
@ -77,8 +76,13 @@ abstract class ProtobufServer {
}
Future stop() async {
final clients = List<ProtobufClient>.from(_clientList);
_clientList.clear();
await _server?.close();
await _secureServer?.close();
for (final client in clients) {
client.dispose();
}
_started = false;
return Future.value();
}

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

@ -1,4 +1,4 @@
// ignore_for_file: avoid_init_to_null, prefer_final_fields, avoid_print
// ignore_for_file: avoid_init_to_null, prefer_final_fields
import 'dart:io';
import 'dart:async';
@ -6,12 +6,10 @@ import 'dart:typed_data';
import 'package:protobuf/protobuf.dart';
import 'communicator.dart';
import 'response_checker.dart';
import 'heartbeat_mixin.dart';
import 'packets/message_common.pb.dart';
abstract class WsProtobufClient extends Communicator {
late int _heartbeatIntervalTime;
late int _heartbeatWaitTime;
abstract class WsProtobufClient extends Communicator with HeartbeatMixin {
final WebSocket _ws;
/// Plain WebSocket connection.
@ -26,16 +24,11 @@ abstract class WsProtobufClient extends Communicator {
customClient:
HttpClient(context: context ?? SecurityContext.defaultContext));
late ResponseChecker? _heartbeatChecker = null;
bool _waitingHeartbeatResponse = false;
List<void Function(WsProtobufClient)> _onDisconnectListenerList = [];
WsProtobufClient(
this._ws,
this._heartbeatIntervalTime,
this._heartbeatWaitTime,
WsProtobufClient(this._ws, int heartbeatIntervalTime, int heartbeatWaitTime,
Map<String, GeneratedMessage Function(List<int>)> parserMap) {
print('Connected New WS Client');
initHeartbeat(heartbeatIntervalTime, heartbeatWaitTime);
isAlive = true;
parserMap.addAll({(HeartBeat).toString(): HeartBeat.fromBuffer});
super.initialize(parserMap);
@ -45,34 +38,6 @@ abstract class WsProtobufClient extends Communicator {
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);
@ -85,21 +50,20 @@ abstract class WsProtobufClient extends Communicator {
}
}
@override
void onDisconnected(dynamic data) {
if (!isAlive) {
return;
}
isAlive = false;
_heartbeatChecker?.responded();
stopHeartbeat();
for (var item in _onDisconnectListenerList) {
item.call(this);
}
_onDisconnectListenerList.clear();
}
void onError(dynamic e) {
print('=========> WS onError: $e');
}
void onError(dynamic e) {}
void _onMessage(dynamic data) {
final bytes = data is List<int> ? data : (data as Uint8List).toList();
@ -115,24 +79,10 @@ abstract class WsProtobufClient extends Communicator {
}
@override
Future<void> send<T extends GeneratedMessage>(T data) async {
if (isAlive) {
try {
await queuePacket(PacketBase()
..typeName = data.info_.qualifiedMessageName
..nonce = ++nonce
..data = data.writeToBuffer());
} catch (e) {
onDisconnected(null);
}
}
return Future.value();
}
void dispose() async {
if (isAlive) {
isAlive = false;
_heartbeatChecker?.responded();
stopHeartbeat();
}
try {
await _ws.close();

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

@ -1,4 +1,4 @@
// ignore_for_file: avoid_print, prefer_final_fields
// ignore_for_file: prefer_final_fields
import 'dart:io';
import 'package:protobuf/protobuf.dart';
@ -30,8 +30,8 @@ abstract class WsProtobufServer {
/// ..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)
WsProtobufServer.secure(this._host, this._port,
SecurityContext securityContext, this._createNewClient)
: _securityContext = securityContext;
Future start() async {
@ -67,7 +67,6 @@ abstract class WsProtobufServer {
void onDisconnectedClient(WsProtobufClient client) {
_clientList.remove(client);
client.dispose();
print('WS Client disconnected');
}
Future broadcast<T extends GeneratedMessage>(T data) async {
@ -78,7 +77,12 @@ abstract class WsProtobufServer {
}
Future stop() async {
final clients = List<WsProtobufClient>.from(_clientList);
_clientList.clear();
await _server?.close(force: true);
for (final client in clients) {
client.dispose();
}
_started = false;
return Future.value();
}

259
dart/test/socket_test.dart
View file

@ -108,6 +108,56 @@ class _TestWsReqServer extends WsProtobufServer {
}
}
class _FastHeartbeatClient extends ProtobufClient {
_FastHeartbeatClient(Socket socket)
: super(socket, 1, 1, {
TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
});
}
class _NoHeartbeatClient extends ProtobufClient {
_NoHeartbeatClient(Socket socket)
: super(socket, 5, 3, {
TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
});
@override
void onHeartBeat(HeartBeat data) {}
}
class _NoHeartbeatServer extends ProtobufServer {
_NoHeartbeatServer()
: super(_host, _testPort, (socket) => _NoHeartbeatClient(socket));
@override
void onClientConnected(ProtobufClient client) {}
}
class _FastHeartbeatWsClient extends WsProtobufClient {
_FastHeartbeatWsClient(WebSocket ws)
: super(ws, 1, 1, {
TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
});
}
class _NoHeartbeatWsClient extends WsProtobufClient {
_NoHeartbeatWsClient(WebSocket ws)
: super(ws, 5, 3, {
TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
});
@override
void onHeartBeat(HeartBeat data) {}
}
class _NoHeartbeatWsServer extends WsProtobufServer {
_NoHeartbeatWsServer()
: super(_host, _testPortWs, (ws) => _NoHeartbeatWsClient(ws));
@override
void onClientConnected(WsProtobufClient client) {}
}
// ── 테스트 ──────────────────────────────────────────────────────
void main() {
@ -161,7 +211,8 @@ void main() {
expect(server.receivedMessages, hasLength(1));
expect(server.receivedMessages.first.index, equals(42));
expect(server.receivedMessages.first.message, equals('hello toki-socket'));
expect(
server.receivedMessages.first.message, equals('hello toki-socket'));
});
test('여러 메시지를 순서대로 수신한다', () async {
@ -187,7 +238,8 @@ void main() {
..index = 99
..message = 'from server');
final received = await completer.future.timeout(const Duration(seconds: 2));
final received =
await completer.future.timeout(const Duration(seconds: 2));
expect(received.index, equals(99));
expect(received.message, equals('from server'));
});
@ -211,6 +263,17 @@ void main() {
expect(completer.isCompleted, isTrue);
});
test('서버 stop 시 클라이언트 disconnect 콜백이 호출된다', () async {
final completer = Completer<void>();
client.addDisconnectListener((_) {
if (!completer.isCompleted) completer.complete();
});
await server.stop();
await completer.future.timeout(const Duration(seconds: 2));
expect(client.isAlive, isFalse);
});
test('HeartBeat interval 동안 연결이 유지된다', () async {
await Future.delayed(const Duration(seconds: 2));
await client.send(TestData()
@ -219,6 +282,94 @@ void main() {
await Future.delayed(const Duration(milliseconds: 200));
expect(server.receivedMessages, hasLength(1));
});
test('dispose 후 isAlive가 false다', () async {
client.dispose();
await Future.delayed(const Duration(milliseconds: 100));
expect(client.isAlive, isFalse);
});
test('서버가 모든 클라이언트에게 브로드캐스트한다', () async {
final socket = await ProtobufClient.connect(_host, _testPort);
final secondClient = _TestClient(socket);
try {
await Future.delayed(const Duration(milliseconds: 100));
final firstReceived = Completer<TestData>();
final secondReceived = Completer<TestData>();
client.addListener<TestData>((data) {
if (!firstReceived.isCompleted) firstReceived.complete(data);
});
secondClient.addListener<TestData>((data) {
if (!secondReceived.isCompleted) secondReceived.complete(data);
});
await server.broadcast(TestData()
..index = 77
..message = 'broadcast');
final first =
await firstReceived.future.timeout(const Duration(seconds: 2));
final second =
await secondReceived.future.timeout(const Duration(seconds: 2));
expect(first.message, equals('broadcast'));
expect(second.message, equals('broadcast'));
} finally {
secondClient.dispose();
}
});
test('dispose 후 send는 무시된다', () async {
client.dispose();
await Future.delayed(const Duration(milliseconds: 100));
await expectLater(
client.send(TestData()
..index = 1
..message = 'ignored'),
completes,
);
expect(client.isAlive, isFalse);
});
});
group('Heartbeat timeout', () {
test('TCP heartbeat 타임아웃 시 disconnect 콜백이 호출된다', () async {
final server = _NoHeartbeatServer();
await server.start();
final socket = await ProtobufClient.connect(_host, _testPort);
final client = _FastHeartbeatClient(socket);
final completer = Completer<void>();
client.addDisconnectListener((_) {
if (!completer.isCompleted) completer.complete();
});
try {
await completer.future.timeout(const Duration(seconds: 4));
expect(client.isAlive, isFalse);
} finally {
client.dispose();
await server.stop();
}
});
test('WS heartbeat 타임아웃 시 disconnect 콜백이 호출된다', () async {
final server = _NoHeartbeatWsServer();
await server.start();
final ws = await WsProtobufClient.connect(_host, _testPortWs);
final client = _FastHeartbeatWsClient(ws);
final completer = Completer<void>();
client.addDisconnectListener((_) {
if (!completer.isCompleted) completer.complete();
});
try {
await completer.future.timeout(const Duration(seconds: 4));
expect(client.isAlive, isFalse);
} finally {
client.dispose();
await server.stop();
}
});
});
// ── SSL/TLS ──────────────────────────────────────────────────
@ -255,10 +406,10 @@ void main() {
server = _TestServerSsl(serverCtx);
await server.start();
final clientCtx = SecurityContext()
..setTrustedCertificates(_certPath);
final clientCtx = SecurityContext()..setTrustedCertificates(_certPath);
final socket = await ProtobufClient.connectSecure(
_host, _testPortSsl,
_host,
_testPortSsl,
context: clientCtx,
);
client = _TestClient(socket);
@ -295,7 +446,8 @@ void main() {
..index = 55
..message = 'from ssl server');
final received = await completer.future.timeout(const Duration(seconds: 2));
final received =
await completer.future.timeout(const Duration(seconds: 2));
expect(received.index, equals(55));
expect(received.message, equals('from ssl server'));
});
@ -374,7 +526,8 @@ void main() {
..index = 99
..message = 'from ws server');
final received = await completer.future.timeout(const Duration(seconds: 2));
final received =
await completer.future.timeout(const Duration(seconds: 2));
expect(received.index, equals(99));
expect(received.message, equals('from ws server'));
});
@ -389,6 +542,65 @@ void main() {
await completer.future.timeout(const Duration(seconds: 2));
expect(completer.isCompleted, isTrue);
});
test('WS 서버 stop 시 클라이언트 disconnect 콜백이 호출된다', () async {
final completer = Completer<void>();
client.addDisconnectListener((_) {
if (!completer.isCompleted) completer.complete();
});
await server.stop();
await completer.future.timeout(const Duration(seconds: 2));
expect(client.isAlive, isFalse);
});
test('WS dispose 후 isAlive가 false다', () async {
client.dispose();
await Future.delayed(const Duration(milliseconds: 100));
expect(client.isAlive, isFalse);
});
test('WS 서버가 모든 클라이언트에게 브로드캐스트한다', () async {
final ws = await WsProtobufClient.connect(_host, _testPortWs);
final secondClient = _TestWsClient(ws);
try {
await Future.delayed(const Duration(milliseconds: 100));
final firstReceived = Completer<TestData>();
final secondReceived = Completer<TestData>();
client.addListener<TestData>((data) {
if (!firstReceived.isCompleted) firstReceived.complete(data);
});
secondClient.addListener<TestData>((data) {
if (!secondReceived.isCompleted) secondReceived.complete(data);
});
await server.broadcast(TestData()
..index = 88
..message = 'ws broadcast');
final first =
await firstReceived.future.timeout(const Duration(seconds: 2));
final second =
await secondReceived.future.timeout(const Duration(seconds: 2));
expect(first.message, equals('ws broadcast'));
expect(second.message, equals('ws broadcast'));
} finally {
secondClient.dispose();
}
});
test('WS dispose 후 send는 무시된다', () async {
client.dispose();
await Future.delayed(const Duration(milliseconds: 100));
await expectLater(
client.send(TestData()
..index = 1
..message = 'ignored'),
completes,
);
expect(client.isAlive, isFalse);
});
});
// ── WebSocket SSL ─────────────────────────────────────────────
@ -427,7 +639,8 @@ void main() {
final clientCtx = SecurityContext()..setTrustedCertificates(_certPath);
final ws = await WsProtobufClient.connectSecure(
_host, _testPortWss,
_host,
_testPortWss,
context: clientCtx,
);
client = _TestWsClient(ws);
@ -500,11 +713,18 @@ void main() {
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'),
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));
final results =
await Future.wait(futures).timeout(const Duration(seconds: 3));
expect(results[0].index, equals(2));
expect(results[1].index, equals(4));
@ -548,11 +768,18 @@ void main() {
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'),
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));
final results =
await Future.wait(futures).timeout(const Duration(seconds: 3));
expect(results[0].index, equals(2));
expect(results[1].index, equals(4));

22
go/test/communicator_test.go
View file

@ -1,6 +1,7 @@
package toki_socket_test
import (
"strings"
"sync"
"testing"
"time"
@ -82,6 +83,27 @@ func TestSendRequestTypeMismatch(t *testing.T) {
}
}
func TestSendRequestTimeout(t *testing.T) {
transport := &fakeTransport{}
communicator := toki.NewCommunicator(transport, testParserMap())
defer communicator.Close()
_, err := toki.SendRequestTyped[*packets.TestData, *packets.TestData](
communicator,
&packets.TestData{Index: 1, Message: "no response"},
25*time.Millisecond,
)
if err == nil {
t.Fatal("expected request timeout error")
}
if !strings.Contains(err.Error(), "timeout") {
t.Fatalf("expected timeout error, got %v", err)
}
if len(transport.sent()) != 1 {
t.Fatalf("expected one request packet to be sent, got %d", len(transport.sent()))
}
}
func TestListenerAndRequestListenerConflict(t *testing.T) {
transport := &fakeTransport{}
communicator := toki.NewCommunicator(transport, testParserMap())

53
go/test/heartbeat_test.go
View file

@ -1,11 +1,16 @@
package toki_socket_test
import (
"context"
"fmt"
"io"
"net"
"net/http"
"testing"
"time"
"nhooyr.io/websocket"
toki "toki-labs.com/toki_socket/go"
)
@ -34,3 +39,51 @@ func TestHeartbeatDisconnectsWithoutResponse(t *testing.T) {
t.Fatal("client is still marked alive after heartbeat timeout")
}
}
func TestWsHeartbeatDisconnectsWithoutResponse(t *testing.T) {
port := freePort(t)
mux := http.NewServeMux()
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
conn, err := websocket.Accept(w, r, nil)
if err != nil {
return
}
defer conn.Close(websocket.StatusNormalClosure, "")
for {
if _, _, err := conn.Read(r.Context()); err != nil {
return
}
}
})
server := &http.Server{Addr: fmt.Sprintf("127.0.0.1:%d", port), Handler: mux}
ln, err := net.Listen("tcp", server.Addr)
if err != nil {
t.Fatal(err)
}
defer server.Close()
go func() {
_ = server.Serve(ln)
}()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
client, err := toki.DialWsWithHeartbeat(ctx, "127.0.0.1", port, "/", 1, 1, testParserMap())
if err != nil {
t.Fatal(err)
}
defer client.Close()
disconnected := make(chan struct{}, 1)
client.AddDisconnectListener(func(*toki.WsClient) {
disconnected <- struct{}{}
})
select {
case <-disconnected:
case <-time.After(3500 * time.Millisecond):
t.Fatal("websocket heartbeat timeout did not disconnect client")
}
if client.IsAlive() {
t.Fatal("websocket client is still marked alive after heartbeat timeout")
}
}

62
go/test/tcp_test.go
View file

@ -61,7 +61,6 @@ func TestTcpBroadcast(t *testing.T) {
if err := server.Start(ctx); err != nil {
t.Fatal(err)
}
defer server.Stop()
client, err := toki.DialTcp(ctx, "127.0.0.1", port, 0, 0, testParserMap())
if err != nil {
@ -74,9 +73,9 @@ func TestTcpBroadcast(t *testing.T) {
received <- m
})
for deadline := time.Now().Add(time.Second); time.Now().Before(deadline) && len(server.Clients()) == 0; {
time.Sleep(time.Millisecond)
}
waitForCondition(t, time.Second, func() bool {
return len(server.Clients()) == 1
}, "client did not connect")
if err := server.Broadcast(&packets.TestData{Index: 9, Message: "broadcast"}); err != nil {
t.Fatal(err)
}
@ -91,6 +90,61 @@ func TestTcpBroadcast(t *testing.T) {
}
}
func TestTcpServerStopDisconnectsClients(t *testing.T) {
port := freePort(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
server := toki.NewTcpServer("127.0.0.1", port, func(conn net.Conn) *toki.TcpClient {
return toki.NewTcpClient(conn, 0, 0, testParserMap())
})
if err := server.Start(ctx); err != nil {
t.Fatal(err)
}
client, err := toki.DialTcp(ctx, "127.0.0.1", port, 0, 0, testParserMap())
if err != nil {
t.Fatal(err)
}
defer client.Close()
disconnected := make(chan struct{}, 1)
client.AddDisconnectListener(func(*toki.TcpClient) {
disconnected <- struct{}{}
})
waitForCondition(t, time.Second, func() bool {
return len(server.Clients()) == 1
}, "client did not connect")
if err := server.Stop(); err != nil {
t.Fatal(err)
}
select {
case <-disconnected:
case <-time.After(2 * time.Second):
t.Fatal("client disconnect callback timed out")
}
if client.IsAlive() {
t.Fatal("client is still marked alive after server stop")
}
}
func TestTcpClientCloseIdempotent(t *testing.T) {
clientConn, peerConn := net.Pipe()
defer peerConn.Close()
client := toki.NewTcpClient(clientConn, 0, 0, testParserMap())
for i := 0; i < 3; i++ {
if err := client.Close(); err != nil && i == 0 {
t.Fatal(err)
}
}
if client.IsAlive() {
t.Fatal("client is still marked alive after close")
}
}
func TestTcpSendReceive(t *testing.T) {
port := freePort(t)
ctx, cancel := context.WithCancel(context.Background())

12
go/test/test_helpers_test.go
View file

@ -33,6 +33,18 @@ func freePort(t *testing.T) int {
return ln.Addr().(*net.TCPAddr).Port
}
func waitForCondition(t *testing.T, timeout time.Duration, condition func() bool, message string) {
t.Helper()
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
if condition() {
return
}
time.Sleep(time.Millisecond)
}
t.Fatal(message)
}
// generateSelfSignedCert creates an in-memory self-signed certificate valid
// for 127.0.0.1. Returns the server tls.Config and a client tls.Config that
// trusts the generated certificate.

44
go/test/tls_test.go
View file

@ -6,6 +6,8 @@ import (
"testing"
"time"
"nhooyr.io/websocket"
toki "toki-labs.com/toki_socket/go"
"toki-labs.com/toki_socket/go/packets"
)
@ -51,3 +53,45 @@ func TestTLSTcp(t *testing.T) {
t.Fatalf("unexpected response: %v", res)
}
}
func TestTLSWs(t *testing.T) {
port := freePort(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
serverTLS, clientTLS := generateSelfSignedCert(t)
server := toki.NewWsServerTLS("127.0.0.1", port, "/", serverTLS, func(conn *websocket.Conn) *toki.WsClient {
return toki.NewWsClient(conn, 0, 0, testParserMap())
})
server.OnClientConnected = func(client *toki.WsClient) {
toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
return &packets.TestData{
Index: req.GetIndex() * 2,
Message: "wss-echo: " + req.GetMessage(),
}, nil
})
}
if err := server.Start(ctx); err != nil {
t.Fatal(err)
}
defer server.Stop()
client, err := toki.DialWssWithHeartbeat(ctx, "127.0.0.1", port, "/", clientTLS, 0, 0, testParserMap())
if err != nil {
t.Fatal(err)
}
defer client.Close()
res, err := toki.SendRequestTyped[*packets.TestData, *packets.TestData](
&client.Communicator,
&packets.TestData{Index: 7, Message: "hello wss"},
2*time.Second,
)
if err != nil {
t.Fatal(err)
}
if res.GetIndex() != 14 || res.GetMessage() != "wss-echo: hello wss" {
t.Fatalf("unexpected response: %v", res)
}
}

132
go/test/ws_test.go
View file

@ -50,3 +50,135 @@ func TestWsRequestResponse(t *testing.T) {
t.Fatalf("unexpected response: %v", res)
}
}
func TestWsSendReceive(t *testing.T) {
port := freePort(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
received := make(chan *packets.TestData, 1)
server := toki.NewWsServer("127.0.0.1", port, "/", func(conn *websocket.Conn) *toki.WsClient {
return toki.NewWsClient(conn, 0, 0, testParserMap())
})
server.OnClientConnected = func(client *toki.WsClient) {
toki.AddListenerTyped[*packets.TestData](&client.Communicator, func(m *packets.TestData) {
received <- m
})
}
if err := server.Start(ctx); err != nil {
t.Fatal(err)
}
client, err := toki.DialWsWithHeartbeat(ctx, "127.0.0.1", port, "/", 0, 0, testParserMap())
if err != nil {
t.Fatal(err)
}
defer client.Close()
if err := client.Send(&packets.TestData{Index: 42, Message: "hello toki-socket ws"}); err != nil {
t.Fatal(err)
}
select {
case msg := <-received:
if msg.GetIndex() != 42 || msg.GetMessage() != "hello toki-socket ws" {
t.Fatalf("unexpected message: %v", msg)
}
case <-time.After(2 * time.Second):
t.Fatal("receive timed out")
}
}
func TestWsBroadcast(t *testing.T) {
port := freePort(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
server := toki.NewWsServer("127.0.0.1", port, "/", func(conn *websocket.Conn) *toki.WsClient {
return toki.NewWsClient(conn, 0, 0, testParserMap())
})
if err := server.Start(ctx); err != nil {
t.Fatal(err)
}
defer server.Stop()
client1, err := toki.DialWsWithHeartbeat(ctx, "127.0.0.1", port, "/", 0, 0, testParserMap())
if err != nil {
t.Fatal(err)
}
defer client1.Close()
client2, err := toki.DialWsWithHeartbeat(ctx, "127.0.0.1", port, "/", 0, 0, testParserMap())
if err != nil {
t.Fatal(err)
}
defer client2.Close()
received1 := make(chan *packets.TestData, 1)
received2 := make(chan *packets.TestData, 1)
toki.AddListenerTyped[*packets.TestData](&client1.Communicator, func(m *packets.TestData) {
received1 <- m
})
toki.AddListenerTyped[*packets.TestData](&client2.Communicator, func(m *packets.TestData) {
received2 <- m
})
waitForCondition(t, time.Second, func() bool {
return len(server.Clients()) == 2
}, "clients did not connect")
if err := server.Broadcast(&packets.TestData{Index: 9, Message: "ws broadcast"}); err != nil {
t.Fatal(err)
}
for name, ch := range map[string]chan *packets.TestData{
"client1": received1,
"client2": received2,
} {
select {
case msg := <-ch:
if msg.GetMessage() != "ws broadcast" {
t.Fatalf("%s got unexpected broadcast: %v", name, msg)
}
case <-time.After(2 * time.Second):
t.Fatalf("%s broadcast timed out", name)
}
}
}
func TestWsServerStopDisconnectsClients(t *testing.T) {
port := freePort(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
server := toki.NewWsServer("127.0.0.1", port, "/", func(conn *websocket.Conn) *toki.WsClient {
return toki.NewWsClient(conn, 0, 0, testParserMap())
})
if err := server.Start(ctx); err != nil {
t.Fatal(err)
}
client, err := toki.DialWsWithHeartbeat(ctx, "127.0.0.1", port, "/", 0, 0, testParserMap())
if err != nil {
t.Fatal(err)
}
defer client.Close()
disconnected := make(chan struct{}, 1)
client.AddDisconnectListener(func(*toki.WsClient) {
disconnected <- struct{}{}
})
waitForCondition(t, time.Second, func() bool {
return len(server.Clients()) == 1
}, "client did not connect")
if err := server.Stop(); err != nil {
t.Fatal(err)
}
select {
case <-disconnected:
case <-time.After(2 * time.Second):
t.Fatal("client disconnect callback timed out")
}
if client.IsAlive() {
t.Fatal("client is still marked alive after server stop")
}
}