From 99460dd436d4d3e4bad3243eceeef256153d8ec5 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Fri, 4 Mar 2022 20:10:28 +0900 Subject: [PATCH] `EngineConnectionStateUpdatedEvent` instead of individual events --- lib/src/core/engine.dart | 62 +++++++++++++------------- lib/src/core/room.dart | 74 +++++++++++++++---------------- lib/src/core/signal_client.dart | 41 ++++++++--------- lib/src/internal/events.dart | 36 +++++++-------- test/core/signal_client_test.dart | 4 +- 5 files changed, 108 insertions(+), 109 deletions(-) diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 93b38fc..003e8ad 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -84,7 +84,8 @@ class Engine extends Disposable with EventsEmittable { events.listen((event) => logger.fine('[EngineEvent] $objectId ${event}')); } - _setUpListeners(); + _setUpEngineListeners(); + _setUpSignalListeners(); onDispose(() async { await events.dispose(); @@ -531,33 +532,6 @@ class Engine extends Disposable with EventsEmittable { await reconnect(); } - void _updateConnectionState(ConnectionState newValue) { - if (_connectionState == newValue) return; - - logger.fine('Engine ConnectionState ' - '${_connectionState.name} -> ${newValue.name}'); - - bool didReconnect = _connectionState == ConnectionState.reconnecting && - newValue == ConnectionState.connected; - // update internal value - _connectionState = newValue; - // emit event - if (_connectionState == ConnectionState.connected) { - if (didReconnect) { - events.emit(const EngineReconnectedEvent()); - // send queued requests if engine re-connected - signalClient.sendQueuedRequests(); - } else { - events.emit(const EngineConnectedEvent()); - } - } else if (_connectionState == ConnectionState.reconnecting) { - events.emit(const EngineReconnectingEvent()); - } else if (_connectionState == ConnectionState.disconnected) { - signalClient.cleanUp(); - events.emit(const EngineDisconnectedEvent()); - } - } - @internal void sendSyncState({ required lk_rtc.UpdateSubscription subscription, @@ -571,7 +545,15 @@ class Engine extends Disposable with EventsEmittable { ); } - void _setUpListeners() => _signalListener + void _setUpEngineListeners() => + events.on((event) async { + if (event.didReconnect) { + // send queued requests if engine re-connected + signalClient.sendQueuedRequests(); + } + }); + + void _setUpSignalListeners() => _signalListener ..on((event) async { // create peer connections _subscriberPrimary = event.response.subscriberPrimary; @@ -666,7 +648,6 @@ class Engine extends Disposable with EventsEmittable { return; } await close(); - events.emit(const EngineDisconnectedEvent()); }) ..on( (event) => events.emit(EngineRemoteMuteChangedEvent( @@ -675,6 +656,27 @@ class Engine extends Disposable with EventsEmittable { ))); } +extension EngineInternalMethods on Engine { + void _updateConnectionState(ConnectionState newValue) { + if (_connectionState == newValue) return; + + logger.fine('Engine ConnectionState ' + '${_connectionState.name} -> ${newValue.name}'); + + bool didReconnect = _connectionState == ConnectionState.reconnecting && + newValue == ConnectionState.connected; + // update internal value + final oldState = _connectionState; + _connectionState = newValue; + + events.emit(EngineConnectionStateUpdatedEvent( + newState: _connectionState, + oldState: oldState, + didReconnect: didReconnect, + )); + } +} + extension EnginePrivateMethods on Engine { // publisher data channel for the reliability rtc.RTCDataChannel? _publisherDataChannel(Reliability reliability) => diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index a903f3a..373e2a4 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -116,21 +116,19 @@ class Room extends DisposableChangeNotifier with EventsEmittable { } void _setUpListeners() => _engineListener - ..on((event) async { - // _connectionState = ConnectionState.connected; + ..on((event) async { + if (event.didReconnect) { + events.emit(const RoomReconnectedEvent()); + await _handlePostReconnect(false); + } else if (event.newState == ConnectionState.reconnecting) { + events.emit(const RoomReconnectingEvent()); + } else if (event.newState == ConnectionState.disconnected) { + await _cleanUp(); + events.emit(const RoomDisconnectedEvent()); + } + // always notify ChangeNotifier notifyListeners(); }) - ..on((event) async { - events.emit(const RoomReconnectedEvent()); - await _handlePostReconnect(false); - }) - ..on((event) async { - events.emit(const RoomReconnectingEvent()); - }) - ..on((event) async { - await _handleClose(); - events.emit(const RoomDisconnectedEvent()); - }) ..on((event) { // during reconnection, need to send sync state upon signal connection. if (event.didReconnect) { @@ -255,7 +253,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { if (connectionState != ConnectionState.disconnected) { engine.signalClient.sendLeave(); } - await _handleClose(); + await _cleanUp(); } Future reconnect() async { @@ -289,29 +287,6 @@ class Room extends DisposableChangeNotifier with EventsEmittable { return participant; } - // there should be no problem calling this method multiple times - Future _handleClose() async { - logger.fine('[$objectId] _handleClose()'); - if (connectionState == ConnectionState.disconnected) { - logger.warning('[$objectId]: close() already disconnected'); - } - - // clean up RemoteParticipants - for (final _ in _participants.values.toList()) { - // RemoteParticipant is responsible for disposing resources - await _.dispose(); - } - _participants.clear(); - - // clean up LocalParticipant - await localParticipant?.unpublishAllTracks(); - - // clean up engine - await engine.close(); - - _activeSpeakers.clear(); - } - Future _onParticipantUpdateEvent( List updates) async { // trigger change notifier only if list of participants membership is changed @@ -510,3 +485,28 @@ class Room extends DisposableChangeNotifier with EventsEmittable { ); } } + +extension RoomPrivateMethods on Room { + // resets internal state to a re-usable state + Future _cleanUp() async { + logger.fine('[$objectId] _handleClose()'); + if (connectionState == ConnectionState.disconnected) { + logger.warning('[$objectId]: close() already disconnected'); + } + + // clean up RemoteParticipants + for (final _ in _participants.values.toList()) { + // RemoteParticipant is responsible for disposing resources + await _.dispose(); + } + _participants.clear(); + + // clean up LocalParticipant + await localParticipant?.unpublishAllTracks(); + + // clean up engine + await engine.close(); + + _activeSpeakers.clear(); + } +} diff --git a/lib/src/core/signal_client.dart b/lib/src/core/signal_client.dart index 32600fe..58836d5 100644 --- a/lib/src/core/signal_client.dart +++ b/lib/src/core/signal_client.dart @@ -143,25 +143,6 @@ class SignalClient extends Disposable with EventsEmittable { _ws?.send(req.writeToBuffer()); } - void _updateConnectionState(ConnectionState newValue) { - if (_connectionState == newValue) return; - - logger.fine('SignalClient ConnectionState ' - '${_connectionState.name} -> ${newValue.name}'); - - bool didReconnect = _connectionState == ConnectionState.reconnecting && - newValue == ConnectionState.connected; - - final oldState = _connectionState; - _connectionState = newValue; - - events.emit(SignalConnectionStateUpdatedEvent( - newState: _connectionState, - oldState: oldState, - didReconnect: didReconnect, - )); - } - Future _onSocketData(dynamic message) async { if (message is! List) return; final msg = lk_rtc.SignalResponse.fromBuffer(message); @@ -413,8 +394,28 @@ extension on lk_rtc.SignalRequest { ].contains(whichMessage()); } -// internal methods +extension SignalClientPrivateMethods on SignalClient { + void _updateConnectionState(ConnectionState newValue) { + if (_connectionState == newValue) return; + logger.fine('SignalClient ConnectionState ' + '${_connectionState.name} -> ${newValue.name}'); + + bool didReconnect = _connectionState == ConnectionState.reconnecting && + newValue == ConnectionState.connected; + + final oldState = _connectionState; + _connectionState = newValue; + + events.emit(SignalConnectionStateUpdatedEvent( + newState: _connectionState, + oldState: oldState, + didReconnect: didReconnect, + )); + } +} + +// internal methods extension SignalClientInternalMethods on SignalClient { @internal void sendQueuedRequests() { diff --git a/lib/src/internal/events.dart b/lib/src/internal/events.dart index dd19dba..afd5a46 100644 --- a/lib/src/internal/events.dart +++ b/lib/src/internal/events.dart @@ -124,6 +124,22 @@ class SignalConnectionStateUpdatedEvent extends ConnectionStateUpdatedEvent ); } +@internal +class EngineConnectionStateUpdatedEvent extends ConnectionStateUpdatedEvent + with EngineEvent { + const EngineConnectionStateUpdatedEvent({ + required ConnectionState newState, + required ConnectionState oldState, + required bool didReconnect, + DisconnectReason? disconnectReason, + }) : super( + newState: newState, + oldState: oldState, + didReconnect: didReconnect, + disconnectReason: disconnectReason, + ); +} + @internal class SignalOfferEvent with SignalEvent, InternalEvent { final rtc.RTCSessionDescription sd; @@ -270,26 +286,6 @@ class SignalTokenUpdatedEvent with SignalEvent, InternalEvent { // Engine events // ---------------------------------------------------------------------- -@internal -class EngineConnectedEvent with EngineEvent, InternalEvent { - const EngineConnectedEvent(); -} - -@internal -class EngineDisconnectedEvent with EngineEvent, InternalEvent { - const EngineDisconnectedEvent(); -} - -@internal -class EngineReconnectingEvent with EngineEvent, InternalEvent { - const EngineReconnectingEvent(); -} - -@internal -class EngineReconnectedEvent with EngineEvent, InternalEvent { - const EngineReconnectedEvent(); -} - @internal class EngineTrackAddedEvent with EngineEvent, InternalEvent { final rtc.MediaStreamTrack track; diff --git a/test/core/signal_client_test.dart b/test/core/signal_client_test.dart index f470e4b..87ea452 100644 --- a/test/core/signal_client_test.dart +++ b/test/core/signal_client_test.dart @@ -35,8 +35,8 @@ void main() { expect( client.events.streamCtrl.stream, emitsInOrder([ - predicate((event) => - event.newState == ConnectionState.reconnecting), + predicate( + (event) => event.newState == ConnectionState.reconnecting), predicate((event) => event.newState == ConnectionState.connected && event.didReconnect == true),