From b1c5751759e9b4162cdf2d0faa603b2a84761d85 Mon Sep 17 00:00:00 2001 From: cloudwebrtc Date: Sat, 23 Dec 2023 00:08:37 +0800 Subject: [PATCH 1/8] try to fix issue #438. --- lib/src/core/engine.dart | 3 ++- lib/src/core/room.dart | 11 +++++++++++ lib/src/internal/events.dart | 5 +++++ 3 files changed, 18 insertions(+), 1 deletion(-) diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 5b881b6a8..dffe549b3 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -835,7 +835,7 @@ class Engine extends Disposable with EventsEmittable { } void _setUpEngineListeners() => - events.on((event) async { + events.on((event) async { // send queued requests if engine re-connected signalClient.sendQueuedRequests(); }); @@ -907,6 +907,7 @@ class Engine extends Disposable with EventsEmittable { }) ..on((event) async { logger.fine('Signal connecting'); + events.emit(const EngineConnectingEvent()); }) ..on((event) async { logger.fine('Signal reconnecting'); diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index 3d635a970..9f5359c0d 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -355,10 +355,17 @@ class Room extends DisposableChangeNotifier with EventsEmittable { }); void _setUpEngineListeners() => _engineListener + ..on((event) async { + notifyListeners(); + }) + ..on((event) async { + notifyListeners(); + }) ..on((event) async { events.emit(const RoomReconnectedEvent()); // re-send tracks permissions localParticipant?.sendTrackSubscriptionPermissions(); + notifyListeners(); }) ..on((event) async { events.emit(const RoomRestartingEvent()); @@ -379,6 +386,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { events.emit(ParticipantDisconnectedEvent(participant: participant)); await participant.dispose(); } + notifyListeners(); }) ..on((event) async { events.emit(const RoomRestartedEvent()); @@ -393,14 +401,17 @@ class Room extends DisposableChangeNotifier with EventsEmittable { } } } + notifyListeners(); }) ..on((event) async { events.emit(const RoomReconnectingEvent()); await _sendSyncState(); + notifyListeners(); }) ..on((event) async { await _cleanUp(); events.emit(RoomDisconnectedEvent(reason: event.reason)); + notifyListeners(); }) ..on( (event) => _onEngineActiveSpeakersUpdateEvent(event.speakers)) diff --git a/lib/src/internal/events.dart b/lib/src/internal/events.dart index 1bdbd62df..e93ef23ab 100644 --- a/lib/src/internal/events.dart +++ b/lib/src/internal/events.dart @@ -149,6 +149,11 @@ class SignalConnectivityChangedEvent with SignalEvent, InternalEvent { }); } +@internal +class EngineConnectingEvent with InternalEvent, EngineEvent { + const EngineConnectingEvent(); +} + @internal class EngineConnectedEvent with InternalEvent, SignalEvent, EngineEvent { const EngineConnectedEvent(); From 1379d0b76b5b4734f2f402333249318ced2019bf Mon Sep 17 00:00:00 2001 From: cloudwebrtc Date: Sat, 23 Dec 2023 01:40:53 +0800 Subject: [PATCH 2/8] Improve reconnection events, add RoomAttemptReconnectEvent. --- example/lib/pages/room.dart | 5 +++++ lib/src/core/engine.dart | 33 +++++++++++++++++++++------------ lib/src/core/room.dart | 22 +++++++++++++++++++--- lib/src/core/signal_client.dart | 3 +-- lib/src/events.dart | 15 +++++++++++++++ lib/src/internal/events.dart | 12 ++++++++++++ lib/src/types/other.dart | 3 ++- 7 files changed, 75 insertions(+), 18 deletions(-) diff --git a/example/lib/pages/room.dart b/example/lib/pages/room.dart index 90f9abf7a..e0137ee38 100644 --- a/example/lib/pages/room.dart +++ b/example/lib/pages/room.dart @@ -94,6 +94,11 @@ class _RoomPageState extends State { ..on((event) { context.showRecordingStatusChangedDialog(event.activeRecording); }) + ..on((event) { + print( + 'Attempting to reconnect ${event.attempt}/${event.maxAttemptsRetry}, ' + '(${event.nextRetryDelaysInMs}ms delay until next attempt)'); + }) ..on((_) => _sortParticipants()) ..on((_) => _sortParticipants()) ..on((_) => _sortParticipants()) diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index dffe549b3..8c4ce2f8f 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -228,6 +228,7 @@ class Engine extends Disposable with EventsEmittable { await signalClient.cleanUp(); fullReconnectOnNext = false; + attemptingReconnect = false; clearPendingReconnect(); } @@ -625,8 +626,25 @@ class Engine extends Disposable with EventsEmittable { reconnectStart = DateTime.now(); } + if (reconnectAttempts! >= _reconnectCount) { + logger.fine('reconnectAttempts exceeded, disconnecting...'); + _isClosed = true; + await cleanUp(); + + events.emit(EngineDisconnectedEvent( + reason: DisconnectReason.reconnectAttemptsExceeded, + )); + return; + } + var delay = defaultRetryDelaysInMs[reconnectAttempts!]; + events.emit(EngineAttemptReconnectEvent( + attempt: reconnectAttempts! + 1, + maxAttempts: _reconnectCount, + nextRetryDelaysInMs: delay, + )); + clearReconnectTimeout(); logger.fine( 'WebSocket reconnecting in $delay ms, retry times $reconnectAttempts'); @@ -656,15 +674,6 @@ class Engine extends Disposable with EventsEmittable { fullReconnectOnNext = true; } - if (reconnectAttempts! >= _reconnectCount) { - logger.fine('reconnectAttempts exceeded, disconnecting...'); - events.emit(EngineDisconnectedEvent( - reason: DisconnectReason.connectionClosed, - )); - await cleanUp(); - return; - } - try { attemptingReconnect = true; @@ -704,7 +713,7 @@ class Engine extends Disposable with EventsEmittable { } else { logger.fine('attemptReconnect: disconnecting...'); events.emit(EngineDisconnectedEvent( - reason: DisconnectReason.connectionClosed, + reason: DisconnectReason.disconnected, )); await cleanUp(); } @@ -718,7 +727,7 @@ class Engine extends Disposable with EventsEmittable { return; } - events.emit(const EngineResumingEvent()); + events.emit(const EngineReconnectingEvent()); // wait for socket to connect rtc server await signalClient.connect( @@ -915,7 +924,7 @@ class Engine extends Disposable with EventsEmittable { }) ..on((event) async { logger.fine('Signal disconnected ${event.reason}'); - if (event.reason == DisconnectReason.connectionClosed && !_isClosed) { + if (event.reason == DisconnectReason.disconnected && !_isClosed) { await handleDisconnect(ClientDisconnectReason.signal); } else { events.emit(EngineDisconnectedEvent( diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index 9f5359c0d..e3b4a3e69 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -13,6 +13,7 @@ // limitations under the License. import 'dart:async'; +import 'dart:math'; import 'package:flutter/foundation.dart'; @@ -408,11 +409,26 @@ class Room extends DisposableChangeNotifier with EventsEmittable { await _sendSyncState(); notifyListeners(); }) - ..on((event) async { - await _cleanUp(); - events.emit(RoomDisconnectedEvent(reason: event.reason)); + ..on((event) async { + events.emit(RoomAttemptReconnectEvent( + attempt: event.attempt, + maxAttemptsRetry: event.maxAttempts, + nextRetryDelaysInMs: event.nextRetryDelaysInMs, + )); notifyListeners(); }) + ..on((event) async { + if (!engine.fullReconnectOnNext && + ![ + DisconnectReason.signalingConnectionFailure, + DisconnectReason.joinFailure, + DisconnectReason.noInternetConnection + ].contains(event.reason)) { + await _cleanUp(); + events.emit(RoomDisconnectedEvent(reason: event.reason)); + notifyListeners(); + } + }) ..on( (event) => _onEngineActiveSpeakersUpdateEvent(event.speakers)) ..on(_onDataMessageEvent) diff --git a/lib/src/core/signal_client.dart b/lib/src/core/signal_client.dart index a492141a7..10536ca2e 100644 --- a/lib/src/core/signal_client.dart +++ b/lib/src/core/signal_client.dart @@ -340,8 +340,7 @@ class SignalClient extends Disposable with EventsEmittable { return; } _connectionState = ConnectionState.disconnected; - events.emit( - SignalDisconnectedEvent(reason: DisconnectReason.connectionClosed)); + events.emit(SignalDisconnectedEvent(reason: DisconnectReason.disconnected)); } void _sendPing() { diff --git a/lib/src/events.dart b/lib/src/events.dart index e2b25d0b0..f18664143 100644 --- a/lib/src/events.dart +++ b/lib/src/events.dart @@ -66,6 +66,21 @@ class RoomReconnectingEvent with RoomEvent { String toString() => '${runtimeType}()'; } +/// report the number of attempts to reconnect to the room. +class RoomAttemptReconnectEvent with RoomEvent { + final int attempt; + final int maxAttemptsRetry; + final int nextRetryDelaysInMs; + const RoomAttemptReconnectEvent({ + required this.attempt, + required this.maxAttemptsRetry, + required this.nextRetryDelaysInMs, + }); + + @override + String toString() => '${runtimeType}()'; +} + /// Connection to room is re-established. All existing state is preserved. /// Emitted by [Room]. class RoomReconnectedEvent with RoomEvent { diff --git a/lib/src/internal/events.dart b/lib/src/internal/events.dart index e93ef23ab..5e6a34b51 100644 --- a/lib/src/internal/events.dart +++ b/lib/src/internal/events.dart @@ -172,6 +172,18 @@ class EngineFullRestartingEvent with InternalEvent, EngineEvent { const EngineFullRestartingEvent(); } +@internal +class EngineAttemptReconnectEvent with InternalEvent, EngineEvent { + int attempt; + int maxAttempts; + int nextRetryDelaysInMs; + EngineAttemptReconnectEvent({ + required this.attempt, + required this.maxAttempts, + required this.nextRetryDelaysInMs, + }); +} + @internal class EngineRestartedEvent with InternalEvent, EngineEvent { const EngineRestartedEvent(); diff --git a/lib/src/types/other.dart b/lib/src/types/other.dart index 615d331b5..28ea642da 100644 --- a/lib/src/types/other.dart +++ b/lib/src/types/other.dart @@ -84,9 +84,10 @@ enum DisconnectReason { roomDeleted, stateMismatch, joinFailure, - connectionClosed, + disconnected, signalingConnectionFailure, noInternetConnection, + reconnectAttemptsExceeded, } /// The reason why a track failed to publish. From a421293568bdb7ed91504b2456dba0a88421a1da Mon Sep 17 00:00:00 2001 From: cloudwebrtc Date: Sat, 23 Dec 2023 01:52:22 +0800 Subject: [PATCH 3/8] update. --- lib/src/core/room.dart | 1 - 1 file changed, 1 deletion(-) diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index e3b4a3e69..c5a24abdf 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -13,7 +13,6 @@ // limitations under the License. import 'dart:async'; -import 'dart:math'; import 'package:flutter/foundation.dart'; From df85be2db52e3268e2095756c3c2c8b6aa0aad08 Mon Sep 17 00:00:00 2001 From: cloudwebrtc Date: Sat, 23 Dec 2023 01:59:01 +0800 Subject: [PATCH 4/8] cleanup. --- lib/src/core/room.dart | 6 ------ 1 file changed, 6 deletions(-) diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index c5a24abdf..2c18d5c2d 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -355,12 +355,6 @@ class Room extends DisposableChangeNotifier with EventsEmittable { }); void _setUpEngineListeners() => _engineListener - ..on((event) async { - notifyListeners(); - }) - ..on((event) async { - notifyListeners(); - }) ..on((event) async { events.emit(const RoomReconnectedEvent()); // re-send tracks permissions From 88c6526d789c076e9dda30a34e9fcc03248a3e7b Mon Sep 17 00:00:00 2001 From: cloudwebrtc Date: Sat, 23 Dec 2023 02:02:05 +0800 Subject: [PATCH 5/8] update. --- lib/src/core/engine.dart | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 8c4ce2f8f..aab234f53 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -727,7 +727,7 @@ class Engine extends Disposable with EventsEmittable { return; } - events.emit(const EngineReconnectingEvent()); + events.emit(const EngineResumingEvent()); // wait for socket to connect rtc server await signalClient.connect( From 24c3702de356d54befa126abd805392840f53f8a Mon Sep 17 00:00:00 2001 From: cloudwebrtc Date: Sun, 24 Dec 2023 00:00:59 +0800 Subject: [PATCH 6/8] update. --- lib/src/core/engine.dart | 12 +++++------- lib/src/core/signal_client.dart | 22 ++++++++++++++-------- lib/src/exceptions.dart | 5 +++++ lib/src/internal/events.dart | 2 ++ 4 files changed, 26 insertions(+), 15 deletions(-) diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index aab234f53..6fea7c8f1 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -677,7 +677,7 @@ class Engine extends Disposable with EventsEmittable { try { attemptingReconnect = true; - if (await signalClient.checkInternetConnection() == false) { + if (await signalClient.networkIsAvailable() == false) { logger.fine('no internet connection, waiting...'); await signalClient.events.waitFor( duration: connectOptions.timeouts.connection * 10, @@ -697,14 +697,12 @@ class Engine extends Disposable with EventsEmittable { } catch (e) { reconnectAttempts = reconnectAttempts! + 1; bool recoverable = true; - if (e is WebSocketException || - e is ConnectException || - e is MediaConnectException) { + if (e is WebSocketException || e is MediaConnectException) { // cannot resume connection, need to do full reconnect fullReconnectOnNext = true; - } else if (e is TimeoutException) { - fullReconnectOnNext = false; - } else { + } + + if (e is UnexpectedConnectionState) { recoverable = false; } diff --git a/lib/src/core/signal_client.dart b/lib/src/core/signal_client.dart index 10536ca2e..bf85c068b 100644 --- a/lib/src/core/signal_client.dart +++ b/lib/src/core/signal_client.dart @@ -61,7 +61,7 @@ class SignalClient extends Disposable with EventsEmittable { ConnectivityResult? _connectivityResult; StreamSubscription? connectivitySubscription; - Future checkInternetConnection() async { + Future networkIsAvailable() async { if (!kIsWeb && !lkPlatformIsTest()) { return true; } @@ -98,15 +98,19 @@ class SignalClient extends Disposable with EventsEmittable { .onConnectivityChanged .listen((ConnectivityResult result) { if (_connectivityResult != result) { - _connectivityResult = result; if (result == ConnectivityResult.none) { - logger.warning('lost internet connection'); + logger.warning('lost Connectivity'); } else { - logger.info('internet connection restored'); - events.emit(SignalConnectivityChangedEvent( - state: result, - )); + logger.info( + 'Connectivity changed, ${_connectivityResult!.name} => ${result.name}'); } + + events.emit(SignalConnectivityChangedEvent( + oldState: _connectivityResult!, + state: result, + )); + + _connectivityResult = result; } }); @@ -140,7 +144,7 @@ class SignalClient extends Disposable with EventsEmittable { // Clean up existing socket await cleanUp(); // Attempt to connect - _ws = await _wsConnector( + var future = _wsConnector( rtcUri, WebSocketEventHandlers( onData: _onSocketData, @@ -148,6 +152,8 @@ class SignalClient extends Disposable with EventsEmittable { onError: _onSocketError, ), ); + future = future.timeout(connectOptions.timeouts.connection); + _ws = await future; // Successful connection _connectionState = ConnectionState.connected; events.emit(const SignalConnectedEvent()); diff --git a/lib/src/exceptions.dart b/lib/src/exceptions.dart index 46245bbf3..9c0fe796f 100644 --- a/lib/src/exceptions.dart +++ b/lib/src/exceptions.dart @@ -88,3 +88,8 @@ class LiveKitE2EEException extends LiveKitException { @override String toString() => 'E2EE Exception: [$runtimeType] $message'; } + +class UnexpectedConnectionState extends LiveKitException { + UnexpectedConnectionState([String msg = 'Unexpected connection state']) + : super._(msg); +} diff --git a/lib/src/internal/events.dart b/lib/src/internal/events.dart index 5e6a34b51..3a39f7e81 100644 --- a/lib/src/internal/events.dart +++ b/lib/src/internal/events.dart @@ -143,8 +143,10 @@ class SignalReconnectResponseEvent with SignalEvent, InternalEvent { @internal class SignalConnectivityChangedEvent with SignalEvent, InternalEvent { + final ConnectivityResult oldState; final ConnectivityResult state; const SignalConnectivityChangedEvent({ + required this.oldState, required this.state, }); } From fdd7da86ba2ef156aa69d254eeb17178dde6976a Mon Sep 17 00:00:00 2001 From: CloudWebRTC Date: Wed, 27 Dec 2023 09:31:49 +0800 Subject: [PATCH 7/8] Update lib/src/core/signal_client.dart Co-authored-by: David Zhao --- lib/src/core/signal_client.dart | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/src/core/signal_client.dart b/lib/src/core/signal_client.dart index bf85c068b..743a458f1 100644 --- a/lib/src/core/signal_client.dart +++ b/lib/src/core/signal_client.dart @@ -99,7 +99,7 @@ class SignalClient extends Disposable with EventsEmittable { .listen((ConnectivityResult result) { if (_connectivityResult != result) { if (result == ConnectivityResult.none) { - logger.warning('lost Connectivity'); + logger.warning('lost connectivity'); } else { logger.info( 'Connectivity changed, ${_connectivityResult!.name} => ${result.name}'); From 8528f9b294abf920b86dc48766e66d9c3900987c Mon Sep 17 00:00:00 2001 From: CloudWebRTC Date: Wed, 27 Dec 2023 09:40:12 +0800 Subject: [PATCH 8/8] Update signal_client.dart --- lib/src/core/signal_client.dart | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/lib/src/core/signal_client.dart b/lib/src/core/signal_client.dart index 743a458f1..7e989c459 100644 --- a/lib/src/core/signal_client.dart +++ b/lib/src/core/signal_client.dart @@ -62,7 +62,8 @@ class SignalClient extends Disposable with EventsEmittable { StreamSubscription? connectivitySubscription; Future networkIsAvailable() async { - if (!kIsWeb && !lkPlatformIsTest()) { + // Skip check for web or flutter test + if (kIsWeb || lkPlatformIsTest()) { return true; } _connectivityResult = await Connectivity().checkConnectivity();