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

Refactor client implementations with baseClient pattern

Browse source 
- Dart: Add explicit Future<void> return type to send() method
- Dart: Improve onDisconnected() logic with proper isAlive handling and heartbeat response
- Go: Introduce baseClient generic base class for shared client logic
- Go: Refactor TcpClient and WsClient to embed baseClient instead of embedding Communicator
- Remove duplicate heartbeat and disconnect handling code across client implementations
- Clean up unused imports (time package)
This commit is contained in:
toki 2026-04-11 13:52:50 +09:00
parent e48722d811
commit cab031ea28
9 changed files with 206 additions and 259 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -42,7 +42,7 @@ abstract class Communicator {
return write;
}
Future send<T extends GeneratedMessage>(T data);
Future<void> send<T extends GeneratedMessage>(T data);
/// Sends [data] as a request and waits for a typed response.
///

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

@ -88,12 +88,15 @@ abstract class ProtobufClient extends Communicator {
}
void onDisconnected(dynamic data) {
if (isAlive) {
for (var item in _onDisconnectListenerList) {
item.call(this);
}
_onDisconnectListenerList.clear();
if (!isAlive) {
return;
}
isAlive = false;
_heartbeatChecker?.responded();
for (var item in _onDisconnectListenerList) {
item.call(this);
}
_onDisconnectListenerList.clear();
}
void onError(dynamic e) {
@ -110,31 +113,35 @@ abstract class ProtobufClient extends Communicator {
}
void _parsing() {
if (_length == null) {
if (_arrivedData.length >= _headerSize) {
_length = Uint8List.fromList(_arrivedData.sublist(0, _headerSize))
while (true) {
if (_length == null) {
if (_arrivedData.length < _headerSize) {
return;
}
final length = Uint8List.fromList(_arrivedData.sublist(0, _headerSize))
.buffer
.asByteData()
.getInt32(0);
if (_length == 0) {
if (length == 0) {
_length = null;
_arrivedData.clear();
return;
}
_parsing();
_length = length;
}
} else {
if (_arrivedData.length >= _length! + _headerSize) {
final packetBytes = List<int>.from(
_arrivedData.sublist(_headerSize, _headerSize + _length!));
_arrivedData = _arrivedData.sublist(_headerSize + _length!);
_length = null;
final common = PacketBase.fromBuffer(packetBytes);
onReceivedData(common.typeName, common.data,
incomingNonce: common.nonce, responseNonce: common.responseNonce);
sendHeartBeat();
_parsing();
if (_arrivedData.length < _length! + _headerSize) {
return;
}
final packetBytes = List<int>.from(
_arrivedData.sublist(_headerSize, _headerSize + _length!));
_arrivedData = _arrivedData.sublist(_headerSize + _length!);
_length = null;
final common = PacketBase.fromBuffer(packetBytes);
onReceivedData(common.typeName, common.data,
incomingNonce: common.nonce, responseNonce: common.responseNonce);
sendHeartBeat();
}
}
@ -148,7 +155,7 @@ abstract class ProtobufClient extends Communicator {
}
@override
Future send<T extends GeneratedMessage>(T data) async {
Future<void> send<T extends GeneratedMessage>(T data) async {
if (isAlive) {
try {
await queuePacket(PacketBase()
@ -166,12 +173,12 @@ abstract class ProtobufClient extends Communicator {
if (isAlive) {
isAlive = false;
_heartbeatChecker?.responded();
try {
await _socket.close();
_socket.destroy();
} catch (_) {
// already closed
}
}
try {
await _socket.close();
_socket.destroy();
} catch (_) {
// already closed
}
}
}

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

@ -31,13 +31,15 @@ abstract class ProtobufServer {
/// ..usePrivateKey('server.key');
/// final server = MyServer.secure('0.0.0.0', 9090, ctx, (s) => MyClient(s));
/// ```
ProtobufServer.secure(
this._host, this._port, SecurityContext securityContext, this._createNewClient)
ProtobufServer.secure(this._host, this._port, SecurityContext securityContext,
this._createNewClient)
: _securityContext = securityContext;
Future start() async {
if (_securityContext != null) {
_secureServer = await SecureServerSocket.bind(_host, _port, _securityContext!);
final securityContext = _securityContext;
if (securityContext != null) {
_secureServer =
await SecureServerSocket.bind(_host, _port, securityContext);
_started = true;
_secureServer?.listen(_onClientSocket);
} else {

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

@ -3,8 +3,8 @@ import 'dart:async';
class ResponseChecker<T> {
final T _responser;
late int _time;
late Timer? _timer = null;
late void Function(T) _notResponseListener;
Timer? _timer;
final void Function(T) _notResponseListener;
T get responser => _responser;

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

@ -86,12 +86,15 @@ abstract class WsProtobufClient extends Communicator {
}
void onDisconnected(dynamic data) {
if (isAlive) {
for (var item in _onDisconnectListenerList) {
item.call(this);
}
_onDisconnectListenerList.clear();
if (!isAlive) {
return;
}
isAlive = false;
_heartbeatChecker?.responded();
for (var item in _onDisconnectListenerList) {
item.call(this);
}
_onDisconnectListenerList.clear();
}
void onError(dynamic e) {
@ -112,7 +115,7 @@ abstract class WsProtobufClient extends Communicator {
}
@override
Future send<T extends GeneratedMessage>(T data) async {
Future<void> send<T extends GeneratedMessage>(T data) async {
if (isAlive) {
try {
await queuePacket(PacketBase()
@ -130,11 +133,11 @@ abstract class WsProtobufClient extends Communicator {
if (isAlive) {
isAlive = false;
_heartbeatChecker?.responded();
try {
await _ws.close();
} catch (_) {
// already closed by peer
}
}
try {
await _ws.close();
} catch (_) {
// already closed by peer
}
}
}

2
dart/test/communicator_test.dart
View file

@ -19,7 +19,7 @@ class _FakeCommunicator extends Communicator {
}
@override
Future send<T extends GeneratedMessage>(T data) async {
Future<void> send<T extends GeneratedMessage>(T data) async {
if (isAlive) {
await queuePacket(PacketBase()
..typeName = data.info_.qualifiedMessageName

121
go/base_client.go Normal file
View file

@ -0,0 +1,121 @@
package toki_socket
import (
"sync"
"time"
"toki-labs.com/toki_socket/go/packets"
)
type baseClient[Self any] struct {
Communicator
heartbeatInterval time.Duration
heartbeatWait time.Duration
waitingHBResponse bool
hbTimer *HeartbeatTimer
hbMu sync.Mutex
disconnectListeners []func(Self)
disconnectMu sync.Mutex
connCloseOnce sync.Once
self Self
doClose func() error
}
func newBaseClient[Self any](self Self, intervalSec, waitSec int, doClose func() error) baseClient[Self] {
return baseClient[Self]{
heartbeatInterval: time.Duration(intervalSec) * time.Second,
heartbeatWait: time.Duration(waitSec) * time.Second,
self: self,
doClose: doClose,
}
}
func (c *baseClient[Self]) AddDisconnectListener(handler func(Self)) {
c.disconnectMu.Lock()
defer c.disconnectMu.Unlock()
c.disconnectListeners = append(c.disconnectListeners, handler)
}
func (c *baseClient[Self]) RemoveDisconnectListeners() {
c.disconnectMu.Lock()
defer c.disconnectMu.Unlock()
c.disconnectListeners = nil
}
func (c *baseClient[Self]) Close() error {
var err error
c.connCloseOnce.Do(func() {
c.Communicator.shutdown()
c.stopHeartbeat()
if c.doClose != nil {
err = c.doClose()
}
c.notifyDisconnected()
})
return err
}
func (c *baseClient[Self]) sendHeartBeat() {
if !c.IsAlive() || c.heartbeatInterval <= 0 {
return
}
c.hbMu.Lock()
if c.hbTimer != nil {
c.hbTimer.Stop()
}
c.hbTimer = NewHeartbeatTimer(c.heartbeatInterval, func() {
if !c.IsAlive() {
return
}
c.hbMu.Lock()
c.waitingHBResponse = true
c.hbMu.Unlock()
_ = c.Send(&packets.HeartBeat{})
c.hbMu.Lock()
if c.hbTimer != nil {
c.hbTimer.Stop()
}
c.hbTimer = NewHeartbeatTimer(c.heartbeatWait, func() {
if c.IsAlive() {
c.onDisconnected()
}
})
c.hbMu.Unlock()
})
c.hbMu.Unlock()
}
func (c *baseClient[Self]) onHeartBeat() {
c.hbMu.Lock()
if c.waitingHBResponse {
c.waitingHBResponse = false
c.hbMu.Unlock()
return
}
c.hbMu.Unlock()
_ = c.Send(&packets.HeartBeat{})
}
func (c *baseClient[Self]) stopHeartbeat() {
c.hbMu.Lock()
defer c.hbMu.Unlock()
if c.hbTimer != nil {
c.hbTimer.Stop()
c.hbTimer = nil
}
}
func (c *baseClient[Self]) onDisconnected() {
_ = c.Close()
}
func (c *baseClient[Self]) notifyDisconnected() {
c.disconnectMu.Lock()
listeners := append([]func(Self){}, c.disconnectListeners...)
c.disconnectListeners = nil
c.disconnectMu.Unlock()
for _, listener := range listeners {
listener(c.self)
}
}

110
go/tcp_client.go
View file

@ -8,7 +8,6 @@ import (
"io"
"net"
"sync"
"time"
"google.golang.org/protobuf/proto"
@ -18,25 +17,18 @@ import (
const MaxPacketSize = 64 << 20
type TcpClient struct {
Communicator
conn net.Conn
writeMu sync.Mutex // Defensive if WritePacket is called outside the queued writer.
heartbeatInterval time.Duration
heartbeatWait time.Duration
waitingHBResponse bool
hbTimer *HeartbeatTimer
hbMu sync.Mutex
disconnectListeners []func(*TcpClient)
disconnectMu sync.Mutex
tcpCloseOnce sync.Once
baseClient[*TcpClient]
conn net.Conn
writeMu sync.Mutex // Defensive if WritePacket is called outside the queued writer.
}
func NewTcpClient(conn net.Conn, intervalSec, waitSec int, parserMap ParserMap) *TcpClient {
c := &TcpClient{
conn: conn,
heartbeatInterval: time.Duration(intervalSec) * time.Second,
heartbeatWait: time.Duration(waitSec) * time.Second,
conn: conn,
}
c.baseClient = newBaseClient(c, intervalSec, waitSec, func() error {
return c.conn.Close()
})
c.Communicator.Initialize(c, parserMap)
c.SetWriteErrorHandler(func(error) {
c.onDisconnected()
@ -113,94 +105,6 @@ func (c *TcpClient) WritePacket(base *packets.PacketBase) error {
return writeFull(c.conn, b)
}
func (c *TcpClient) AddDisconnectListener(handler func(*TcpClient)) {
c.disconnectMu.Lock()
defer c.disconnectMu.Unlock()
c.disconnectListeners = append(c.disconnectListeners, handler)
}
func (c *TcpClient) RemoveDisconnectListeners() {
c.disconnectMu.Lock()
defer c.disconnectMu.Unlock()
c.disconnectListeners = nil
}
func (c *TcpClient) Close() error {
var err error
c.tcpCloseOnce.Do(func() {
c.Communicator.shutdown()
c.stopHeartbeat()
err = c.conn.Close()
c.notifyDisconnected()
})
return err
}
func (c *TcpClient) sendHeartBeat() {
if !c.IsAlive() || c.heartbeatInterval <= 0 {
return
}
c.hbMu.Lock()
if c.hbTimer != nil {
c.hbTimer.Stop()
}
c.hbTimer = NewHeartbeatTimer(c.heartbeatInterval, func() {
if !c.IsAlive() {
return
}
c.hbMu.Lock()
c.waitingHBResponse = true
c.hbMu.Unlock()
_ = c.Send(&packets.HeartBeat{})
c.hbMu.Lock()
if c.hbTimer != nil {
c.hbTimer.Stop()
}
c.hbTimer = NewHeartbeatTimer(c.heartbeatWait, func() {
if c.IsAlive() {
c.onDisconnected()
}
})
c.hbMu.Unlock()
})
c.hbMu.Unlock()
}
func (c *TcpClient) onHeartBeat() {
c.hbMu.Lock()
if c.waitingHBResponse {
c.waitingHBResponse = false
c.hbMu.Unlock()
return
}
c.hbMu.Unlock()
_ = c.Send(&packets.HeartBeat{})
}
func (c *TcpClient) stopHeartbeat() {
c.hbMu.Lock()
defer c.hbMu.Unlock()
if c.hbTimer != nil {
c.hbTimer.Stop()
c.hbTimer = nil
}
}
func (c *TcpClient) onDisconnected() {
_ = c.Close()
}
func (c *TcpClient) notifyDisconnected() {
c.disconnectMu.Lock()
listeners := append([]func(*TcpClient){}, c.disconnectListeners...)
c.disconnectListeners = nil
c.disconnectMu.Unlock()
for _, listener := range listeners {
listener(c)
}
}
func writeFull(w io.Writer, b []byte) error {
for len(b) > 0 {
n, err := w.Write(b)

128
go/ws_client.go
View file

@ -6,7 +6,6 @@ import (
"fmt"
"net/http"
"sync"
"time"
"google.golang.org/protobuf/proto"
"nhooyr.io/websocket"
@ -20,30 +19,30 @@ const (
)
type WsClient struct {
Communicator
conn *websocket.Conn
writeMu sync.Mutex // Defensive if WritePacket is called outside the queued writer.
readCtx context.Context
cancelRead context.CancelFunc
heartbeatInterval time.Duration
heartbeatWait time.Duration
waitingHBResponse bool
hbTimer *HeartbeatTimer
hbMu sync.Mutex
disconnectListeners []func(*WsClient)
disconnectMu sync.Mutex
wsCloseOnce sync.Once
baseClient[*WsClient]
conn *websocket.Conn
writeMu sync.Mutex // Defensive if WritePacket is called outside the queued writer.
readCtx context.Context
cancelRead context.CancelFunc
writeCtx context.Context
cancelWrite context.CancelFunc
}
func NewWsClient(conn *websocket.Conn, intervalSec, waitSec int, parserMap ParserMap) *WsClient {
readCtx, cancelRead := context.WithCancel(context.Background())
writeCtx, cancelWrite := context.WithCancel(context.Background())
c := &WsClient{
conn: conn,
readCtx: readCtx,
cancelRead: cancelRead,
heartbeatInterval: time.Duration(intervalSec) * time.Second,
heartbeatWait: time.Duration(waitSec) * time.Second,
conn: conn,
readCtx: readCtx,
cancelRead: cancelRead,
writeCtx: writeCtx,
cancelWrite: cancelWrite,
}
c.baseClient = newBaseClient(c, intervalSec, waitSec, func() error {
c.cancelRead()
c.cancelWrite()
return c.conn.Close(websocket.StatusNormalClosure, "")
})
c.Communicator.Initialize(c, parserMap)
c.SetWriteErrorHandler(func(error) {
c.onDisconnected()
@ -113,94 +112,5 @@ func (c *WsClient) WritePacket(base *packets.PacketBase) error {
}
c.writeMu.Lock()
defer c.writeMu.Unlock()
return c.conn.Write(context.Background(), websocket.MessageBinary, b)
}
func (c *WsClient) AddDisconnectListener(handler func(*WsClient)) {
c.disconnectMu.Lock()
defer c.disconnectMu.Unlock()
c.disconnectListeners = append(c.disconnectListeners, handler)
}
func (c *WsClient) RemoveDisconnectListeners() {
c.disconnectMu.Lock()
defer c.disconnectMu.Unlock()
c.disconnectListeners = nil
}
func (c *WsClient) Close() error {
var err error
c.wsCloseOnce.Do(func() {
c.Communicator.shutdown()
c.stopHeartbeat()
c.cancelRead()
err = c.conn.Close(websocket.StatusNormalClosure, "")
c.notifyDisconnected()
})
return err
}
func (c *WsClient) sendHeartBeat() {
if !c.IsAlive() || c.heartbeatInterval <= 0 {
return
}
c.hbMu.Lock()
if c.hbTimer != nil {
c.hbTimer.Stop()
}
c.hbTimer = NewHeartbeatTimer(c.heartbeatInterval, func() {
if !c.IsAlive() {
return
}
c.hbMu.Lock()
c.waitingHBResponse = true
c.hbMu.Unlock()
_ = c.Send(&packets.HeartBeat{})
c.hbMu.Lock()
if c.hbTimer != nil {
c.hbTimer.Stop()
}
c.hbTimer = NewHeartbeatTimer(c.heartbeatWait, func() {
if c.IsAlive() {
c.onDisconnected()
}
})
c.hbMu.Unlock()
})
c.hbMu.Unlock()
}
func (c *WsClient) onHeartBeat() {
c.hbMu.Lock()
if c.waitingHBResponse {
c.waitingHBResponse = false
c.hbMu.Unlock()
return
}
c.hbMu.Unlock()
_ = c.Send(&packets.HeartBeat{})
}
func (c *WsClient) stopHeartbeat() {
c.hbMu.Lock()
defer c.hbMu.Unlock()
if c.hbTimer != nil {
c.hbTimer.Stop()
c.hbTimer = nil
}
}
func (c *WsClient) onDisconnected() {
_ = c.Close()
}
func (c *WsClient) notifyDisconnected() {
c.disconnectMu.Lock()
listeners := append([]func(*WsClient){}, c.disconnectListeners...)
c.disconnectListeners = nil
c.disconnectMu.Unlock()
for _, listener := range listeners {
listener(c)
}
return c.conn.Write(c.writeCtx, websocket.MessageBinary, b)
}