ignore new data on disposed socket

Author: hiroshihorie

lib/src/core/engine.dart

@@ -259,7 +259,6 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
     _reconnectAttempts++;
 
     try {
-      // isReconnecting = true;
       await signalClient.reconnect(
         url,
         token,
@@ -272,7 +271,6 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
 
       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<EngineEvent> {
     ..on<SignalStreamStateUpdatedEvent>((event) => events.emit(event))
     ..on<SignalLeaveEvent>((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());

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<int> data);
-  Future<void> dispose();
 
   static Future<LiveKitWebSocket> connect(
     Uri uri, [

lib/src/support/websocket/io.dart

@@ -11,7 +11,7 @@ Future<LiveKitWebSocketIO> 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<void> dispose() async {
-    if (_ws.readyState != io.WebSocket.closed) {
-      await _ws.close();
-    }
+    onDispose(() async {
+      if (_ws.readyState != io.WebSocket.closed) {
+        await _ws.close();
+      }
+    });
   }
 
   @override

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<LiveKitWebSocketWeb> 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<int> data) => _ws.send(data);
 
-  @override
-  Future<void> dispose() async {
-    if (_ws.readyState != html.WebSocket.CLOSED) {
-      _ws.close();
-    }
-  }
-
   static Future<LiveKitWebSocketWeb> connect(
     Uri uri, [
     WebSocketEventHandlers? options,