diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 6401d66..2830031 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -259,7 +259,6 @@ class Engine extends Disposable with EventsEmittable { _reconnectAttempts++; try { - // isReconnecting = true; await signalClient.reconnect( url, token, @@ -272,7 +271,6 @@ class Engine extends Disposable with EventsEmittable { subscriber!.restartingIce = true; - // await negotiate(iceRestart: true); if (_hasPublished) { logger.fine('Reconnect: negotiating publisher...'); await publisher!.createAndSendOffer(const RTCOfferOptions( @@ -604,8 +602,7 @@ class Engine extends Disposable with EventsEmittable { ..on((event) => events.emit(event)) ..on((event) async { if (connectionState == ConnectionState.reconnecting) { - logger.fine('Ignoring leave signal since engine is reconnecting...'); - return; + logger.warning('Received leave signal while engine is reconnecting.'); } await close(); events.emit(const EngineDisconnectedEvent()); diff --git a/lib/src/support/websocket.dart b/lib/src/support/websocket.dart index 525d974..cf94e6d 100644 --- a/lib/src/support/websocket.dart +++ b/lib/src/support/websocket.dart @@ -1,3 +1,4 @@ +import '../support/disposable.dart'; import 'websocket/io.dart' if (dart.library.html) 'websocket/web.dart'; class WebSocketException implements Exception { @@ -29,9 +30,8 @@ class WebSocketEventHandlers { }); } -abstract class LiveKitWebSocket { +abstract class LiveKitWebSocket extends Disposable { void send(List data); - Future dispose(); static Future connect( Uri uri, [ diff --git a/lib/src/support/websocket/io.dart b/lib/src/support/websocket/io.dart index 65102ba..910ba26 100644 --- a/lib/src/support/websocket/io.dart +++ b/lib/src/support/websocket/io.dart @@ -11,7 +11,7 @@ Future lkWebSocketConnect( ]) => LiveKitWebSocketIO.connect(uri, options); -class LiveKitWebSocketIO implements LiveKitWebSocket { +class LiveKitWebSocketIO extends LiveKitWebSocket { final io.WebSocket _ws; final WebSocketEventHandlers? options; late final StreamSubscription _subscription; @@ -21,19 +21,24 @@ class LiveKitWebSocketIO implements LiveKitWebSocket { this.options, ]) { _subscription = _ws.listen( - (dynamic data) => options?.onData?.call(data), + (dynamic data) { + if (isDisposed) { + logger.warning('$objectId already disposed, ignoring received data.'); + return; + } + options?.onData?.call(data); + }, onDone: () async { await _subscription.cancel(); options?.onDispose?.call(); }, ); - } - @override - Future dispose() async { - if (_ws.readyState != io.WebSocket.closed) { - await _ws.close(); - } + onDispose(() async { + if (_ws.readyState != io.WebSocket.closed) { + await _ws.close(); + } + }); } @override diff --git a/lib/src/support/websocket/web.dart b/lib/src/support/websocket/web.dart index dce987d..33aa2a6 100644 --- a/lib/src/support/websocket/web.dart +++ b/lib/src/support/websocket/web.dart @@ -2,6 +2,8 @@ import 'dart:async'; import 'dart:html' as html; import 'dart:typed_data'; +import '../../extensions.dart'; +import '../../logger.dart'; import '../websocket.dart'; // ignore: avoid_web_libraries_in_flutter @@ -12,7 +14,7 @@ Future lkWebSocketConnect( ]) => LiveKitWebSocketWeb.connect(uri, options); -class LiveKitWebSocketWeb implements LiveKitWebSocket { +class LiveKitWebSocketWeb extends LiveKitWebSocket { final html.WebSocket _ws; final WebSocketEventHandlers? options; late final StreamSubscription _messageSubscription; @@ -24,6 +26,10 @@ class LiveKitWebSocketWeb implements LiveKitWebSocket { ]) { _ws.binaryType = 'arraybuffer'; _messageSubscription = _ws.onMessage.listen((_) { + if (isDisposed) { + logger.warning('$objectId already disposed, ignoring received data.'); + return; + } dynamic _data = _.data is ByteBuffer ? _.data.asUint8List() : _.data; options?.onData?.call(_data); }); @@ -32,18 +38,17 @@ class LiveKitWebSocketWeb implements LiveKitWebSocket { await _closeSubscription.cancel(); options?.onDispose?.call(); }); + + onDispose(() async { + if (_ws.readyState != html.WebSocket.CLOSED) { + _ws.close(); + } + }); } @override void send(List data) => _ws.send(data); - @override - Future dispose() async { - if (_ws.readyState != html.WebSocket.CLOSED) { - _ws.close(); - } - } - static Future connect( Uri uri, [ WebSocketEventHandlers? options,