diff --git a/example/lib/exts.dart b/example/lib/exts.dart index 309aca0..aa1504a 100644 --- a/example/lib/exts.dart +++ b/example/lib/exts.dart @@ -154,4 +154,25 @@ extension LKExampleExt on BuildContext { ], ), ); + + Future showSimulateScenarioDialog() => + showDialog( + context: this, + builder: (ctx) => SimpleDialog( + title: const Text('Simulate Scenario'), + children: SimulateScenarioResult.values + .map((e) => SimpleDialogOption( + child: Text(e.name), + onPressed: () => Navigator.pop(ctx, e), + )) + .toList(), + ), + ); +} + +enum SimulateScenarioResult { + nodeFailure, + migration, + serverLeave, + clear, } diff --git a/example/lib/main.dart b/example/lib/main.dart index b6ec9af..10f71a3 100644 --- a/example/lib/main.dart +++ b/example/lib/main.dart @@ -2,10 +2,9 @@ import 'package:flutter/material.dart'; import 'package:livekit_example/theme.dart'; import 'package:logging/logging.dart'; import 'package:intl/intl.dart'; - import 'pages/connect.dart'; -void main() { +void main() async { final format = DateFormat('HH:mm:ss'); // configure logs for debugging Logger.root.level = Level.FINE; diff --git a/example/lib/widgets/controls.dart b/example/lib/widgets/controls.dart index 68b9fec..1bbb951 100644 --- a/example/lib/widgets/controls.dart +++ b/example/lib/widgets/controls.dart @@ -149,6 +149,18 @@ class _ControlsWidgetState extends State { } } + void _onTapSimulateScenario() async { + final result = await context.showSimulateScenarioDialog(); + if (result != null) { + print('${result}'); + await widget.room.simulateScenario( + nodeFailure: result == SimulateScenarioResult.nodeFailure ? true : null, + migration: result == SimulateScenarioResult.migration ? true : null, + serverLeave: result == SimulateScenarioResult.serverLeave ? true : null, + ); + } + } + void _onTapSendData() async { final result = await context.showSendDataDialog(); if (result == true) { @@ -238,6 +250,11 @@ class _ControlsWidgetState extends State { icon: const Icon(EvaIcons.settings2), tooltip: 'Subscribe permission', ), + IconButton( + onPressed: _onTapSimulateScenario, + icon: const Icon(EvaIcons.alertTriangle), + tooltip: 'Simulate scenario', + ), ], ), ); diff --git a/example/pubspec.lock b/example/pubspec.lock index 75e84ff..71a0ad1 100644 --- a/example/pubspec.lock +++ b/example/pubspec.lock @@ -50,6 +50,13 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "3.0.1" + csslib: + dependency: transitive + description: + name: csslib + url: "https://pub.dartlang.org" + source: hosted + version: "0.17.1" dart_webrtc: dependency: transitive description: @@ -84,7 +91,7 @@ packages: name: device_info_plus_platform_interface url: "https://pub.dartlang.org" source: hosted - version: "2.3.0" + version: "2.3.0+1" device_info_plus_web: dependency: transitive description: @@ -184,6 +191,13 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "2.2.0" + html: + dependency: transitive + description: + name: html + url: "https://pub.dartlang.org" + source: hosted + version: "0.15.0" http: dependency: transitive description: @@ -338,13 +352,6 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "3.1.0" - platform_detect: - dependency: transitive - description: - name: platform_detect - url: "https://pub.dartlang.org" - source: hosted - version: "2.0.3" plugin_platform_interface: dependency: transitive description: @@ -373,13 +380,6 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "6.0.2" - pub_semver: - dependency: transitive - description: - name: pub_semver - url: "https://pub.dartlang.org" - source: hosted - version: "2.1.0" shared_preferences: dependency: "direct main" description: @@ -497,6 +497,20 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "1.3.0" + universal_html: + dependency: transitive + description: + name: universal_html + url: "https://pub.dartlang.org" + source: hosted + version: "2.0.8" + universal_io: + dependency: transitive + description: + name: universal_io + url: "https://pub.dartlang.org" + source: hosted + version: "2.0.4" uuid: dependency: transitive description: @@ -511,6 +525,13 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "2.1.1" + web_browser_detect: + dependency: transitive + description: + name: web_browser_detect + url: "https://pub.dartlang.org" + source: hosted + version: "2.0.2" webrtc_interface: dependency: transitive description: diff --git a/lib/livekit_client.dart b/lib/livekit_client.dart index 8c3e48c..6060855 100644 --- a/lib/livekit_client.dart +++ b/lib/livekit_client.dart @@ -24,4 +24,5 @@ export 'src/track/remote/remote.dart'; export 'src/track/remote/video.dart'; export 'src/track/track.dart'; export 'src/types.dart'; +export 'src/utils.dart'; export 'src/widget/video_track_renderer.dart'; diff --git a/lib/src/constants.dart b/lib/src/constants.dart index b78fccf..eaab5c4 100644 --- a/lib/src/constants.dart +++ b/lib/src/constants.dart @@ -2,6 +2,6 @@ class Timeouts { static const connection = Duration(seconds: 10); static const debounce = Duration(milliseconds: 100); static const publish = Duration(seconds: 10); - static const iceConnection = Duration(seconds: 10); + static const peerConnection = Duration(seconds: 10); static const iceRestart = Duration(seconds: 10); } diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 9f2fb9b..a6e1c8e 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -19,6 +19,7 @@ import '../proto/livekit_models.pb.dart' as lk_models; import '../proto/livekit_rtc.pb.dart' as lk_rtc; import '../support/disposable.dart'; import '../types.dart'; +import '../utils.dart'; import 'room.dart'; import 'signal_client.dart'; import 'transport.dart'; @@ -26,7 +27,6 @@ import 'transport.dart'; class Engine extends Disposable with EventsEmittable { static const _lossyDCLabel = '_lossy'; static const _reliableDCLabel = '_reliable'; - static const _maxReconnectAttempts = 5; // Reference to the Room final Room room; @@ -48,8 +48,6 @@ class Engine extends Disposable with EventsEmittable { rtc.RTCDataChannel? _reliableDCSub; rtc.RTCDataChannel? _lossyDCSub; - bool _iceConnected = false; - ConnectionState _connectionState = ConnectionState.disconnected; /// Connection state of the [Room]. @@ -68,9 +66,6 @@ class Engine extends Disposable with EventsEmittable { // server-provided ice servers List _serverProvidedIceServers = []; - // internal - int _reconnectAttempts = 0; - late final _signalListener = signalClient.createListener(synchronized: true); final delays = CancelableDelayManager(); @@ -94,34 +89,51 @@ class Engine extends Disposable with EventsEmittable { }); } - Future connect( + Future connect( String url, String token, ) async { this.url = url; this.token = token; - // connect to rtc server - await signalClient.connect( - url, - token, - connectOptions: room.connectOptions, - ); + _updateConnectionState(ConnectionState.connecting); - // wait for join response - final event = await _signalListener.waitFor( - duration: Timeouts.connection, - onTimeout: () => throw ConnectException(), - ); + try { + // wait for socket to connect rtc server + await signalClient.connect( + url, + token, + connectOptions: room.connectOptions, + ); - return event.response; + // wait for join response + await _signalListener.waitFor( + duration: Timeouts.connection, + onTimeout: () => throw ConnectException(), + ); + + logger.fine('Waiting for engine to connect...'); + + // wait until engine is connected + await events.waitFor( + filter: (event) => event.isPrimary && event.state.isConnected(), + duration: Timeouts.connection, + onTimeout: () => throw ConnectException(), + ); + + _updateConnectionState(ConnectionState.connected); + } catch (error) { + logger.fine('Connect Error $error'); + _updateConnectionState(ConnectionState.disconnected); + rethrow; + } } /// Close connection between the server. Future close() async { - logger.fine('[$objectId] close()'); + logger.fine('${runtimeType}.close()'); if (_connectionState == ConnectionState.disconnected) { - logger.warning('[$objectId]: close() already disconnected'); + logger.warning('${runtimeType}.close() already disconnected'); } // _statsTimer.cancel(); // cancel all ongoing delays @@ -134,9 +146,9 @@ class Engine extends Disposable with EventsEmittable { await subscriber?.dispose(); subscriber = null; - await signalClient.close(); + await signalClient.disconnect(); - _connectionState = ConnectionState.disconnected; + _updateConnectionState(ConnectionState.disconnected); // notifyListeners(); } @@ -148,7 +160,7 @@ class Engine extends Disposable with EventsEmittable { required lk_models.TrackSource source, VideoDimensions? dimensions, bool? dtx, - List? videoLayers, + Iterable? videoLayers, }) async { // TODO: Check if cid already published @@ -205,19 +217,19 @@ class Engine extends Disposable with EventsEmittable { if (_subscriberPrimary) { // make sure publisher transport is connected - if (publisher?.pc.iceConnectionState?.isConnected() != true) { + if (publisher?.pc.connectionState?.isConnected() != true) { logger.fine('Publisher is not connected...'); // start negotiation - if (publisher?.pc.iceConnectionState != - rtc.RTCIceConnectionState.RTCIceConnectionStateChecking) { + if (publisher?.pc.connectionState != + rtc.RTCPeerConnectionState.RTCPeerConnectionStateConnecting) { await negotiate(); } logger.fine('Waiting for publisher to ice-connect...'); - await events.waitFor( - filter: (event) => event.iceState.isConnected(), - duration: Timeouts.iceConnection, + await events.waitFor( + filter: (event) => event.state.isConnected(), + duration: Timeouts.peerConnection, ); } @@ -251,25 +263,17 @@ class Engine extends Disposable with EventsEmittable { return; } - final url = this.url; - final token = this.token; - if (url == null || token == null) { throw ConnectException('could not reconnect without url and token'); } - _connectionState = ConnectionState.reconnecting; - - if (_reconnectAttempts == 0) { - events.emit(const EngineReconnectingEvent()); - } - _reconnectAttempts++; - - try { - await signalClient.reconnect( - url, - token, + Future sequence() async { + // + await signalClient.connect( + url!, + token!, connectOptions: room.connectOptions, + reconnect: true, ); if (publisher == null || subscriber == null) { @@ -285,29 +289,37 @@ class Engine extends Disposable with EventsEmittable { )); } - final iceConnected = - primary?.pc.iceConnectionState?.isConnected() ?? false; + final iceConnected = primary?.pc.connectionState?.isConnected() ?? false; logger.fine('Reconnect: iceConnected: $iceConnected'); if (!iceConnected) { logger.fine('Reconnect: Waiting for primary to connect...'); - await events.waitFor( - filter: (event) => event.isPrimary && event.iceState.isConnected(), + await events.waitFor( + filter: (event) => event.isPrimary && event.state.isConnected(), duration: Timeouts.iceRestart, + onTimeout: () => throw ConnectException(), ); } + } - logger.fine('Reconnect: success'); - _connectionState = ConnectionState.connected; - events.emit(const EngineReconnectedEvent()); - _reconnectAttempts = 0; + try { + _updateConnectionState(ConnectionState.reconnecting); + await Utils.retry( + (tries, errors) { + logger.fine('Retrying connect sequence remaining ${tries} tries...'); + return sequence(); + }, + retryCondition: (_, __) => + _connectionState == ConnectionState.reconnecting, + tries: 3, + delay: const Duration(seconds: 3), + ); + _updateConnectionState(ConnectionState.connected); } catch (error) { - logger.fine('Reconnect: error ${error}'); - // Pass up all exceptions - _connectionState = ConnectionState.disconnected; - rethrow; + // + _updateConnectionState(ConnectionState.disconnected); } } @@ -353,39 +365,26 @@ class Engine extends Disposable with EventsEmittable { subscriber?.pc.onDataChannel = _onDataChannel; } - subscriber?.pc.onIceConnectionState = - (state) => events.emit(EngineSubscriberIceStateUpdatedEvent( - iceState: state, + subscriber?.pc.onConnectionState = + (state) => events.emit(EngineSubscriberPeerStateUpdatedEvent( + state: state, isPrimary: _subscriberPrimary, )); - publisher?.pc.onIceConnectionState = - (state) => events.emit(EnginePublisherIceStateUpdatedEvent( + publisher?.pc.onConnectionState = + (state) => events.emit(EnginePublisherPeerStateUpdatedEvent( state: state, isPrimary: !_subscriberPrimary, )); - events.on((event) { + events.on((event) { // only listen to primary ice events if (!event.isPrimary) return; - if (event.iceState == - rtc.RTCIceConnectionState.RTCIceConnectionStateConnected) { - if (!_iceConnected) { - _iceConnected = true; - if (_connectionState == ConnectionState.reconnecting) { - events.emit(const EngineReconnectedEvent()); - } else { - events.emit(const EngineConnectedEvent()); - } - } - } else if (event.iceState == - rtc.RTCIceConnectionState.RTCIceConnectionStateFailed) { + if (event.state == + rtc.RTCPeerConnectionState.RTCPeerConnectionStateFailed) { // trigger reconnect sequence - if (_iceConnected) { - _iceConnected = false; - _onDisconnected('peerconnection'); - } + _onDisconnected(DisconnectReason.peerConnection); } }); @@ -502,8 +501,9 @@ class Engine extends Disposable with EventsEmittable { final dp = lk_models.DataPacket.fromBuffer(message.binary); if (dp.whichValue() == lk_models.DataPacket_Value.speaker) { // Speaker packet - events - .emit(EngineActiveSpeakersUpdateEvent(speakers: dp.speaker.speakers)); + events.emit(EngineActiveSpeakersUpdateEvent( + speakers: dp.speaker.speakers, + )); } else if (dp.whichValue() == lk_models.DataPacket_Value.user) { // User packet events.emit(EngineDataPacketReceivedEvent( @@ -513,44 +513,63 @@ class Engine extends Disposable with EventsEmittable { } } - Future _onDisconnected(String reason) async { - logger.info('onDisconnected reason: $reason'); + Future _onDisconnected(DisconnectReason reason) async { + logger + .info('onDisconnected state:${_connectionState} reason:${reason.name}'); if (_connectionState == ConnectionState.disconnected) { - logger.fine('[$objectId] Already disconnected $reason'); + logger.fine('[$objectId] Already disconnected... $reason'); + return; + } + if (_connectionState == ConnectionState.reconnecting) { + logger.fine('[$objectId] Already reconnecting...'); return; } - logger.fine('[$objectId] Disconnected $reason'); + logger.fine('[$runtimeType] Should attempt reconnect sequence...'); - if (_reconnectAttempts >= _maxReconnectAttempts) { - logger.info('[$objectId] Could not connect ' - 'after ${_reconnectAttempts} attempts, giving up'); - await close(); - events.emit(const EngineDisconnectedEvent()); - return; - } + await reconnect(); + } - final delay = - Duration(milliseconds: (_reconnectAttempts * _reconnectAttempts) * 300); + void _updateConnectionState(ConnectionState newValue) { + if (_connectionState == newValue) return; - // if this instance is disposed, we probably don't want to continue any more - // so the whole block will be canceled from being executed - await delays.waitFor(delay, ifNotCancelled: () async { - try { - await reconnect(); - _reconnectAttempts = 0; - } catch (_) { - // doesn't need to be awaited - // ignore: unawaited_futures - _onDisconnected(reason); + 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()); + } else { + events.emit(const EngineConnectedEvent()); } - }); + } else if (_connectionState == ConnectionState.reconnecting) { + events.emit(const EngineReconnectingEvent()); + } else if (_connectionState == ConnectionState.disconnected) { + events.emit(const EngineDisconnectedEvent()); + } + } + + @internal + void sendSyncState({ + required lk_rtc.UpdateSubscription subscription, + required Iterable? publishTracks, + }) async { + final answer = (await subscriber?.pc.getLocalDescription())?.toPBType(); + signalClient.sendSyncState( + answer: answer, + subscription: subscription, + publishTracks: publishTracks, + ); } void _setUpListeners() => _signalListener - ..on((event) async { + ..on((event) async { // create peer connections - _connectionState = ConnectionState.connected; _subscriberPrimary = event.response.subscriberPrimary; _serverProvidedIceServers = event.response.iceServers; @@ -564,9 +583,16 @@ class Engine extends Disposable with EventsEmittable { // for subscriberPrimary, we negotiate when necessary (lazy) await negotiate(); } + + // Relay to Room + events.emit(event); }) - ..on((_) async { - await _onDisconnected('signal'); + ..on((event) async { + if (event.connectionState == ConnectionState.disconnected) { + await _onDisconnected(DisconnectReason.signal); + } + // Relay to Room + events.emit(event); }) ..on((event) async { if (subscriber == null) { @@ -611,21 +637,23 @@ class Engine extends Disposable with EventsEmittable { await publisher!.addIceCandidate(event.candidate); } }) - // relay + // relay to Room ..on((event) => events.emit(event)) - // relay + // relay to Room ..on((event) => events.emit(event)) - // relay + // relay to Room ..on((event) => events.emit(event)) - // relay + // relay to Room ..on((event) => events.emit(event)) // relay to Room ..on((event) => events.emit(event)) // relay to Room ..on((event) => events.emit(event)) ..on((event) async { - if (connectionState == ConnectionState.reconnecting) { - logger.warning('Received leave signal while engine is reconnecting.'); + if (_connectionState == ConnectionState.reconnecting) { + logger.warning( + '[Signal] Received Leave while engine is reconnecting, ignoring...'); + return; } await close(); events.emit(const EngineDisconnectedEvent()); diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index 000d51c..97725be 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -2,9 +2,8 @@ import 'dart:collection'; import 'package:collection/collection.dart'; -import '../constants.dart'; +import '../core/signal_client.dart'; import '../events.dart'; -import '../exceptions.dart'; import '../extensions.dart'; import '../internal/events.dart'; import '../logger.dart'; @@ -18,7 +17,6 @@ import '../proto/livekit_rtc.pb.dart' as lk_rtc; import '../support/disposable.dart'; import '../track/track.dart'; import '../types.dart'; -import '../core/signal_client.dart'; import 'engine.dart'; /// Room is the primary construct for LiveKit conferences. It contains a @@ -106,38 +104,10 @@ class Room extends DisposableChangeNotifier with EventsEmittable { this.connectOptions = connectOptions ?? this.connectOptions; this.roomOptions = roomOptions ?? this.roomOptions; - final joinResponse = await engine.connect( + return engine.connect( url, token, ); - - sid = joinResponse.room.sid; - name = joinResponse.room.name; - _serverVersion = joinResponse.serverVersion; - - logger.fine( - 'Connected to LiveKit server, version: ${joinResponse.serverVersion}'); - - localParticipant = LocalParticipant( - room: this, - info: joinResponse.participant, - ); - - for (final info in joinResponse.otherParticipants) { - logger.fine('Creating RemoteParticipant: ${info.sid}(${info.identity}) ' - 'tracks:${info.tracks.map((e) => e.sid)}'); - _getOrCreateRemoteParticipant(info.sid, info); - } - - logger.fine('Waiting to engine connect...'); - - // wait until engine is connected - await _engineListener.waitFor( - duration: Timeouts.connection, - onTimeout: () => throw ConnectException(), - ); - - logger.fine('Room Connect completed'); } void _setUpListeners() => _engineListener @@ -154,6 +124,34 @@ class Room extends DisposableChangeNotifier with EventsEmittable { events.emit(const RoomReconnectingEvent()); }) ..on((event) => _handleClose()) + ..on((event) { + // during reconnection, need to send sync state upon signal connection. + if (event.didReconnect) { + logger.fine('Sending syncState'); + _sendSyncState(); + } + }) + ..on((event) { + sid = event.response.room.sid; + name = event.response.room.name; + _serverVersion = event.response.serverVersion; + + logger.fine('[Engine] Received JoinResponse, ' + 'serverVersion: ${event.response.serverVersion}'); + + localParticipant = LocalParticipant( + room: this, + info: event.response.participant, + ); + + for (final info in event.response.otherParticipants) { + logger.fine('Creating RemoteParticipant: ${info.sid}(${info.identity}) ' + 'tracks:${info.tracks.map((e) => e.sid)}'); + _getOrCreateRemoteParticipant(info.sid, info); + } + + logger.fine('Room Connect completed'); + }) ..on( (event) => _onParticipantUpdateEvent(event.participants)) ..on( @@ -459,4 +457,32 @@ class Room extends DisposableChangeNotifier with EventsEmittable { events.emit(ParticipantDisconnectedEvent(participant: participant)); } + + Future _sendSyncState() async { + final connectOptions = this.connectOptions ?? const ConnectOptions(); + final sendUnSub = connectOptions.autoSubscribe; + engine.sendSyncState( + subscription: lk_rtc.UpdateSubscription( + subscribe: !sendUnSub, + participantTracks: + participants.values.map((e) => e.participantTracks()), + ), + publishTracks: localParticipant?.publishedTracksInfo(), + ); + } + + /// To be used for internal testing purposes only. + Future simulateScenario({ + int? speakerUpdate, + bool? nodeFailure, + bool? migration, + bool? serverLeave, + }) async { + engine.signalClient.sendSimulateScenario( + speakerUpdate: speakerUpdate, + nodeFailure: nodeFailure, + migration: migration, + serverLeave: serverLeave, + ); + } } diff --git a/lib/src/core/signal_client.dart b/lib/src/core/signal_client.dart index fce0608..90fcdd5 100644 --- a/lib/src/core/signal_client.dart +++ b/lib/src/core/signal_client.dart @@ -2,6 +2,7 @@ import 'dart:async'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'package:http/http.dart' as http; +import 'package:meta/meta.dart'; import '../events.dart'; import '../exceptions.dart'; @@ -23,6 +24,7 @@ class SignalClient extends Disposable with EventsEmittable { LiveKitWebSocket? _ws; + @internal SignalClient() { events.listen((event) { logger.fine('[SignalEvent] $event'); @@ -30,24 +32,33 @@ class SignalClient extends Disposable with EventsEmittable { onDispose(() async { await events.dispose(); - await close(); + await _cleanUp(); }); } + @internal Future connect( String uriString, String token, { ConnectOptions? connectOptions, + bool reconnect = false, }) async { final rtcUri = await Utils.buildUri( uriString, token: token, connectOptions: connectOptions, + reconnect: reconnect, ); logger.fine('SignalClient connecting with url: $rtcUri'); try { + _updateConnectionState(reconnect + ? ConnectionState.reconnecting + : ConnectionState.connecting); + // Clean up existing socket + await _cleanUp(); + // Attempt to connect _ws = await LiveKitWebSocket.connect( rtcUri, WebSocketEventHandlers( @@ -56,18 +67,23 @@ class SignalClient extends Disposable with EventsEmittable { onError: _handleError, ), ); + // Successful connection + _updateConnectionState(ConnectionState.connected); } catch (socketError) { - // Re-build same uri for validate mode - final validateUri = await Utils.buildUri( - uriString, - token: token, - connectOptions: connectOptions, - validate: true, - forceSecure: rtcUri.isSecureScheme, - ); - // Attempt Validation try { + // Skip validation if reconnect mode + if (reconnect) rethrow; + + // Re-build same uri for validate mode + final validateUri = await Utils.buildUri( + uriString, + token: token, + connectOptions: connectOptions, + validate: true, + forceSecure: rtcUri.isSecureScheme, + ); + final validateResponse = await http.get(validateUri); if (validateResponse.statusCode != 200) { throw ConnectException(validateResponse.body); @@ -78,44 +94,22 @@ class SignalClient extends Disposable with EventsEmittable { if (error is ConnectException) rethrow; // HTTP doesn't work either throw ConnectException(); + } finally { + _updateConnectionState(ConnectionState.disconnected); + rethrow; } } } - Future reconnect( - String uriString, - String token, { - ConnectOptions? connectOptions, - }) async { - logger.fine('SignalClient reconnecting...'); - _connectionState = ConnectionState.reconnecting; + Future _cleanUp() async { await _ws?.dispose(); _ws = null; - - final rtcUri = await Utils.buildUri( - uriString, - token: token, - reconnect: true, - connectOptions: connectOptions, - ); - - _ws = await LiveKitWebSocket.connect( - rtcUri, - WebSocketEventHandlers( - onData: _onSocketData, - onDispose: _onSocketDispose, - onError: _handleError, - ), - ); - - logger.fine('SignalClient socket reconnected'); - _connectionState = ConnectionState.connected; } - Future close() async { - logger.fine('SignalClient close'); - await _ws?.dispose(); - _ws = null; + @internal + Future disconnect() async { + logger.fine('SignalClient disconnect'); + await _cleanUp(); } void _sendRequest(lk_rtc.SignalRequest req) { @@ -129,13 +123,30 @@ class SignalClient extends Disposable with EventsEmittable { _ws?.send(buf); } + void _updateConnectionState(ConnectionState newValue) { + if (_connectionState == newValue) return; + + logger.fine('SignalClient ConnectionState ' + '${_connectionState.name} -> ${newValue.name}'); + + bool didReconnect = _connectionState == ConnectionState.reconnecting && + newValue == ConnectionState.connected; + + _connectionState = newValue; + + events.emit(SignalConnectionStateUpdatedEvent( + connectionState: _connectionState, + didReconnect: didReconnect, + )); + } + Future _onSocketData(dynamic message) async { if (message is! List) return; final msg = lk_rtc.SignalResponse.fromBuffer(message); switch (msg.whichMessage()) { case lk_rtc.SignalResponse_Message.join: - events.emit(SignalConnectedEvent(response: msg.join)); + events.emit(SignalJoinResponseEvent(response: msg.join)); break; case lk_rtc.SignalResponse_Message.answer: events.emit(SignalAnswerEvent(sd: msg.answer.toSDKType())); @@ -209,23 +220,25 @@ class SignalClient extends Disposable with EventsEmittable { // don't emit event's when reconnecting state if (_connectionState != ConnectionState.reconnecting) { logger.fine('SignalClient did disconnect ${_connectionState}'); - _connectionState = ConnectionState.disconnected; - events.emit(const SignalCloseEvent()); + _updateConnectionState(ConnectionState.disconnected); } } } extension SignalClientRequests on SignalClient { + @internal void sendOffer(rtc.RTCSessionDescription offer) => _sendRequest(lk_rtc.SignalRequest( - offer: offer.toSDKType(), + offer: offer.toPBType(), )); + @internal void sendAnswer(rtc.RTCSessionDescription answer) => _sendRequest(lk_rtc.SignalRequest( - answer: answer.toSDKType(), + answer: answer.toPBType(), )); + @internal void sendIceCandidate( rtc.RTCIceCandidate candidate, lk_rtc.SignalTarget target) => _sendRequest( @@ -237,6 +250,7 @@ extension SignalClientRequests on SignalClient { ), ); + @internal void sendMuteTrack(String trackSid, bool muted) => _sendRequest(lk_rtc.SignalRequest( mute: lk_rtc.MuteTrackRequest( @@ -245,6 +259,7 @@ extension SignalClientRequests on SignalClient { ), )); + @internal void sendAddTrack({ required String cid, required String name, @@ -252,7 +267,7 @@ extension SignalClientRequests on SignalClient { required lk_models.TrackSource source, VideoDimensions? dimensions, bool? dtx, - List? videoLayers, + Iterable? videoLayers, }) { final req = lk_rtc.AddTrackRequest( cid: cid, @@ -284,19 +299,22 @@ extension SignalClientRequests on SignalClient { )); } + @internal void sendUpdateTrackSettings(lk_rtc.UpdateTrackSettings settings) => _sendRequest(lk_rtc.SignalRequest( trackSetting: settings, )); + @internal void sendUpdateSubscription(lk_rtc.UpdateSubscription subscription) => _sendRequest(lk_rtc.SignalRequest( subscription: subscription, )); + @internal void sendUpdateVideoLayers( String trackSid, - List layers, + Iterable layers, ) => _sendRequest(lk_rtc.SignalRequest( updateLayers: lk_rtc.UpdateVideoLayers( @@ -305,9 +323,10 @@ extension SignalClientRequests on SignalClient { ), )); + @internal void sendUpdateSubscriptionPermissions({ required bool allParticipants, - required List trackPermissions, + required Iterable trackPermissions, }) => _sendRequest(lk_rtc.SignalRequest( subscriptionPermissions: lk_rtc.UpdateSubscriptionPermissions( @@ -316,7 +335,38 @@ extension SignalClientRequests on SignalClient { ), )); + @internal void sendLeave() => _sendRequest(lk_rtc.SignalRequest( leave: lk_rtc.LeaveRequest(), )); + + @internal + void sendSyncState({ + required lk_rtc.SessionDescription? answer, + required lk_rtc.UpdateSubscription subscription, + required Iterable? publishTracks, + }) => + _sendRequest(lk_rtc.SignalRequest( + syncState: lk_rtc.SyncState( + answer: answer, + subscription: subscription, + publishTracks: publishTracks, + ), + )); + + @internal + void sendSimulateScenario({ + int? speakerUpdate, + bool? nodeFailure, + bool? migration, + bool? serverLeave, + }) => + _sendRequest(lk_rtc.SignalRequest( + simulate: lk_rtc.SimulateScenario( + speakerUpdate: speakerUpdate, + nodeFailure: nodeFailure, + migration: migration, + serverLeave: serverLeave, + ), + )); } diff --git a/lib/src/core/transport.dart b/lib/src/core/transport.dart index 1149cce..8927375 100644 --- a/lib/src/core/transport.dart +++ b/lib/src/core/transport.dart @@ -31,7 +31,7 @@ class PCTransport extends Disposable { // Ensure callbacks won't fire any more pc.onRenegotiationNeeded = null; pc.onIceCandidate = null; - pc.onIceConnectionState = null; + pc.onConnectionState = null; pc.onTrack = null; // Remove all senders diff --git a/lib/src/extensions.dart b/lib/src/extensions.dart index d09b0ff..9551c88 100644 --- a/lib/src/extensions.dart +++ b/lib/src/extensions.dart @@ -42,6 +42,7 @@ extension ProtocolVersionExt on ProtocolVersion { ProtocolVersion.v3: '3', ProtocolVersion.v4: '4', ProtocolVersion.v5: '5', + ProtocolVersion.v6: '6', }[this]!; } @@ -65,11 +66,9 @@ extension RTCIceCandidateExt on rtc.RTCIceCandidate { String toJson() => json.encode(toMap()); } -extension RTCIceConnectionStateExt on rtc.RTCIceConnectionState { - bool isConnected() => [ - rtc.RTCIceConnectionState.RTCIceConnectionStateConnected, - rtc.RTCIceConnectionState.RTCIceConnectionStateCompleted, - ].contains(this); +extension RTCPeerConnectionStateExt on rtc.RTCPeerConnectionState { + bool isConnected() => + this == rtc.RTCPeerConnectionState.RTCPeerConnectionStateConnected; } extension RTCIceTransportPolicyExt on RTCIceTransportPolicy { @@ -82,7 +81,7 @@ extension RTCIceTransportPolicyExt on RTCIceTransportPolicy { // not so neat to directly expose protobuf types so we // define our own types (and convert methods) extension RTCSessionDescriptionExt on rtc.RTCSessionDescription { - lk_rtc.SessionDescription toSDKType() { + lk_rtc.SessionDescription toPBType() { return lk_rtc.SessionDescription(type: type, sdp: sdp); } } diff --git a/lib/src/internal/events.dart b/lib/src/internal/events.dart index 81e7ae6..d0d146d 100644 --- a/lib/src/internal/events.dart +++ b/lib/src/internal/events.dart @@ -10,42 +10,43 @@ import '../types.dart'; abstract class InternalEvent implements LiveKitEvent {} @internal -abstract class EngineIceStateUpdatedEvent with EngineEvent, InternalEvent { - final rtc.RTCIceConnectionState iceState; +abstract class EnginePeerStateUpdatedEvent with EngineEvent, InternalEvent { + final rtc.RTCPeerConnectionState state; final bool isPrimary; - const EngineIceStateUpdatedEvent({ - required this.iceState, + const EnginePeerStateUpdatedEvent({ + required this.state, required this.isPrimary, }); } @internal -class EngineSubscriberIceStateUpdatedEvent extends EngineIceStateUpdatedEvent { - const EngineSubscriberIceStateUpdatedEvent({ - required rtc.RTCIceConnectionState iceState, +class EngineSubscriberPeerStateUpdatedEvent + extends EnginePeerStateUpdatedEvent { + const EngineSubscriberPeerStateUpdatedEvent({ + required rtc.RTCPeerConnectionState state, required bool isPrimary, }) : super( - iceState: iceState, + state: state, isPrimary: isPrimary, ); @override String toString() => - '${runtimeType}(state: ${iceState}, isPrimary: ${isPrimary})'; + '${runtimeType}(state: ${state}, isPrimary: ${isPrimary})'; } @internal -class EnginePublisherIceStateUpdatedEvent extends EngineIceStateUpdatedEvent { - const EnginePublisherIceStateUpdatedEvent({ - required rtc.RTCIceConnectionState state, +class EnginePublisherPeerStateUpdatedEvent extends EnginePeerStateUpdatedEvent { + const EnginePublisherPeerStateUpdatedEvent({ + required rtc.RTCPeerConnectionState state, required bool isPrimary, }) : super( - iceState: state, + state: state, isPrimary: isPrimary, ); @override String toString() => - '${runtimeType}(state: ${iceState}, isPrimary: ${isPrimary})'; + '${runtimeType}(state: ${state}, isPrimary: ${isPrimary})'; } @internal @@ -80,19 +81,29 @@ class InternalTrackMuteUpdatedEvent with TrackEvent, InternalEvent { // @internal -class SignalConnectedEvent with SignalEvent, InternalEvent { +// Received a JoinResponse from the server. +class SignalJoinResponseEvent with SignalEvent, EngineEvent, InternalEvent { final lk_rtc.JoinResponse response; - const SignalConnectedEvent({ + const SignalJoinResponseEvent({ required this.response, }); } @internal -class SignalCloseEvent with SignalEvent, InternalEvent { - final CloseReason? reason; - const SignalCloseEvent({ - this.reason, +class SignalConnectionStateUpdatedEvent + with SignalEvent, EngineEvent, InternalEvent { + final ConnectionState connectionState; + final bool didReconnect; + final DisconnectReason? disconnectReason; + const SignalConnectionStateUpdatedEvent({ + required this.connectionState, + required this.didReconnect, + this.disconnectReason, }); + @override + String toString() => '$runtimeType(state: ${connectionState.name}, ' + 'didReconnect: ${didReconnect}, ' + 'disconnectReason: ${disconnectReason})'; } @internal diff --git a/lib/src/managers/event.dart b/lib/src/managers/event.dart index 61f2a6e..a4d58f2 100644 --- a/lib/src/managers/event.dart +++ b/lib/src/managers/event.dart @@ -79,7 +79,7 @@ abstract class EventsListenable extends Disposable { Future cancelAll() async { if (_listeners.isNotEmpty) { // Stop listening to all events - logger.fine('${objectId} cancelling ${_listeners.length} listeners(s)'); + logger.finer('${objectId} cancelling ${_listeners.length} listeners(s)'); for (final listener in _listeners) { await listener.cancel(); } @@ -135,7 +135,11 @@ abstract class EventsListenable extends Disposable { final completer = Completer(); final _cancelFunc = on( - (event) => completer.complete(event), + (event) { + if (!completer.isCompleted) { + completer.complete(event); + } + }, filter: filter, ); diff --git a/lib/src/options.dart b/lib/src/options.dart index d0eefc8..a402c87 100644 --- a/lib/src/options.dart +++ b/lib/src/options.dart @@ -22,7 +22,7 @@ class ConnectOptions { const ConnectOptions({ this.autoSubscribe = true, this.rtcConfiguration = const RTCConfiguration(), - this.protocolVersion = ProtocolVersion.v5, + this.protocolVersion = ProtocolVersion.v6, }); } diff --git a/lib/src/participant/local.dart b/lib/src/participant/local.dart index bc5e38e..c3baea8 100644 --- a/lib/src/participant/local.dart +++ b/lib/src/participant/local.dart @@ -10,6 +10,7 @@ import '../extensions.dart'; import '../logger.dart'; import '../options.dart'; import '../proto/livekit_models.pb.dart' as lk_models; +import '../proto/livekit_rtc.pb.dart' as lk_rtc; import '../publication/local.dart'; import '../track/local/audio.dart'; import '../track/local/local.dart'; @@ -331,4 +332,8 @@ class LocalParticipant extends Participant { allParticipants: allParticipantsAllowed, trackPermissions: trackPermissions.map((e) => e.toPBType()).toList(), ); + + @internal + Iterable publishedTracksInfo() => + trackPublications.values.map((e) => e.toPBTrackPublishedResponse()); } diff --git a/lib/src/participant/remote.dart b/lib/src/participant/remote.dart index 3e847ae..2a64fd4 100644 --- a/lib/src/participant/remote.dart +++ b/lib/src/participant/remote.dart @@ -216,4 +216,11 @@ class RemoteParticipant extends Participant { await pub.dispose(); } + + @internal + lk_models.ParticipantTracks participantTracks() => + lk_models.ParticipantTracks( + participantSid: sid, + trackSids: trackPublications.values.map((e) => e.sid), + ); } diff --git a/lib/src/publication/local.dart b/lib/src/publication/local.dart index dfdea85..41af929 100644 --- a/lib/src/publication/local.dart +++ b/lib/src/publication/local.dart @@ -80,4 +80,10 @@ class LocalTrackPublication extends TrackPublication { logger.fine('Update publishing layers: nothing to change'); } } + + lk_rtc.TrackPublishedResponse toPBTrackPublishedResponse() => + lk_rtc.TrackPublishedResponse( + cid: track?.mediaStreamTrack.id, + track: latestInfo, + ); } diff --git a/lib/src/publication/track_publication.dart b/lib/src/publication/track_publication.dart index c8354f4..6416dc6 100644 --- a/lib/src/publication/track_publication.dart +++ b/lib/src/publication/track_publication.dart @@ -48,6 +48,9 @@ abstract class TrackPublication extends Disposable { bool get subscribed => track != null; + @internal + lk_models.TrackInfo? latestInfo; + TrackPublication({ required lk_models.TrackInfo info, }) : sid = info.sid, @@ -69,6 +72,7 @@ abstract class TrackPublication extends Disposable { if (info.type == lk_models.TrackType.VIDEO) { _dimensions = VideoDimensions(info.width, info.height); } + latestInfo = info; } // Equality operators diff --git a/lib/src/support/disposable.dart b/lib/src/support/disposable.dart index a395ad5..09743b5 100644 --- a/lib/src/support/disposable.dart +++ b/lib/src/support/disposable.dart @@ -19,17 +19,17 @@ mixin _Disposer { Future _dispose() async { if (!_isDisposed) { - logger.fine('[${objectId}] dispose()'); + logger.finer('[${objectId}] dispose()'); _isDisposed = true; if (_disposeFuncs.isNotEmpty) { - logger.fine( + logger.finer( '[$objectId] running ${_disposeFuncs.length} dispose funcs...'); // call dispose funcs in reverse order for (final _func in _disposeFuncs.reversed) { await _func(); } _disposeFuncs.clear(); - logger.fine('[$objectId] dispose complete.'); + logger.finer('[$objectId] dispose complete.'); } return true; } else { diff --git a/lib/src/types.dart b/lib/src/types.dart index 52927ae..d6fcf4f 100644 --- a/lib/src/types.dart +++ b/lib/src/types.dart @@ -11,16 +11,18 @@ typedef CancelListenFunc = Function(); /// Usually it's not recommended to change this. enum ProtocolVersion { v2, - v3, + v3, // Subscriber as primary v4, v5, + v6, // Session migration } /// Connection state type used throughout the SDK. enum ConnectionState { disconnected, - connected, + connecting, reconnecting, + connected, } /// Connection quality between the [Participant] and server. @@ -58,9 +60,10 @@ enum StreamState { active, } -enum CloseReason { - network, - // ... +enum DisconnectReason { + user, + peerConnection, + signal, } /// The reason why a track failed to publish. diff --git a/lib/src/utils.dart b/lib/src/utils.dart index a7c7007..64784f3 100644 --- a/lib/src/utils.dart +++ b/lib/src/utils.dart @@ -2,9 +2,10 @@ import 'dart:async'; import 'package:collection/collection.dart'; import 'package:device_info_plus/device_info_plus.dart'; +import 'package:flutter/foundation.dart'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'package:meta/meta.dart'; -import 'package:platform_detect/platform_detect.dart' as pd; +import 'package:web_browser_detect/web_browser_detect.dart' as bd; import './proto/livekit_models.pb.dart' as lk_models; import './support/native.dart'; @@ -21,18 +22,57 @@ extension UriExt on Uri { bool get isSecureScheme => ['https', 'wss'].contains(scheme); } +typedef RetryFuture = Future Function( + int triesLeft, + List errors, +); +typedef RetryCondition = bool Function( + int triesLeft, + List errors, +); + // Collection of state-less static methods class Utils { + /// Returns a [Future] that will retry [future] while it throws + /// for a maximum of [tries] times with [delay] in between. + /// If all the attempts throws, the future will throw a [List] of the + /// thrown objects by the [future]. + static Future retry( + RetryFuture future, { + + /// number of total tries (first try + retries) + int tries = 1, + Duration delay = const Duration(seconds: 1), + RetryCondition? retryCondition, + }) async { + List errors = []; + while (tries-- > 0) { + try { + return await future(tries, errors); + } catch (error) { + logger.fine('[Retry] Caught error ${error}...'); + errors.add(error); + if (!(retryCondition?.call(tries, errors) ?? true)) break; + } + if (tries > 0) { + logger.fine('[Retry] Waiting ${delay}...'); + await Future.delayed(delay); + } + } + throw errors; + } + // DeviceInfoPlugin caches internally static final _deviceInfoPlugin = DeviceInfoPlugin(); static Future _clientInfo() async { switch (lkPlatform()) { case PlatformType.web: + final browser = bd.Browser(); return lk_models.ClientInfo( - os: pd.operatingSystem.name.toLowerCase(), - browser: pd.browser.name.toLowerCase(), - browserVersion: pd.browser.version.canonicalizedVersion, + os: defaultTargetPlatform.name, + browser: browser.browserAgent.name, + browserVersion: browser.version, ); case PlatformType.windows: return lk_models.ClientInfo( diff --git a/pubspec.lock b/pubspec.lock index 1cdf0b0..b2cbb98 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -57,6 +57,13 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "3.0.1" + csslib: + dependency: transitive + description: + name: csslib + url: "https://pub.dartlang.org" + source: hosted + version: "0.17.1" dart_webrtc: dependency: "direct main" description: @@ -163,6 +170,13 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "0.8.1" + html: + dependency: transitive + description: + name: html + url: "https://pub.dartlang.org" + source: hosted + version: "0.15.0" http: dependency: "direct main" description: @@ -282,13 +296,6 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "3.1.0" - platform_detect: - dependency: "direct main" - description: - name: platform_detect - url: "https://pub.dartlang.org" - source: hosted - version: "2.0.3" plugin_platform_interface: dependency: transitive description: @@ -310,13 +317,6 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "2.0.1" - pub_semver: - dependency: transitive - description: - name: pub_semver - url: "https://pub.dartlang.org" - source: hosted - version: "2.1.0" sky_engine: dependency: transitive description: flutter @@ -385,6 +385,20 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "1.3.0" + universal_html: + dependency: transitive + description: + name: universal_html + url: "https://pub.dartlang.org" + source: hosted + version: "2.0.8" + universal_io: + dependency: transitive + description: + name: universal_io + url: "https://pub.dartlang.org" + source: hosted + version: "2.0.4" uuid: dependency: "direct main" description: @@ -399,6 +413,13 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "2.1.1" + web_browser_detect: + dependency: "direct main" + description: + name: web_browser_detect + url: "https://pub.dartlang.org" + source: hosted + version: "2.0.2" webrtc_interface: dependency: transitive description: @@ -412,7 +433,7 @@ packages: name: win32 url: "https://pub.dartlang.org" source: hosted - version: "2.3.4" + version: "2.3.6" xdg_directories: dependency: transitive description: diff --git a/pubspec.yaml b/pubspec.yaml index 6f8924c..66cb0c6 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -25,7 +25,7 @@ dependencies: flutter_webrtc: ^0.8.1 dart_webrtc: ^1.0.3 device_info_plus: ^3.2.1 - platform_detect: ^2.0.3 + web_browser_detect: ^2.0.2 dev_dependencies: flutter_test: diff --git a/test/livekit_client_test.dart b/test/livekit_client_test.dart deleted file mode 100644 index 16f13c7..0000000 --- a/test/livekit_client_test.dart +++ /dev/null @@ -1,5 +0,0 @@ -import 'package:flutter_test/flutter_test.dart'; - -void main() { - test('placeholder', () {}); -} diff --git a/test/utils_test.dart b/test/utils_test.dart new file mode 100644 index 0000000..0603580 --- /dev/null +++ b/test/utils_test.dart @@ -0,0 +1,42 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:livekit_client/src/utils.dart'; + +void main() { + group('retry', () { + // test if List of errors are thrown + test( + 'throw all and throw error list', + () => expect( + Utils.retry( + (triesLeft, _) => throw 'error-${triesLeft}', + tries: 3, + delay: Duration.zero, + ), + throwsA([ + 'error-2', + 'error-1', + 'error-0', + ]), + ), + ); + test( + 'throw once and return result', + () => expectLater( + Utils.retry( + (triesLeft, _) async { + expect( + triesLeft, + isNot(0), + reason: 'should be never 0 because returning on 1', + ); + if (triesLeft == 1) return 'result-${triesLeft}'; + throw 'error${triesLeft}'; + }, + tries: 3, + delay: Duration.zero, + ), + completion('result-1'), + ), + ); + }); +}