diff --git a/dart/lib/src/communicator.dart b/dart/lib/src/communicator.dart index 8453ae7..e623b39 100644 --- a/dart/lib/src/communicator.dart +++ b/dart/lib/src/communicator.dart @@ -42,7 +42,7 @@ abstract class Communicator { return write; } - Future send(T data); + Future send(T data); /// Sends [data] as a request and waits for a typed response. /// diff --git a/dart/lib/src/protobuf_client.dart b/dart/lib/src/protobuf_client.dart index 16b8c83..d8127aa 100644 --- a/dart/lib/src/protobuf_client.dart +++ b/dart/lib/src/protobuf_client.dart @@ -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.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.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 data) async { + Future send(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 } } } diff --git a/dart/lib/src/protobuf_server.dart b/dart/lib/src/protobuf_server.dart index f8153a9..e5a329b 100644 --- a/dart/lib/src/protobuf_server.dart +++ b/dart/lib/src/protobuf_server.dart @@ -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 { diff --git a/dart/lib/src/response_checker.dart b/dart/lib/src/response_checker.dart index a07e985..f2bcb82 100644 --- a/dart/lib/src/response_checker.dart +++ b/dart/lib/src/response_checker.dart @@ -3,8 +3,8 @@ import 'dart:async'; class ResponseChecker { 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; diff --git a/dart/lib/src/ws_protobuf_client.dart b/dart/lib/src/ws_protobuf_client.dart index 149a715..005a87f 100644 --- a/dart/lib/src/ws_protobuf_client.dart +++ b/dart/lib/src/ws_protobuf_client.dart @@ -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 data) async { + Future send(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 } } } diff --git a/dart/test/communicator_test.dart b/dart/test/communicator_test.dart index 8346145..e58f0ee 100644 --- a/dart/test/communicator_test.dart +++ b/dart/test/communicator_test.dart @@ -19,7 +19,7 @@ class _FakeCommunicator extends Communicator { } @override - Future send(T data) async { + Future send(T data) async { if (isAlive) { await queuePacket(PacketBase() ..typeName = data.info_.qualifiedMessageName diff --git a/go/base_client.go b/go/base_client.go new file mode 100644 index 0000000..c88def3 --- /dev/null +++ b/go/base_client.go @@ -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) + } +} diff --git a/go/tcp_client.go b/go/tcp_client.go index 01fc2d9..d6d17a9 100644 --- a/go/tcp_client.go +++ b/go/tcp_client.go @@ -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) diff --git a/go/ws_client.go b/go/ws_client.go index ebd0a6b..4bb94c4 100644 --- a/go/ws_client.go +++ b/go/ws_client.go @@ -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) }