diff --git a/example/ios/Flutter/AppFrameworkInfo.plist b/example/ios/Flutter/AppFrameworkInfo.plist index 9367d48..8d4492f 100644 --- a/example/ios/Flutter/AppFrameworkInfo.plist +++ b/example/ios/Flutter/AppFrameworkInfo.plist @@ -21,6 +21,6 @@ CFBundleVersion 1.0 MinimumOSVersion - 8.0 + 9.0 diff --git a/example/ios/Podfile b/example/ios/Podfile index f99a34a..c6a6a5c 100644 --- a/example/ios/Podfile +++ b/example/ios/Podfile @@ -44,6 +44,10 @@ post_install do |installer| # target.build_configurations.each do |config| config.build_settings['IPHONEOS_DEPLOYMENT_TARGET'] = '12.1' + # + # Make it compile with M1 macs + # + config.build_settings["EXCLUDED_ARCHS[sdk=iphonesimulator*]"] = 'arm64' end end diff --git a/example/ios/Podfile.lock b/example/ios/Podfile.lock index 555d945..416537a 100644 --- a/example/ios/Podfile.lock +++ b/example/ios/Podfile.lock @@ -2,14 +2,14 @@ PODS: - Flutter (1.0.0) - flutter_webrtc (0.2.2): - Flutter - - GoogleWebRTC (= 1.1.31999) - Libyuv (= 1703) - - GoogleWebRTC (1.1.31999) + - WebRTC-SDK (= 92.4515.05) - Libyuv (1703) - path_provider (0.0.1): - Flutter - shared_preferences (0.0.1): - Flutter + - WebRTC-SDK (92.4515.05) DEPENDENCIES: - Flutter (from `Flutter`) @@ -19,8 +19,8 @@ DEPENDENCIES: SPEC REPOS: trunk: - - GoogleWebRTC - Libyuv + - WebRTC-SDK EXTERNAL SOURCES: Flutter: @@ -33,13 +33,13 @@ EXTERNAL SOURCES: :path: ".symlinks/plugins/shared_preferences/ios" SPEC CHECKSUMS: - Flutter: 434fef37c0980e73bb6479ef766c45957d4b510c - flutter_webrtc: 39898454258b54ba51996850d5da8d5d53bf1524 - GoogleWebRTC: b39a78c4f5cc6b0323415b9233db03a2faa7b0f0 + Flutter: 50d75fe2f02b26cc09d224853bb45737f8b3214a + flutter_webrtc: c0cb88c7cbd057e6e667ab1560e10c74e2ceb65b Libyuv: 5f79ced0ee66e60a612ca97de1e6ccacd187a437 - path_provider: abfe2b5c733d04e238b0d8691db0cfd63a27a93c - shared_preferences: af6bfa751691cdc24be3045c43ec037377ada40d + path_provider: d1e9807085df1f9cc9318206cd649dc0b76be3de + shared_preferences: 5033afbb22d372e15aff8ff766df9021b845f273 + WebRTC-SDK: 7c76a541dbbffb0fc212aeb9902ec45a43c23996 -PODFILE CHECKSUM: 6055d9653e1011c0b3b671abb92cdea979357e63 +PODFILE CHECKSUM: 82aed1035f46bfa5b522f0d0dbf4730f17ec65ff -COCOAPODS: 1.10.1 +COCOAPODS: 1.11.2 diff --git a/example/ios/Runner.xcodeproj/project.pbxproj b/example/ios/Runner.xcodeproj/project.pbxproj index 1a4d320..3a41195 100644 --- a/example/ios/Runner.xcodeproj/project.pbxproj +++ b/example/ios/Runner.xcodeproj/project.pbxproj @@ -331,6 +331,7 @@ DEBUG_INFORMATION_FORMAT = "dwarf-with-dsym"; ENABLE_NS_ASSERTIONS = NO; ENABLE_STRICT_OBJC_MSGSEND = YES; + EXCLUDED_ARCHS = ""; GCC_C_LANGUAGE_STANDARD = gnu99; GCC_NO_COMMON_BLOCKS = YES; GCC_WARN_64_TO_32_BIT_CONVERSION = YES; @@ -342,7 +343,6 @@ IPHONEOS_DEPLOYMENT_TARGET = 12.1; MTL_ENABLE_DEBUG_INFO = NO; SDKROOT = iphoneos; - SUPPORTED_PLATFORMS = iphoneos; TARGETED_DEVICE_FAMILY = "1,2"; VALIDATE_PRODUCT = YES; }; @@ -406,6 +406,7 @@ DEBUG_INFORMATION_FORMAT = dwarf; ENABLE_STRICT_OBJC_MSGSEND = YES; ENABLE_TESTABILITY = YES; + EXCLUDED_ARCHS = ""; GCC_C_LANGUAGE_STANDARD = gnu99; GCC_DYNAMIC_NO_PIC = NO; GCC_NO_COMMON_BLOCKS = YES; @@ -461,6 +462,7 @@ DEBUG_INFORMATION_FORMAT = "dwarf-with-dsym"; ENABLE_NS_ASSERTIONS = NO; ENABLE_STRICT_OBJC_MSGSEND = YES; + EXCLUDED_ARCHS = ""; GCC_C_LANGUAGE_STANDARD = gnu99; GCC_NO_COMMON_BLOCKS = YES; GCC_WARN_64_TO_32_BIT_CONVERSION = YES; @@ -472,7 +474,6 @@ IPHONEOS_DEPLOYMENT_TARGET = 12.1; MTL_ENABLE_DEBUG_INFO = NO; SDKROOT = iphoneos; - SUPPORTED_PLATFORMS = iphoneos; SWIFT_COMPILATION_MODE = wholemodule; SWIFT_OPTIMIZATION_LEVEL = "-O"; TARGETED_DEVICE_FAMILY = "1,2"; diff --git a/example/lib/exts.dart b/example/lib/exts.dart index 0cfd4c5..45a362f 100644 --- a/example/lib/exts.dart +++ b/example/lib/exts.dart @@ -2,6 +2,24 @@ import 'package:flutter/material.dart'; extension LKExampleExt on BuildContext { // + Future showPublishDialog() => showDialog( + context: this, + builder: (ctx) => AlertDialog( + title: const Text('Publish'), + content: const Text('Would you like to publish your Camera & Mic ?'), + actions: [ + TextButton( + onPressed: () => Navigator.pop(ctx, false), + child: const Text('NO'), + ), + TextButton( + onPressed: () => Navigator.pop(ctx, true), + child: const Text('YES'), + ), + ], + ), + ); + Future showErrorDialog(dynamic exception) => showDialog( context: this, builder: (ctx) => AlertDialog( @@ -74,7 +92,7 @@ extension LKExampleExt on BuildContext { context: this, builder: (ctx) => AlertDialog( title: const Text('Received data'), - content: Text('"${data}"'), + content: Text(data), actions: [ TextButton( onPressed: () => Navigator.pop(ctx, true), diff --git a/example/lib/main.dart b/example/lib/main.dart index 1b5a2aa..54ba4ef 100644 --- a/example/lib/main.dart +++ b/example/lib/main.dart @@ -1,10 +1,12 @@ import 'package:flutter/material.dart'; +import 'package:livekit_client/livekit_client.dart'; import 'package:livekit_example/theme.dart'; import 'package:logging/logging.dart'; import 'pages/connect.dart'; void main() { + print('This is a test for ${SignalTrickleEvent} test.'); // configure logs for debugging Logger.root.level = Level.FINE; Logger.root.onRecord.listen((record) { diff --git a/example/lib/pages/connect.dart b/example/lib/pages/connect.dart index 04a3992..3bbc960 100644 --- a/example/lib/pages/connect.dart +++ b/example/lib/pages/connect.dart @@ -62,6 +62,9 @@ class _ConnectPageState extends State { _busy = true; }); + // Save for next time + await _writePrefs(); + print('Connecting with url: ${_uriCtrl.text}, token: ${_tokenCtrl.text}...'); final room = await LiveKitClient.connect( @@ -74,9 +77,6 @@ class _ConnectPageState extends State { ), ); - // Save for next time - await _writePrefs(); - await Navigator.push( ctx, MaterialPageRoute(builder: (_) => RoomPage(room)), @@ -122,10 +122,14 @@ class _ConnectPageState extends State { mainAxisSize: MainAxisSize.min, children: [ TextField( + enableSuggestions: false, + autocorrect: false, controller: _uriCtrl, decoration: const InputDecoration(labelText: 'URL'), ), TextField( + enableSuggestions: false, + autocorrect: false, controller: _tokenCtrl, decoration: const InputDecoration(labelText: 'Token'), ), diff --git a/example/lib/pages/room.dart b/example/lib/pages/room.dart index 236fa43..76376d9 100644 --- a/example/lib/pages/room.dart +++ b/example/lib/pages/room.dart @@ -2,12 +2,12 @@ import 'dart:convert'; import 'dart:math' as math; import 'package:flutter/material.dart'; +import 'package:flutter/widgets.dart'; import 'package:livekit_client/livekit_client.dart'; -import 'package:provider/provider.dart'; +import '../exts.dart'; import '../widgets/controls.dart'; import '../widgets/participant.dart'; -import '../exts.dart'; class RoomPage extends StatefulWidget { // @@ -22,29 +22,51 @@ class RoomPage extends StatefulWidget { State createState() => _RoomPageState(); } -class _RoomPageState extends State with RoomDelegate { +class _RoomPageState extends State { // List participants = []; + late final _listener = EventsListener(widget.room.events); @override void initState() { super.initState(); - widget.room.delegate = this; - widget.room.addListener(_onChange); - _onConnected(); + widget.room.addListener(_onRoomDidUpdate); + _setUpListeners(); + _sortParticipants(); + WidgetsBinding.instance?.addPostFrameCallback((_) => _askPublish()); } @override void dispose() { - widget.room.delegate = null; - widget.room.removeListener(_onChange); + // always dispose listener + (() async { + widget.room.removeListener(_onRoomDidUpdate); + await _listener.dispose(); + await widget.room.dispose(); + })(); super.dispose(); } - void _onConnected() async { + void _setUpListeners() => _listener + ..on((_) => Navigator.pop(context)) + ..on((event) { + String decoded = 'Failed to decode'; + try { + decoded = utf8.decode(event.data); + } catch (_) { + print('Failed to decode: $_'); + } + context.showDataReceivedDialog(decoded); + }); + + void _askPublish() async { + final result = await context.showPublishDialog(); + if (result != true) return; // video will fail when running in ios simulator try { - final localVideo = await LocalVideoTrack.createCameraTrack(); // Defaults to camera + // Create video track + final localVideo = await LocalVideoTrack.createCameraTrack(); + // Try to publish the video await widget.room.localParticipant.publishVideoTrack( localVideo, // options: TrackPublishOptions( @@ -52,20 +74,22 @@ class _RoomPageState extends State with RoomDelegate { // videoEncoding: VideoParameters.presetQVGA169.encoding, // ), ); - } catch (e) { - print('could not publish video: $e'); + + // Create mic track + final localAudio = await LocalAudioTrack.create(); + // // Try to publish audio + await widget.room.localParticipant.publishAudioTrack(localAudio); + } catch (error) { + print('could not publish video: $error'); + await context.showErrorDialog(error); } - - final localAudio = await LocalAudioTrack.create(); - await widget.room.localParticipant.publishAudioTrack(localAudio); - sortParticipants(); } - void _onChange() { - sortParticipants(); + void _onRoomDidUpdate() { + _sortParticipants(); } - void sortParticipants() { + void _sortParticipants() { List participants = []; participants.addAll(widget.room.participants.values); // sort speakers for the grid @@ -106,48 +130,31 @@ class _RoomPageState extends State with RoomDelegate { }); } - @override - void onDataReceived(RemoteParticipant participant, List data) async { - await context.showDataReceivedDialog(utf8.decode(data)); - } - - @override - void onDisconnected() { - print('disconnected: $context'); - Navigator.pop(context); - } - @override Widget build(BuildContext context) => Scaffold( - // with a provider, any child/descendent widget can be updated if they - // are a Consumer of Room. - body: ChangeNotifierProvider.value( - value: widget.room, - child: Column( - children: [ - Expanded( - child: participants.isNotEmpty - ? ParticipantWidget(participants.first) - : Container()), - SizedBox( - height: 100, - child: ListView.builder( - scrollDirection: Axis.horizontal, - itemCount: math.max(0, participants.length - 1), - itemBuilder: (BuildContext context, int index) => Container( - width: 100, - height: 100, - padding: const EdgeInsets.all(2), - child: ParticipantWidget(participants[index + 1], quality: VideoQuality.LOW), - ), + body: Column( + children: [ + Expanded( + child: + participants.isNotEmpty ? ParticipantWidget(participants.first) : Container()), + SizedBox( + height: 100, + child: ListView.builder( + scrollDirection: Axis.horizontal, + itemCount: math.max(0, participants.length - 1), + itemBuilder: (BuildContext context, int index) => Container( + width: 100, + height: 100, + padding: const EdgeInsets.all(2), + child: ParticipantWidget(participants[index + 1], quality: VideoQuality.LOW), ), ), - SafeArea( - top: false, - child: ControlsWidget(widget.room), - ), - ], - ), + ), + SafeArea( + top: false, + child: ControlsWidget(widget.room), + ), + ], ), ); } diff --git a/example/lib/widgets/controls.dart b/example/lib/widgets/controls.dart index 194b9af..0ba975d 100644 --- a/example/lib/widgets/controls.dart +++ b/example/lib/widgets/controls.dart @@ -102,8 +102,8 @@ class _ControlsWidgetState extends State { // final lp = widget.room.localParticipant; - for (final tracks in lp.videoTracks) { - await lp.unpublishTrack(tracks.track!); + for (final track in lp.videoTracks) { + await lp.unpublishTrack(track.sid); } try { diff --git a/example/lib/widgets/participant.dart b/example/lib/widgets/participant.dart index 8e49e60..3226925 100644 --- a/example/lib/widgets/participant.dart +++ b/example/lib/widgets/participant.dart @@ -21,7 +21,7 @@ class ParticipantWidget extends StatefulWidget { State createState() => _ParticipantWidgetState(); } -class _ParticipantWidgetState extends State with ParticipantDelegate { +class _ParticipantWidgetState extends State { // TrackPublication? videoPub; TrackPublication? audioPub; diff --git a/example/pubspec.lock b/example/pubspec.lock index b2c64ef..68cbe27 100644 --- a/example/pubspec.lock +++ b/example/pubspec.lock @@ -56,7 +56,7 @@ packages: name: eva_icons_flutter url: "https://pub.dartlang.org" source: hosted - version: "3.0.0" + version: "3.0.2" fake_async: dependency: transitive description: @@ -110,9 +110,11 @@ packages: flutter_webrtc: dependency: transitive description: - name: flutter_webrtc - url: "https://pub.dartlang.org" - source: hosted + path: "." + ref: use-custom-webrtc-build + resolved-ref: "4942e7faec2e5775d35c42e22c2929ca6ca53769" + url: "https://github.com/livekit/flutter-webrtc" + source: git version: "0.6.7" google_fonts: dependency: "direct main" @@ -197,14 +199,14 @@ packages: name: path_provider url: "https://pub.dartlang.org" source: hosted - version: "2.0.3" + version: "2.0.5" path_provider_linux: dependency: transitive description: name: path_provider_linux url: "https://pub.dartlang.org" source: hosted - version: "2.0.2" + version: "2.1.0" path_provider_macos: dependency: transitive description: @@ -274,7 +276,7 @@ packages: name: shared_preferences url: "https://pub.dartlang.org" source: hosted - version: "2.0.7" + version: "2.0.8" shared_preferences_linux: dependency: transitive description: @@ -391,7 +393,7 @@ packages: name: win32 url: "https://pub.dartlang.org" source: hosted - version: "2.2.8" + version: "2.2.9" xdg_directories: dependency: transitive description: @@ -400,5 +402,5 @@ packages: source: hosted version: "0.2.0" sdks: - dart: ">=2.13.0 <3.0.0" - flutter: ">=2.0.0" + dart: ">=2.14.0 <3.0.0" + flutter: ">=2.5.0" diff --git a/lib/livekit_client.dart b/lib/livekit_client.dart index 4b11945..7c53d2b 100644 --- a/lib/livekit_client.dart +++ b/lib/livekit_client.dart @@ -2,7 +2,9 @@ library livekit_client; export 'src/errors.dart'; +export 'src/events.dart'; export 'src/livekit.dart'; +export 'src/managers/event.dart'; export 'src/options.dart'; export 'src/participant/local_participant.dart'; export 'src/participant/local_participant.dart'; diff --git a/lib/src/classes/change_notifier.dart b/lib/src/classes/change_notifier.dart new file mode 100644 index 0000000..b8b4f53 --- /dev/null +++ b/lib/src/classes/change_notifier.dart @@ -0,0 +1,32 @@ +import 'package:flutter/material.dart'; +import 'package:livekit_client/src/logger.dart'; + +// dispose safe change notifier +abstract class LKChangeNotifier extends ChangeNotifier { + bool _disposed = false; + bool get isDisposed => _disposed; + + @override + void dispose() { + _disposed = true; + super.dispose(); + } + + @override + void addListener(VoidCallback listener) { + if (_disposed) { + logger.warning('calling addListener on a disposed ChangeNotifier'); + return; + } + super.addListener(listener); + } + + @override + void removeListener(VoidCallback listener) { + if (_disposed) { + logger.warning('calling removeListener on a disposed ChangeNotifier'); + return; + } + super.removeListener(listener); + } +} diff --git a/lib/src/constants.dart b/lib/src/constants.dart new file mode 100644 index 0000000..54039be --- /dev/null +++ b/lib/src/constants.dart @@ -0,0 +1,7 @@ +class Timeouts { + static const connection = Duration(seconds: 5); + static const debounce = Duration(milliseconds: 100); + static const publish = Duration(seconds: 3); + static const iceConnection = Duration(seconds: 5); + static const iceRestart = Duration(seconds: 10); +} diff --git a/lib/src/events.dart b/lib/src/events.dart index f11fffa..835f60c 100644 --- a/lib/src/events.dart +++ b/lib/src/events.dart @@ -1,120 +1,257 @@ import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; +import 'package:livekit_client/livekit_client.dart'; +import 'package:meta/meta.dart'; +import 'participant/participant.dart'; +import 'participant/remote_participant.dart'; import 'proto/livekit_models.pb.dart' as lk_models; +import 'proto/livekit_rtc.pb.dart' as lk_rtc; +import 'track/remote_track_publication.dart'; +import 'track/track.dart'; +import 'types.dart'; abstract class LiveKitEvent {} -abstract class RoomEvent implements LiveKitEvent { - const RoomEvent(); +abstract class RoomEvent implements LiveKitEvent {} + +abstract class ParticipantEvent implements LiveKitEvent {} + +abstract class TrackEvent implements LiveKitEvent {} + +abstract class EngineEvent implements LiveKitEvent {} + +abstract class SignalEvent implements LiveKitEvent {} + +/// When the connection to the server has been interrupted and it's attempting +/// to reconnect. +/// Emitted by [Room]. +class RoomReconnectingEvent with RoomEvent { + const RoomReconnectingEvent(); } -abstract class ParticipantEvent implements LiveKitEvent { - const ParticipantEvent(); +/// Connection to room is re-established. All existing state is preserved. +/// Emitted by [Room]. +class RoomReconnectedEvent with RoomEvent { + const RoomReconnectedEvent(); } -abstract class EngineEvent implements LiveKitEvent { - const EngineEvent(); +/// Disconnected from the room +/// Emitted by [Room]. +class RoomDisconnectedEvent with RoomEvent { + const RoomDisconnectedEvent(); } -abstract class TrackEvent implements LiveKitEvent { - const TrackEvent(); +/// When a new [RemoteParticipant] joins *after* the current participant has connected +/// It will not fire for participants that are already in the room +/// Emitted by [Room]. +class ParticipantConnectedEvent with RoomEvent { + final RemoteParticipant participant; + const ParticipantConnectedEvent({ + required this.participant, + }); +} + +/// When a [RemoteParticipant] leaves the room +/// Emitted by [Room]. +class ParticipantDisconnectedEvent with RoomEvent { + final RemoteParticipant participant; + const ParticipantDisconnectedEvent({ + required this.participant, + }); +} + +/// Active speakers changed. List of speakers are ordered by their audio level. +/// loudest speakers first. This will include the [LocalParticipant] too. +class ActiveSpeakersChangedEvent with RoomEvent { + final List speakers; + const ActiveSpeakersChangedEvent({ + required this.speakers, + }); +} + +class AudioPlaybackChangedEvent with RoomEvent { + const AudioPlaybackChangedEvent(); +} + +/// When a new [Track] is published to [Room] *after* the current participant has +/// joined. It will not fire for tracks that are already published. +/// Emitted by [Room] and [RemoteParticipant]. +class TrackPublishedEvent with RoomEvent, ParticipantEvent { + final RemoteParticipant participant; + final RemoteTrackPublication publication; + const TrackPublishedEvent({ + required this.participant, + required this.publication, + }); +} + +/// The participant has unpublished one of their [Track]. +/// Emitted by [Room] and [RemoteParticipant]. +class TrackUnpublishedEvent with RoomEvent, ParticipantEvent { + final RemoteParticipant participant; + final RemoteTrackPublication publication; + const TrackUnpublishedEvent({ + required this.participant, + required this.publication, + }); +} + +/// [LocalParticipant] has subscribed to a new track published by a +/// [RemoteParticipant]. +/// Emitted by [Room] and [RemoteParticipant]. +class TrackSubscribedEvent with RoomEvent, ParticipantEvent { + final RemoteParticipant participant; + final Track track; + final RemoteTrackPublication publication; + const TrackSubscribedEvent({ + required this.participant, + required this.track, + required this.publication, + }); +} + +@internal +class ParticipantInfoUpdatedEvent with ParticipantEvent { + final RemoteParticipant participant; + const ParticipantInfoUpdatedEvent({ + required this.participant, + }); +} + +/// An error has occured during track subscription. +/// Emitted by [Room] and [RemoteParticipant]. +class TrackSubscriptionExceptionEvent with RoomEvent, ParticipantEvent { + final RemoteParticipant participant; + final String? sid; + final TrackSubscribeFailReason reason; + const TrackSubscriptionExceptionEvent({ + required this.participant, + this.sid, + required this.reason, + }); +} + +/// The [LocalParticipant] has unsubscribed from a track published by a +/// [RemoteParticipant]. This event is fired when the track was unpublished. +/// Emitted by [Room] and [RemoteParticipant]. +class TrackUnsubscribedEvent with RoomEvent, ParticipantEvent { + final RemoteParticipant participant; + final Track track; + final RemoteTrackPublication publication; + const TrackUnsubscribedEvent({ + required this.participant, + required this.track, + required this.publication, + }); +} + +/// A Participant has muted one of the track. +/// Emitted on [RemoteParticipant] and [LocalParticipant]. +class TrackMutedEvent with RoomEvent, ParticipantEvent { + final Participant participant; + final TrackPublication track; + const TrackMutedEvent({ + required this.participant, + required this.track, + }); +} + +/// This participant has unmuted one of their tracks +/// Emitted on [RemoteParticipant] and [LocalParticipant]. +class TrackUnmutedEvent with RoomEvent, ParticipantEvent { + final Participant participant; + final TrackPublication track; + const TrackUnmutedEvent({ + required this.participant, + required this.track, + }); } // -// Room events +// common events for both Room/Participant. // -class RoomReconnectingEvent extends RoomEvent {} -class RoomReconnectedEvent extends RoomEvent {} +/// Participant metadata is a simple way for app-specific state to be pushed to +/// all users. When RoomService.UpdateParticipantMetadata is called to change a +/// [Participant]'s state, *all* [Participant]s in the room will fire this event. +/// Emitted on [Participant]. +class ParticipantMetadataUpdatedEvent with RoomEvent, ParticipantEvent { + final Participant participant; + const ParticipantMetadataUpdatedEvent({ + required this.participant, + }); +} -class RoomDisconnectedEvent extends RoomEvent {} +/// Data received from [RemoteParticipant]. +/// Data packets provides the ability to use LiveKit to send/receive arbitrary +/// payloads. +/// Emitted on [Room] and [RemoteParticipant]. +class DataReceivedEvent with RoomEvent, ParticipantEvent { + /// Sender of the data. This may be null if data is sent from Server API. + final RemoteParticipant? participant; + final List data; + const DataReceivedEvent({ + required this.participant, + required this.data, + }); +} -class RoomParticipantConnectedEvent extends RoomEvent {} - -class RoomParticipantDisconnectedEvent extends RoomEvent {} - -class RoomTrackPublishedEvent extends RoomEvent {} - -class RoomTrackSubscribedEvent extends RoomEvent {} - -class RoomTrackSubscriptionFailedEvent extends RoomEvent {} - -class RoomTrackUnpublishedEvent extends RoomEvent {} - -class RoomTrackUnsubscribedEvent extends RoomEvent {} - -class RoomTrackMutedEvent extends RoomEvent {} - -class RoomTrackUnmutedEvent extends RoomEvent {} - -class RoomActiveSpeakerChangedEvent extends RoomEvent {} - -class RoomMetadataChangedEvent extends RoomEvent {} - -class RoomDataReceivedEvent extends RoomEvent {} - -class RoomAudioPlaybackChangedEvent extends RoomEvent {} - -// -// Participant events -// -class ParticipantTrackPublishedEvent extends ParticipantEvent {} - -class ParticipantTrackSubscribedEvent extends ParticipantEvent {} - -class ParticipantTrackSubscriptionFailedEvent extends ParticipantEvent {} - -class ParticipantTrackUnpublishedEvent extends ParticipantEvent {} - -class ParticipantTrackUnsubscribedEvent extends ParticipantEvent {} - -class ParticipantTrackMutedEvent extends ParticipantEvent {} - -class ParticipantTrackUnmutedEvent extends ParticipantEvent {} - -class ParticipantMetadataChangedEvent extends ParticipantEvent {} - -class ParticipantDataReceivedEvent extends ParticipantEvent {} - -class ParticipantSpeakingChangedEvent extends ParticipantEvent {} +/// The participant's isSpeaking property has changed +/// Emitted on [Participant]. +class SpeakingChangedEvent with RoomEvent, ParticipantEvent { + final Participant participant; + final bool speaking; + const SpeakingChangedEvent({ + required this.participant, + required this.speaking, + }); +} // // Engine events // -class EngineConnectedEvent extends EngineEvent {} +class EngineConnectedEvent with EngineEvent { + const EngineConnectedEvent(); +} -class EngineDisconnectedEvent extends EngineEvent {} +class EngineDisconnectedEvent with EngineEvent { + const EngineDisconnectedEvent(); +} -class EngineReconnectingEvent extends EngineEvent {} +class EngineReconnectingEvent with EngineEvent { + const EngineReconnectingEvent(); +} -class EngineReconnectedEvent extends EngineEvent {} +class EngineReconnectedEvent with EngineEvent { + const EngineReconnectedEvent(); +} -class EngineParticipantUpdateEvent extends EngineEvent { +class EngineParticipantUpdateEvent with EngineEvent { final List participants; const EngineParticipantUpdateEvent({ required this.participants, }); } -class EngineMediaTrackAddedEvent extends EngineEvent { +class EngineTrackAddedEvent with EngineEvent { final rtc.MediaStreamTrack track; - final rtc.MediaStream? stream; + final rtc.MediaStream stream; final rtc.RTCRtpReceiver? receiver; - const EngineMediaTrackAddedEvent({ + const EngineTrackAddedEvent({ required this.track, required this.stream, required this.receiver, }); } -class EngineSpeakersUpdateEvent extends EngineEvent { +class EngineSpeakersUpdateEvent with EngineEvent { final List speakers; const EngineSpeakersUpdateEvent({ required this.speakers, }); } -class EngineDataPacketReceivedEvent extends EngineEvent { +class EngineDataPacketReceivedEvent with EngineEvent { final lk_models.UserPacket packet; final lk_models.DataPacket_Kind kind; const EngineDataPacketReceivedEvent({ @@ -123,7 +260,7 @@ class EngineDataPacketReceivedEvent extends EngineEvent { }); } -class EngineRemoteMuteChangedEvent extends EngineEvent { +class EngineRemoteMuteChangedEvent with EngineEvent { final String sid; final bool muted; const EngineRemoteMuteChangedEvent({ @@ -133,7 +270,7 @@ class EngineRemoteMuteChangedEvent extends EngineEvent { } // added -abstract class EngineIceStateUpdatedEvent implements EngineEvent { +abstract class EngineIceStateUpdatedEvent with EngineEvent { final rtc.RTCIceConnectionState iceState; final bool isPrimary; const EngineIceStateUpdatedEvent({ @@ -166,16 +303,101 @@ class EnginePublisherIceStateUpdatedEvent extends EngineIceStateUpdatedEvent { // Track events // -class TrackMessageEvent extends TrackEvent {} +class TrackMessageEvent with TrackEvent { + const TrackMessageEvent(); +} -class TrackMutedEvent extends TrackEvent {} +class TrackUpdateSettingsEvent with TrackEvent { + const TrackUpdateSettingsEvent(); +} -class TrackUnmutedEvent extends TrackEvent {} +class TrackUpdateSubscriptionEvent with TrackEvent { + const TrackUpdateSubscriptionEvent(); +} -class TrackUpdateSettingsEvent extends TrackEvent {} +class TrackAudioPlaybackStartedEvent with TrackEvent { + const TrackAudioPlaybackStartedEvent(); +} -class TrackUpdateSubscriptionEvent extends TrackEvent {} +class TrackAudioPlaybackFailedEvent with TrackEvent { + const TrackAudioPlaybackFailedEvent(); +} -class TrackAudioPlaybackStartedEvent extends TrackEvent {} +// +// Signal events +// +class SignalConnectedEvent with SignalEvent { + final lk_rtc.JoinResponse response; + const SignalConnectedEvent({ + required this.response, + }); +} -class TrackAudioPlaybackFailedEvent extends TrackEvent {} +class SignalCloseEvent with SignalEvent { + final CloseReason? reason; + const SignalCloseEvent({ + this.reason, + }); +} + +class SignalOfferEvent with SignalEvent { + final rtc.RTCSessionDescription sd; + const SignalOfferEvent({ + required this.sd, + }); +} + +class SignalAnswerEvent with SignalEvent { + final rtc.RTCSessionDescription sd; + const SignalAnswerEvent({ + required this.sd, + }); +} + +class SignalTrickleEvent with SignalEvent { + final rtc.RTCIceCandidate candidate; + final lk_rtc.SignalTarget target; + const SignalTrickleEvent({ + required this.candidate, + required this.target, + }); +} + +class SignalParticipantUpdateEvent with SignalEvent { + final List updates; + const SignalParticipantUpdateEvent({ + required this.updates, + }); +} + +class SignalLocalTrackPublishedEvent with SignalEvent { + final String cid; + final lk_models.TrackInfo track; + const SignalLocalTrackPublishedEvent({ + required this.cid, + required this.track, + }); +} + +class SignalActiveSpeakersChangedEvent with SignalEvent { + final List speakers; + const SignalActiveSpeakersChangedEvent({ + required this.speakers, + }); +} + +class SignalLeaveEvent with SignalEvent { + final bool canReconnect; + const SignalLeaveEvent({ + required this.canReconnect, + }); +} + +class SignalMuteTrackEvent with SignalEvent { + final String sid; + final bool muted; + const SignalMuteTrackEvent({ + required this.sid, + required this.muted, + }); +} diff --git a/lib/src/extensions.dart b/lib/src/extensions.dart index 21f5f3a..7c464ca 100644 --- a/lib/src/extensions.dart +++ b/lib/src/extensions.dart @@ -2,43 +2,51 @@ import 'dart:convert'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; -import 'proto/livekit_rtc.pb.dart' as lk_rtc; +import 'events.dart'; +import 'managers/event.dart'; import 'proto/livekit_models.pb.dart' as lk_models; - +import 'proto/livekit_rtc.pb.dart' as lk_rtc; import 'types.dart'; +extension DataPacketKindExt on lk_models.DataPacket_Kind { + Reliability toSDKType() => { + lk_models.DataPacket_Kind.RELIABLE: Reliability.reliable, + lk_models.DataPacket_Kind.LOSSY: Reliability.lossy, + }[this]!; +} + +extension LiveKitEventExt on Iterable> { + void emit(LiveKitEvent event) => forEach((emitter) => emitter.emit(event)); +} + +extension ICEServerExt on lk_rtc.ICEServer { + RTCIceServer toSDKType() => RTCIceServer( + urls: urls, + username: username.isNotEmpty ? username : null, + credential: credential.isNotEmpty ? username : null, + ); +} + extension IterableExt on Iterable { E? elementAtOrNull(int index) => (index >= 0 && index < length) ? elementAt(index) : null; } -extension RTCIceConnectionStateExt on rtc.RTCIceConnectionState { - bool isConnected() => [ - rtc.RTCIceConnectionState.RTCIceConnectionStateConnected, - rtc.RTCIceConnectionState.RTCIceConnectionStateCompleted, - ].contains(this); -} - extension ObjectExt on Object { String get objectId => '${runtimeType}#${hashCode}'; } -extension RTCIceTransportPolicyExt on RTCIceTransportPolicy { +extension ProtocolVersionExt on ProtocolVersion { String toStringValue() => { - RTCIceTransportPolicy.all: 'all', - RTCIceTransportPolicy.relay: 'relay', + ProtocolVersion.protocol2: '2', + ProtocolVersion.protocol3: '3', }[this]!; } -extension SessionDescriptionExt on lk_rtc.SessionDescription { - rtc.RTCSessionDescription toSDKType() { - return rtc.RTCSessionDescription(sdp, type); - } -} - -extension RTCSessionDescriptionExt on rtc.RTCSessionDescription { - lk_rtc.SessionDescription toSDKType() { - return lk_rtc.SessionDescription(type: type, sdp: sdp); - } +extension ReliabilityExt on Reliability { + lk_models.DataPacket_Kind toPBType() => { + Reliability.reliable: lk_models.DataPacket_Kind.RELIABLE, + Reliability.lossy: lk_models.DataPacket_Kind.LOSSY, + }[this]!; } extension RTCIceCandidateExt on rtc.RTCIceCandidate { @@ -54,26 +62,30 @@ extension RTCIceCandidateExt on rtc.RTCIceCandidate { String toJson() => json.encode(toMap()); } -extension ICEServerExt on lk_rtc.ICEServer { - RTCIceServer toSDKType() => RTCIceServer( - urls: urls, - username: username.isNotEmpty ? username : null, - credential: credential.isNotEmpty ? username : null, - ); +extension RTCIceConnectionStateExt on rtc.RTCIceConnectionState { + bool isConnected() => [ + rtc.RTCIceConnectionState.RTCIceConnectionStateConnected, + rtc.RTCIceConnectionState.RTCIceConnectionStateCompleted, + ].contains(this); +} + +extension RTCIceTransportPolicyExt on RTCIceTransportPolicy { + String toStringValue() => { + RTCIceTransportPolicy.all: 'all', + RTCIceTransportPolicy.relay: 'relay', + }[this]!; } // not so neat to directly expose protobuf types so we // define our own types (and convert methods) -extension DataPacketKindExt on lk_models.DataPacket_Kind { - Reliability toSDKType() => { - lk_models.DataPacket_Kind.RELIABLE: Reliability.reliable, - lk_models.DataPacket_Kind.LOSSY: Reliability.lossy, - }[this]!; +extension RTCSessionDescriptionExt on rtc.RTCSessionDescription { + lk_rtc.SessionDescription toSDKType() { + return lk_rtc.SessionDescription(type: type, sdp: sdp); + } } -extension ReliabilityExt on Reliability { - lk_models.DataPacket_Kind toPBType() => { - Reliability.reliable: lk_models.DataPacket_Kind.RELIABLE, - Reliability.lossy: lk_models.DataPacket_Kind.LOSSY, - }[this]!; +extension SessionDescriptionExt on lk_rtc.SessionDescription { + rtc.RTCSessionDescription toSDKType() { + return rtc.RTCSessionDescription(sdp, type); + } } diff --git a/lib/src/livekit.dart b/lib/src/livekit.dart index 57bc64e..a1a345c 100644 --- a/lib/src/livekit.dart +++ b/lib/src/livekit.dart @@ -11,12 +11,10 @@ class LiveKitClient { String url, String token, { ConnectOptions? options, - }) { - final room = Room(); - return room.connect( - url, - token, - options: options, - ); - } + }) => + Room.connect( + url, + token, + options: options, + ); } diff --git a/lib/src/managers/delay.dart b/lib/src/managers/delay.dart index dca4e1a..8fe30ce 100644 --- a/lib/src/managers/delay.dart +++ b/lib/src/managers/delay.dart @@ -1,6 +1,3 @@ -// -// -// import 'package:async/async.dart'; class CancelableDelayManager { diff --git a/lib/src/managers/event.dart b/lib/src/managers/event.dart index a24e4fd..022a715 100644 --- a/lib/src/managers/event.dart +++ b/lib/src/managers/event.dart @@ -1,19 +1,24 @@ import 'dart:async'; import 'package:flutter/material.dart'; +import 'package:synchronized/synchronized.dart' as sync; import '../errors.dart'; -import '../events.dart'; import '../extensions.dart'; import '../logger.dart'; import '../types.dart'; // Type-safe, multi-listenable, dispose safe event handling +// TODO: Move to a separate package -class EventsEmitter extends EventsListenable { +class EventsEmitter extends EventsListenable { // suppport for multiple event listeners final streamCtrl = StreamController.broadcast(sync: false); + EventsEmitter({ + bool listenSynchronized = false, + }) : super(synchronized: listenSynchronized); + @override EventsEmitter get emitter => this; @@ -25,6 +30,7 @@ class EventsEmitter extends EventsListenable { } @override + @mustCallSuper Future dispose() async { await streamCtrl.close(); await super.dispose(); @@ -32,21 +38,31 @@ class EventsEmitter extends EventsListenable { } // for listening only -class EventsListener extends EventsListenable { +class EventsListener extends EventsListenable { @override final EventsEmitter emitter; - EventsListener({ - required this.emitter, - }); + EventsListener( + this.emitter, { + bool synchronized = false, + }) : super( + synchronized: synchronized, + ); } // ensures all listeners will close on dispose -abstract class EventsListenable { +abstract class EventsListenable { // the emitter to listen to EventsEmitter get emitter; + + bool synchronized; // keep track of listeners to cancel later final _listeners = >[]; + final _syncLock = sync.Lock(); + + EventsListenable({ + required this.synchronized, + }); @mustCallSuper Future dispose() async { @@ -58,8 +74,19 @@ abstract class EventsListenable { } // listens to all events, guaranteed to be cancelled on dispose - CancelListenFunc listen(Function(T) onEvent) { - final listener = emitter.streamCtrl.stream.listen(onEvent); + CancelListenFunc listen(FutureOr Function(T) onEvent) { + // + FutureOr Function(T) _func = onEvent; + if (synchronized) { + // ensure `onEvent` will trigger one by one (waits for previous `onEvent` to complete) + _func = (event) async { + await _syncLock.synchronized(() async { + await onEvent(event); + }); + }; + } + + final listener = emitter.streamCtrl.stream.listen(_func); _listeners.add(listener); // make a cancel func to cancel listening and remove from list in 1 call @@ -74,34 +101,34 @@ abstract class EventsListenable { // convenience method to listen & filter a specific event type CancelListenFunc on( - Function(E) then, { + FutureOr Function(E) then, { bool Function(E)? filter, }) => - listen((event) { + listen((event) async { // event must be E if (event is! E) return; // filter must be true (if filter is used) - if (filter != null && !filter(event as E)) return; + if (filter != null && !filter(event)) return; // cast to E - then(event as E); + await then(event); }); // waits for a specific event type - Future waitFor({ + Future waitFor({ required Duration duration, bool Function(E)? filter, - FutureOr Function()? onTimeout, + FutureOr Function()? onTimeout, }) async { - final completer = Completer(); + final completer = Completer(); final _cancelFunc = on( - (event) => completer.complete(), + (event) => completer.complete(event), filter: filter, ); try { // wait to complete with timeout - await completer.future.timeout( + return await completer.future.timeout( duration, onTimeout: onTimeout ?? () => throw TimeoutException(), ); diff --git a/lib/src/participant/local_participant.dart b/lib/src/participant/local_participant.dart index 2cea886..86896de 100644 --- a/lib/src/participant/local_participant.dart +++ b/lib/src/participant/local_participant.dart @@ -2,15 +2,16 @@ import 'package:flutter/foundation.dart'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import '../errors.dart'; +import '../events.dart'; import '../extensions.dart'; import '../logger.dart'; +import '../managers/event.dart'; import '../options.dart'; import '../proto/livekit_models.pb.dart' as lk_models; import '../rtc_engine.dart'; import '../track/local_audio_track.dart'; import '../track/local_track_publication.dart'; import '../track/local_video_track.dart'; -import '../track/track.dart'; import '../track/track_publication.dart'; import '../types.dart'; import '../utils.dart'; @@ -25,8 +26,13 @@ class LocalParticipant extends Participant { required RTCEngine engine, required lk_models.ParticipantInfo info, this.defaultPublishOptions, + required EventsEmitter roomEvents, }) : _engine = engine, - super(info.sid, info.identity) { + super( + info.sid, + info.identity, + roomEvents: roomEvents, + ) { updateFromInfo(info); } @@ -135,21 +141,25 @@ class LocalParticipant extends Participant { } /// Unpublish a track that's already published - Future unpublishTrack(Track track) async { - final existing = tracks.values.where((element) => element.track == track); - if (existing.isEmpty) return; + @override + Future unpublishTrack(String trackSid, {bool notify = false}) async { + logger.finer('Unpublish track sid: $trackSid, notify: $notify'); + final pub = trackPublications.remove(trackSid); + if (pub is! LocalTrackPublication) return; - final pub = existing.first; + // final existing = tracks.values.where((element) => element.track == track); + // if (existing.isEmpty) return; + // final pub = existing.first; + final track = pub.track; + if (track != null) { + await track.stop(); - await track.stop(); - - final sender = track.transceiver?.sender; - if (sender != null) { - await engine.publisher?.pc.removeTrack(sender); - await engine.negotiate(); + final sender = track.transceiver?.sender; + if (sender != null) { + await engine.publisher?.pc.removeTrack(sender); + await engine.negotiate(); + } } - - tracks.remove(pub.sid); } /// Publish a new data payload to the room. diff --git a/lib/src/participant/participant.dart b/lib/src/participant/participant.dart index bf4c1fa..32ed8be 100644 --- a/lib/src/participant/participant.dart +++ b/lib/src/participant/participant.dart @@ -1,50 +1,16 @@ +import 'package:collection/collection.dart'; import 'package:flutter/foundation.dart'; +import 'package:meta/meta.dart'; +import '../classes/change_notifier.dart'; import '../events.dart'; +import '../extensions.dart'; +import '../logger.dart'; import '../managers/event.dart'; import '../proto/livekit_models.pb.dart' as lk_models; -import '../track/remote_track_publication.dart'; -import '../track/track.dart'; import '../track/track_publication.dart'; import 'remote_participant.dart'; -/// Callbacks for participant changes -mixin ParticipantDelegate { - /// The participant's metadata has changed - void onMetadataChanged(Participant participant) {} - - /// The participant's isSpeaking property has changed - void onSpeakingChanged(Participant participant, bool speaking) {} - - /// This participant has muted one of their tracks - void onTrackMuted(Participant participant, TrackPublication publication) {} - - /// This participant has unmuted one of their tracks - void onTrackUnmuted(Participant participant, TrackPublication publication) {} - - /// This participant has published a new [Track] to the [Room]. - void onTrackPublished(RemoteParticipant participant, RemoteTrackPublication publication) {} - - /// This participant has unpublished one of their [Track]. - void onTrackUnpublished(RemoteParticipant participant, RemoteTrackPublication publication) {} - - /// The [LocalParticipant] has subscribed to a new track published by this - /// [RemoteParticipant] - void onTrackSubscribed( - RemoteParticipant participant, Track track, RemoteTrackPublication publication) {} - - /// The [LocalParticipant] has unsubscribed from a track published by this - /// [RemoteParticipant]. This event is fired when the track was unpublished - void onTrackUnsubscribed( - RemoteParticipant participant, Track track, RemoteTrackPublication publication) {} - - /// Data received from this [RemoteParticipant]. - void onDataReceived(RemoteParticipant participant, List data) {} - - /// An error has occured during track subscription. - void onTrackSubscriptionFailed(RemoteParticipant participant, String sid, String? message) {} -} - /// Represents a Participant in the room, notifies changes via delegates as /// well as ChangeNotifier/providers. /// A change notification is triggered when @@ -52,15 +18,18 @@ mixin ParticipantDelegate { /// - mute status changed /// - added/removed subscribed tracks /// - metadata changed -class Participant extends ChangeNotifier { + +/// Base for [RemoteParticipant] and [LocalParticipant], +/// can not be instantiated directly. +abstract class Participant extends LKChangeNotifier { /// map of track sid => published track - Map tracks = {}; + final trackPublications = {}; /// audio level between 0-1, 1 being the loudest double audioLevel = 0; /// server assigned unique id - String sid; + final String sid; /// user-assigned identity String identity; @@ -71,16 +40,12 @@ class Participant extends ChangeNotifier { /// when the participant had last spoken DateTime? lastSpokeAt; - ParticipantDelegate? roomDelegate; - - /// delegate to receive participant callbacks - ParticipantDelegate? delegate; - lk_models.ParticipantInfo? _participantInfo; bool _isSpeaking = false; // suppport for multiple event listeners final events = EventsEmitter(); + final EventsEmitter roomEvents; /// when the participant joined the room DateTime get joinedAt { @@ -95,23 +60,40 @@ class Participant extends ChangeNotifier { bool get isSpeaking => _isSpeaking; /// true if participant is publishing an audio track and is muted - bool get isMuted { - if (audioTracks.isEmpty) return false; - return audioTracks.first.muted; - } + bool get isMuted => audioTracks.firstOrNull?.muted ?? true; bool get hasAudio => audioTracks.isNotEmpty; bool get hasVideo => videoTracks.isNotEmpty; /// tracks that are subscribed to - List get subscribedTracks => tracks.values.where((e) => e.subscribed).toList(); + List get subscribedTracks => + trackPublications.values.where((e) => e.subscribed).toList(); /// for internal use /// {@nodoc} + @internal bool get hasInfo => _participantInfo != null; - Participant(this.sid, this.identity); + Participant( + this.sid, + this.identity, { + required this.roomEvents, + }) { + // Any event emitted will trigger ChangeNotifier + events.listen((event) { + logger.fine('[ParticipantEvent] $event, will notifyListeners()'); + notifyListeners(); + }); + } + + @override + @mustCallSuper + Future dispose() async { + logger.fine('$objectId dispose()'); + await events.dispose(); + super.dispose(); + } /// for internal use /// {@nodoc} @@ -123,26 +105,29 @@ class Participant extends ChangeNotifier { if (speaking) { lastSpokeAt = DateTime.now(); } - delegate?.onSpeakingChanged(this, speaking); - roomDelegate?.onSpeakingChanged(this, speaking); - notifyListeners(); + + [events, roomEvents].emit(SpeakingChangedEvent( + participant: this, + speaking: speaking, + )); } void _setMetadata(String md) { final changed = _participantInfo?.metadata != md; metadata = md; if (changed) { - delegate?.onMetadataChanged(this); - roomDelegate?.onMetadataChanged(this); - notifyListeners(); + [events, roomEvents].emit(ParticipantMetadataUpdatedEvent( + participant: this, + )); } } /// for internal use /// {@nodoc} + @internal void updateFromInfo(lk_models.ParticipantInfo info) { identity = info.identity; - sid = info.sid; + // participantSid = info.sid; if (info.metadata.isNotEmpty) { _setMetadata(info.metadata); } @@ -151,23 +136,36 @@ class Participant extends ChangeNotifier { /// for internal use /// {@nodoc} - void muteChanged() { - notifyListeners(); - } - - /// for internal use - /// {@nodoc} + @internal void addTrackPublication(TrackPublication pub) { pub.track?.sid = pub.sid; - tracks[pub.sid] = pub; + trackPublications[pub.sid] = pub; } + + // Must implement + Future unpublishTrack(String trackSid, {bool notify = false}); + + Future unpublishAllTracks() async { + final _ = List.from(trackPublications.values); + for (final track in _) { + await unpublishTrack(track.sid); + } + } + + // Equality operators + // Object is considered equal when sid is equal + @override + int get hashCode => sid.hashCode; + + @override + bool operator ==(Object other) => other is Participant && sid == other.sid; } // Convenience extension extension ParticipantExt on Participant { List get videoTracks => - tracks.values.where((e) => e.kind == lk_models.TrackType.VIDEO).toList(); + trackPublications.values.where((e) => e.kind == lk_models.TrackType.VIDEO).toList(); List get audioTracks => - tracks.values.where((e) => e.kind == lk_models.TrackType.AUDIO).toList(); + trackPublications.values.where((e) => e.kind == lk_models.TrackType.AUDIO).toList(); } diff --git a/lib/src/participant/remote_participant.dart b/lib/src/participant/remote_participant.dart index 88b8c7b..fdcf8a1 100644 --- a/lib/src/participant/remote_participant.dart +++ b/lib/src/participant/remote_participant.dart @@ -1,12 +1,18 @@ import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; +import 'package:meta/meta.dart'; +import '../constants.dart'; +import '../events.dart'; +import '../extensions.dart'; import '../logger.dart'; +import '../managers/event.dart'; import '../proto/livekit_models.pb.dart' as lk_models; import '../signal_client.dart'; import '../track/audio_track.dart'; import '../track/remote_track_publication.dart'; import '../track/track.dart'; import '../track/video_track.dart'; +import '../types.dart'; import 'participant.dart'; /// Represents other participant in the [Room]. @@ -18,140 +24,154 @@ class RemoteParticipant extends Participant { RemoteParticipant( this._client, String sid, - String identity, - ) : super(sid, identity); + String identity, { + required EventsEmitter roomEvents, + }) : super( + sid, + identity, + roomEvents: roomEvents, + ); RemoteParticipant.fromInfo( this._client, - lk_models.ParticipantInfo info, - ) : super(info.sid, info.identity) { + lk_models.ParticipantInfo info, { + required EventsEmitter roomEvents, + }) : super( + info.sid, + info.identity, + roomEvents: roomEvents, + ) { updateFromInfo(info); } RemoteTrackPublication? getTrackPublication(String sid) { - final pub = tracks[sid]; + final pub = trackPublications[sid]; if (pub is RemoteTrackPublication) return pub; } /// for internal use /// {@nodoc} - void addSubscribedMediaTrack( + @internal + Future addSubscribedMediaTrack( rtc.MediaStreamTrack mediaTrack, rtc.MediaStream stream, - String? sid, + String trackSid, ) async { - if (sid == null) { - const msg = 'addSubscribedMediaTrack received null sid'; - delegate?.onTrackSubscriptionFailed(this, '', msg); - roomDelegate?.onTrackSubscriptionFailed(this, '', msg); - return; - } + logger.fine('addSubscribedMediaTrack()'); - var pub = getTrackPublication(sid); + // If publication doesn't exist yet... + RemoteTrackPublication? pub = getTrackPublication(trackSid); if (pub == null) { - // we may have received the track prior to metadata. wait up to 3s - pub = await _waitForTrackPublication(sid, const Duration(seconds: 3)); - if (pub == null) { - const msg = 'no track metadata found'; - delegate?.onTrackSubscriptionFailed(this, sid, msg); - roomDelegate?.onTrackSubscriptionFailed(this, sid, msg); - return; - } + logger.fine('addSubscribedMediaTrack() pub is null, will wait...'); + // Wait for the metadata to arrive + final event = await events.waitFor( + filter: (event) => event.participant == this && event.publication.sid == trackSid, + duration: Timeouts.publish, + onTimeout: () => throw TrackSubscriptionExceptionEvent( + participant: this, + sid: trackSid, + reason: TrackSubscribeFailReason.notTrackMetadataFound, + ), + ); + pub = event.publication; + logger.fine('addSubscribedMediaTrack() did receive pub'); } - Track? track; + // Check if track type is supported, throw if not. + if (![lk_models.TrackType.AUDIO, lk_models.TrackType.VIDEO].contains(pub.kind)) { + throw TrackSubscriptionExceptionEvent( + participant: this, + sid: trackSid, + reason: TrackSubscribeFailReason.unsupportedTrackType, + ); + } + + // create Track + final Track track; if (pub.kind == lk_models.TrackType.AUDIO) { + // audio track final audioTrack = AudioTrack(pub.name, mediaTrack, stream); audioTrack.start(); track = audioTrack; - } else if (pub.kind == lk_models.TrackType.VIDEO) { - track = VideoTrack(pub.name, mediaTrack, stream); } else { - final msg = 'unsupported track type ${pub.kind}'; - delegate?.onTrackSubscriptionFailed(this, sid, msg); - roomDelegate?.onTrackSubscriptionFailed(this, sid, msg); - return; + // video track + track = VideoTrack(pub.name, mediaTrack, stream); } pub.track = track; addTrackPublication(pub); - delegate?.onTrackSubscribed(this, track, pub); - roomDelegate?.onTrackSubscribed(this, track, pub); - notifyListeners(); + [events, roomEvents].emit(TrackSubscribedEvent( + participant: this, + track: track, + publication: pub, + )); } /// for internal use /// {@nodoc} @override - void updateFromInfo(lk_models.ParticipantInfo info) async { + @internal + Future updateFromInfo(lk_models.ParticipantInfo info) async { final hadInfo = hasInfo; super.updateFromInfo(info); // figuring out deltas between tracks - final validPubs = {}; - final newPubs = {}; - - for (final info in info.tracks) { - final sid = info.sid; - var pub = getTrackPublication(sid); + final newPubs = {}; + for (final trackInfo in info.tracks) { + RemoteTrackPublication? pub = getTrackPublication(trackInfo.sid); if (pub == null) { - pub = RemoteTrackPublication(info, this); - newPubs[sid] = pub; + pub = RemoteTrackPublication(trackInfo, this); + newPubs.add(pub); addTrackPublication(pub); } else { - pub.updateFromInfo(info); + pub.updateFromInfo(trackInfo); } - - validPubs[sid] = pub; } // notify listeners when it's not a new participant if (hadInfo) { - for (final pub in newPubs.values) { - delegate?.onTrackPublished(this, pub); - roomDelegate?.onTrackPublished(this, pub); + for (final pub in newPubs) { + final event = TrackPublishedEvent( + participant: this, + publication: pub, + ); + [events, roomEvents].emit(event); } } - // remove tracks - final removeTrackSids = - tracks.values.where((e) => !validPubs.containsKey(e.sid)).map((e) => e.sid).toList(); - - for (final sid in removeTrackSids) { - await unpublishTrack(sid, true); + // unpublish any track that is not in the info + final validSids = info.tracks.map((e) => e.sid); + final removeSids = + trackPublications.values.where((e) => !validSids.contains(e.sid)).map((e) => e.sid); + for (final sid in removeSids) { + await unpublishTrack(sid, notify: true); } } - Future unpublishTrack(String sid, [bool notify = false]) async { - logger.finer('Unpublish track sid: $sid, notify: $notify'); - final pub = tracks.remove(sid); - if (pub == null || pub is! RemoteTrackPublication) return; + @override + Future unpublishTrack(String trackSid, {bool notify = false}) async { + logger.finer('Unpublish track sid: $trackSid, notify: $notify'); + final pub = trackPublications.remove(trackSid); + if (pub is! RemoteTrackPublication) return; final track = pub.track; + // if has track if (track != null) { await track.stop(); - delegate?.onTrackUnsubscribed(this, track, pub); - roomDelegate?.onTrackUnsubscribed(this, track, pub); - notifyListeners(); + [events, roomEvents].emit(TrackUnsubscribedEvent( + participant: this, + track: track, + publication: pub, + )); } if (notify) { - delegate?.onTrackUnpublished(this, pub); - roomDelegate?.onTrackUnpublished(this, pub); - } - } - - Future _waitForTrackPublication(String sid, Duration delay) async { - final endTime = DateTime.now().add(delay); - while (DateTime.now().isBefore(endTime)) { - final pub = - await Future.delayed(const Duration(milliseconds: 100), () { - return getTrackPublication(sid); - }); - - if (pub != null) return pub; + [events, roomEvents].emit(TrackUnpublishedEvent( + participant: this, + publication: pub, + )); } } } diff --git a/lib/src/room.dart b/lib/src/room.dart index 22e70eb..59abc16 100644 --- a/lib/src/room.dart +++ b/lib/src/room.dart @@ -2,8 +2,9 @@ import 'dart:async'; import 'dart:collection'; import 'package:flutter/foundation.dart'; -import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; +import 'classes/change_notifier.dart'; +import 'constants.dart'; import 'errors.dart'; import 'events.dart'; import 'extensions.dart'; @@ -14,84 +15,13 @@ import 'participant/local_participant.dart'; import 'participant/participant.dart'; import 'participant/remote_participant.dart'; import 'proto/livekit_models.pb.dart' as lk_models; +import 'proto/livekit_rtc.pb.dart' as lk_rtc; import 'rtc_engine.dart'; import 'signal_client.dart'; -import 'track/remote_track_publication.dart'; import 'track/track.dart'; import 'track/track_publication.dart'; import 'types.dart'; -enum RoomState { - disconnected, - connected, - reconnecting, -} - -/// Delegate for [Room] callbacks -mixin RoomDelegate { - // room level callbacks - /// When the connection to the server has been interrupted and it's attempting - /// to reconnect. - void onReconnecting() {} - - /// Connection to room is re-established. All existing state is preserved. - void onReconnected() {} - - /// Disconnected from the room - void onDisconnected() {} - - /// When a new [RemoteParticipant] joins *after* the current participant has connected - /// It will not fire for participants that are already in the room - void onParticipantConnected(Participant participant) {} - - /// When a [RemoteParticipant] leaves the room - void onParticipantDisconnected(Participant participant) {} - - /// Active speakers changed. List of speakers are ordered by their audio level. - /// loudest speakers first. This will include the [LocalParticipant] too. - void onActiveSpeakersChanged(List participants) {} - - // callbacks about participant events - - /// Participant metadata is a simple way for app-specific state to be pushed to - /// all users. - /// When RoomService.UpdateParticipantMetadata is called to change a - /// participant's state, *all* participants in the room will fire this event. - void onMetadataChanged(Participant participant) {} - - /// A track that was muted, fires on both [RemoteParticipant]s and - /// [LocalParticipant] - void onTrackMuted(Participant participant, TrackPublication publication) {} - - /// A track that was unmuted, fires on both [RemoteParticipant]s and - /// [LocalParticipant] - void onTrackUnmuted(Participant participant, TrackPublication publication) {} - - /// When a new track is published to room *after* the current participant has - /// joined. It will not fire for tracks that are already published - void onTrackPublished(RemoteParticipant participant, RemoteTrackPublication publication) {} - - /// A [RemoteParticipant] has unpublished a track - void onTrackUnpublished(RemoteParticipant participant, RemoteTrackPublication publication) {} - - /// The [LocalParticipant] has subscribed to a new track. This event will **always** - /// fire as long as new tracks are ready for use. - void onTrackSubscribed( - RemoteParticipant participant, Track track, RemoteTrackPublication publication) {} - - /// A subscribed track is no longer available. - void onTrackUnsubscribed( - RemoteParticipant participant, Track track, RemoteTrackPublication publication) {} - - /// Data received from another [RemoteParticipant]. - /// Data packets provides the ability to use LiveKit to send/receive arbitrary - /// payloads. - void onDataReceived(RemoteParticipant participant, List data) {} - - /// Encountered failure attempting to subscribe to track. - void onTrackSubscriptionFailed(RemoteParticipant participant, String sid, String? message) {} -} - /// Room is the primary construct for LiveKit conferences. It contains a /// group of [Participant]s, each publishing and subscribing to [Track]s. /// Notifies changes to its state via two ways, by assigning a delegate, or using @@ -101,11 +31,12 @@ mixin RoomDelegate { /// * participant membership changes /// * active speakers are different /// {@category Room} -class Room extends ChangeNotifier with ParticipantDelegate { - RoomState _connectionState = RoomState.disconnected; +class Room extends LKChangeNotifier { + // Room is only instantiated if connected, so defaults to connected. + ConnectionState _connectionState = ConnectionState.connected; /// connection state of the room - RoomState get state => _connectionState; + ConnectionState get connectionState => _connectionState; final Map _participants = {}; @@ -114,13 +45,13 @@ class Room extends ChangeNotifier with ParticipantDelegate { UnmodifiableMapView(_participants); /// the current participant - late LocalParticipant localParticipant; + late final LocalParticipant localParticipant; /// name of the room - late String name; + late final String name; /// sid of the room - late String sid; + late final String sid; List _activeSpeakers = []; @@ -128,63 +59,28 @@ class Room extends ChangeNotifier with ParticipantDelegate { UnmodifiableListView get activeSpeakers => UnmodifiableListView(_activeSpeakers); - /// delegate for room events - RoomDelegate? delegate; - - final RTCEngine _engine; + final RTCEngine engine; // suppport for multiple event listeners final events = EventsEmitter(); - late final _engineListener = EventsListener(emitter: _engine.events); + late final _engineListener = EventsListener(engine.events); /// internal use /// {@nodoc} - Room([RTCConfiguration? rtcConfig]) : _engine = RTCEngine(SignalClient(), rtcConfig) { - _engine.onTrack = _onTrackAdded; - _engine.onICEConnected = _handleICEConnected; - _engine.onDisconnected = _handleDisconnect; - _engine.onParticipantUpdated = _handleParticipantUpdate; - _engine.onActiveSpeakerUpdated = _handleSpeakerUpdate; - _engine.onDataMessage = _handleDataPacket; - _engine.onRemoteMute = _onRemoteMuteChanged; - _engine.onReconnected = () { - _connectionState = RoomState.connected; - delegate?.onReconnected(); - notifyListeners(); - }; - _engine.onReconnecting = () { - _connectionState = RoomState.reconnecting; - delegate?.onReconnecting(); - notifyListeners(); - }; - } - - @override - Future dispose() async { - await events.dispose(); - await _engineListener.dispose(); - super.dispose(); - } - - Future connect( - String url, - String token, { - ConnectOptions? options, - }) async { - final joinResponse = await _engine.join( - url, - token, - options: options, - ); - - logger.fine('connected to LiveKit server, version: ${joinResponse.serverVersion}'); + Room._({ + required this.engine, + required lk_rtc.JoinResponse joinResponse, + ConnectOptions? connectOptions, + }) { + // + _setUpListeners(); localParticipant = LocalParticipant( - engine: _engine, + engine: engine, info: joinResponse.participant, - defaultPublishOptions: options?.defaultPublishOptions, + defaultPublishOptions: connectOptions?.defaultPublishOptions, + roomEvents: events, ); - localParticipant.roomDelegate = this; sid = joinResponse.room.sid; name = joinResponse.room.name; @@ -193,89 +89,199 @@ class Room extends ChangeNotifier with ParticipantDelegate { _getOrCreateRemoteParticipant(info.sid, info); } - // room is not ready until ICE is connected. + // Any event emitted will trigger ChangeNotifier + events.listen((event) { + logger.fine('[RoomEvent] $event, will notifyListeners()'); + notifyListeners(); + }); + } + + @override + Future dispose() async { + // dispose local participant + await localParticipant.dispose(); + // dispose Room's events emitter + await events.dispose(); + // dispose all listeners for RTCEngine + await _engineListener.dispose(); + // dispose the engine + await engine.dispose(); + + super.dispose(); + } + + static Future connect( + String url, + String token, { + ConnectOptions? options, + RTCConfiguration? rtcConfig, + }) async { + // + final engine = RTCEngine( + SignalClient(), + rtcConfig, + ); + + Room? room; + try { - await _engineListener.waitFor( - filter: (event) => event.iceState.isConnected(), - duration: const Duration(seconds: 5), + final joinResponse = await engine.join( + url, + token, + options: options, + ); + + logger.fine('Connected to LiveKit server, version: ${joinResponse.serverVersion}'); + + // create Room first to listen to events + room = Room._( + engine: engine, + joinResponse: joinResponse, + ); + + logger.fine('Waiting to engine connect...'); + + // wait until engine is connected + await room._engineListener.waitFor( + duration: Timeouts.connection, onTimeout: () => throw ConnectException(), ); + return room; // catch any exception } catch (_) { - _connectionState = RoomState.disconnected; - notifyListeners(); - - // pass on the exception + // dispose engine if there was any exception while connecting + if (room != null) { + // room.dispose will also dispose engine + await room.dispose(); + } else { + await engine.dispose(); + } rethrow; } - - return this; } + void _setUpListeners() => _engineListener + ..on((event) async { + _connectionState = ConnectionState.connected; + notifyListeners(); + }) + ..on((event) async { + _connectionState = ConnectionState.connected; + events.emit(const RoomReconnectedEvent()); + notifyListeners(); + }) + ..on((event) async { + _connectionState = ConnectionState.reconnecting; + events.emit(const RoomReconnectingEvent()); + notifyListeners(); + }) + ..on((event) => _onDisconnectedEvent()) + ..on((event) => _onParticipantUpdateEvent(event.participants)) + ..on((event) => _onSpeakerUpdateEvent(event.speakers)) + ..on(_onDataMessageEvent) + ..on((event) async { + final track = localParticipant.trackPublications[event.sid]; + track?.muted = event.muted; + }) + ..on((event) async { + final idParts = event.stream.id.split('|'); + final participantSid = idParts[0]; + final trackSid = idParts.elementAtOrNull(1) ?? event.track.id; + final participant = _getOrCreateRemoteParticipant(participantSid, null); + try { + if (trackSid == null || trackSid.isEmpty) { + throw TrackSubscriptionExceptionEvent( + participant: participant, + reason: TrackSubscribeFailReason.invalidServerResponse, + ); + } + await participant.addSubscribedMediaTrack( + event.track, + event.stream, + trackSid, + ); + } on TrackSubscriptionExceptionEvent catch (event) { + logger.warning('addSubscribedMediaTrack() throwed ${event}'); + [participant.roomEvents, participant.events].emit(event); + } catch (exception) { + // We don't want to pass up any exception so catch everything here. + logger.warning('Unknown exception on addSubscribedMediaTrack() ${exception}'); + } + }); + /// Disconnects from the room, notifying server of disconnection. Future disconnect() async { - _engine.client.sendLeave(); - await _handleDisconnect(); + engine.signalClient.sendLeave(); + await _onDisconnectedEvent(); } Future reconnect() async { - await _engine.reconnect(); + await engine.reconnect(); } RemoteParticipant _getOrCreateRemoteParticipant(String sid, lk_models.ParticipantInfo? info) { - var participant = _participants[sid]; + RemoteParticipant? participant = _participants[sid]; if (participant != null) { return participant; } if (info == null) { - participant = RemoteParticipant(_engine.client, sid, ''); + participant = RemoteParticipant( + engine.signalClient, + sid, + '', + roomEvents: events, + ); } else { - participant = RemoteParticipant.fromInfo(_engine.client, info); + participant = RemoteParticipant.fromInfo( + engine.signalClient, + info, + roomEvents: events, + ); } - participant.roomDelegate = this; + _participants[sid] = participant; return participant; } - void _handleICEConnected() { - // _connectCompleter?.complete(this); - // _connectCompleter = null; - _connectionState = RoomState.connected; - notifyListeners(); - } - - Future _handleDisconnect() async { - if (_connectionState == RoomState.disconnected) { + Future _onDisconnectedEvent() async { + if (_connectionState == ConnectionState.disconnected) { logger.fine('$objectId: _handleDisconnect() already disconnected'); return; } // we need to flag room as disconnected immediately to avoid // this method firing multiple times since the following code // is being awaited - _connectionState = RoomState.disconnected; + _connectionState = ConnectionState.disconnected; - for (final p in _participants.values) { - final tracks = List.from(p.tracks.values); - for (final pub in tracks) { - await p.unpublishTrack(pub.sid); - } + // clean up RemoteParticipants + for (final _ in _participants.values) { + // RemoteParticipant is responsible for disposing resources + await _.unpublishAllTracks(); + await _.dispose(); } - for (final pub in localParticipant.tracks.values) { - await pub.track?.stop(); - } - - await _engine.close(); _participants.clear(); + + // clean up LocalParticipant + // for (final pub in localParticipant.tracks.values) { + // await pub.track?.stop(); + // } + await localParticipant.unpublishAllTracks(); + + // await localParticipant.dispose(); + // localParticipant = null; + + await engine.close(); + _activeSpeakers.clear(); notifyListeners(); - delegate?.onDisconnected(); + events.emit(const RoomDisconnectedEvent()); } - void _handleParticipantUpdate(List updates) { + void _onParticipantUpdateEvent(List updates) async { // trigger change notifier only if list of participants membership is changed var hasChanged = false; for (final info in updates) { @@ -295,9 +301,9 @@ class Room extends ChangeNotifier with ParticipantDelegate { if (isNew) { hasChanged = true; - delegate?.onParticipantConnected(participant); + events.emit(ParticipantConnectedEvent(participant: participant)); } else { - participant.updateFromInfo(info); + await participant.updateFromInfo(info); } } @@ -306,7 +312,7 @@ class Room extends ChangeNotifier with ParticipantDelegate { } } - void _handleSpeakerUpdate(List speakers) { + void _onSpeakerUpdateEvent(List speakers) { final seenSids = {}; List newSpeakers = []; for (final info in speakers) { @@ -339,47 +345,29 @@ class Room extends ChangeNotifier with ParticipantDelegate { } } + events.emit(ActiveSpeakersChangedEvent(speakers: newSpeakers)); + _activeSpeakers = newSpeakers; - delegate?.onActiveSpeakersChanged(newSpeakers); notifyListeners(); } - void _handleDataPacket(lk_models.UserPacket packet, lk_models.DataPacket_Kind kind) { - final participant = participants[packet.participantSid]; - if (participant == null) { - return; + void _onDataMessageEvent(EngineDataPacketReceivedEvent dataPacketEvent) { + // participant may be null if data is sent from Server-API + final senderSid = dataPacketEvent.packet.participantSid; + RemoteParticipant? senderParticipant; + if (senderSid.isNotEmpty) { + senderParticipant = participants[dataPacketEvent.packet.participantSid]; } - participant.delegate?.onDataReceived(participant, packet.payload); - delegate?.onDataReceived(participant, packet.payload); - } + // participant.delegate?.onDataReceived(participant, event.packet.payload); - void _onRemoteMuteChanged(String sid, bool mute) { - final track = localParticipant.tracks[sid]; - // - // This will trigger signalClient.sendMuteTrack(sid, mute); - // - track?.muted = mute; - } + final event = DataReceivedEvent( + participant: senderParticipant, + data: dataPacketEvent.packet.payload, + ); - void _onTrackAdded( - rtc.MediaStreamTrack track, - rtc.MediaStream? stream, - rtc.RTCRtpReceiver? receiver, - ) { - if (stream == null) { - // we need the stream to get the track's id - logger.severe('received track without mediastream'); - return; - } - - final idParts = stream.id.split('|'); - - final participantSid = idParts[0]; - final trackSid = idParts.elementAtOrNull(1) ?? track.id; - - final participant = _getOrCreateRemoteParticipant(participantSid, null); - participant.addSubscribedMediaTrack(track, stream, trackSid); + senderParticipant?.events.emit(event); + events.emit(event); } void _handleParticipantDisconnect(String sid) { @@ -388,58 +376,11 @@ class Room extends ChangeNotifier with ParticipantDelegate { return; } - final toRemove = List.from(participant.tracks.values); + final toRemove = List.from(participant.trackPublications.values); for (final track in toRemove) { - participant.unpublishTrack(track.sid, true); + participant.unpublishTrack(track.sid, notify: true); } - delegate?.onParticipantDisconnected(participant); - } - //----------------- forward participant delegate calls ---------------------// - - @override - void onMetadataChanged(Participant participant) { - delegate?.onMetadataChanged(participant); - } - - @override - void onTrackMuted(Participant participant, TrackPublication publication) { - delegate?.onTrackMuted(participant, publication); - } - - @override - void onTrackUnmuted(Participant participant, TrackPublication publication) { - delegate?.onTrackUnmuted(participant, publication); - } - - @override - void onTrackPublished(RemoteParticipant participant, RemoteTrackPublication publication) { - delegate?.onTrackPublished(participant, publication); - } - - @override - void onTrackUnpublished(RemoteParticipant participant, RemoteTrackPublication publication) { - delegate?.onTrackUnpublished(participant, publication); - } - - @override - void onTrackSubscribed( - RemoteParticipant participant, Track track, RemoteTrackPublication publication) { - delegate?.onTrackSubscribed(participant, track, publication); - } - - @override - void onTrackUnsubscribed( - RemoteParticipant participant, Track track, RemoteTrackPublication publication) { - delegate?.onTrackUnsubscribed(participant, track, publication); - } - - // omitted because data dispatching is handled in _handleDataPacket - @override - void onDataReceived(RemoteParticipant participant, List data) {} - - @override - void onTrackSubscriptionFailed(RemoteParticipant participant, String sid, String? message) { - delegate?.onTrackSubscriptionFailed(participant, sid, message); + events.emit(ParticipantDisconnectedEvent(participant: participant)); } } diff --git a/lib/src/rtc_engine.dart b/lib/src/rtc_engine.dart index 221e4da..7626c9d 100644 --- a/lib/src/rtc_engine.dart +++ b/lib/src/rtc_engine.dart @@ -4,6 +4,7 @@ import 'package:collection/collection.dart'; import 'package:flutter/foundation.dart'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; +import 'constants.dart'; import 'errors.dart'; import 'events.dart'; import 'extensions.dart'; @@ -14,31 +15,15 @@ import 'options.dart'; import 'proto/livekit_models.pb.dart' as lk_models; import 'proto/livekit_rtc.pb.dart' as lk_rtc; import 'signal_client.dart'; -import 'track/track.dart'; import 'transport.dart'; import 'types.dart'; -typedef GenericCallback = void Function(); -typedef TrackCallback = void Function( - rtc.MediaStreamTrack track, - rtc.MediaStream? stream, - rtc.RTCRtpReceiver? receiver, -); -typedef ParticipantUpdateCallback = void Function(List participants); -typedef ActiveSpeakerChangedCallback = void Function(List speakers); -typedef DataPacketCallback = void Function( - lk_models.UserPacket packet, lk_models.DataPacket_Kind kind); -typedef RemoteMuteCallback = void Function(String sid, bool mute); - -class RTCEngine with SignalClientDelegate { +class RTCEngine { static const _lossyDCLabel = '_lossy'; static const _reliableDCLabel = '_reliable'; static const _maxReconnectAttempts = 5; - static const _maxICEConnectTimeout = Duration(seconds: 5); - static const _connectionTimeout = Duration(seconds: 5); - static const _iceRestartTimeout = Duration(seconds: 10); - final SignalClient client; + final SignalClient signalClient; // config for RTCPeerConnection final RTCConfiguration? rtcConfig; @@ -67,42 +52,44 @@ class RTCEngine with SignalClientDelegate { // server-provided ice servers List _providedIceServers = []; - // delegate methods - GenericCallback? onICEConnected; - TrackCallback? onTrack; - ParticipantUpdateCallback? onParticipantUpdated; - ActiveSpeakerChangedCallback? onActiveSpeakerUpdated; - DataPacketCallback? onDataMessage; - RemoteMuteCallback? onRemoteMute; - GenericCallback? onReconnecting; - GenericCallback? onReconnected; - GenericCallback? onDisconnected; - - // // internal - // - final Map> _pendingTrackResolvers = {}; int _reconnectAttempts = 0; - // to complete join request - Completer? _joinCompleter; final events = EventsEmitter(); + late final _signalListener = EventsListener(signalClient.events, synchronized: true); final delays = CancelableDelayManager(); + // late final Timer _statsTimer; + RTCEngine( - this.client, + this.signalClient, this.rtcConfig, ) { - client.delegate = this; - if (kDebugMode) { - events.listen((event) => logger.fine('[LISTENER] $objectId ${event.runtimeType}')); - events.on( - (event) => logger.fine('[LISTENER] event is a EngineIceStateUpdatedEvent')); + // log all EngineEvents + events.listen((event) => logger.fine('[EngineEvent] $objectId ${event.runtimeType}')); } + + _setUpListeners(); + // _statsTimer = Timer.periodic(const Duration(seconds: 1), _onStatTimer); } + Future dispose() async { + await events.dispose(); + await _signalListener.dispose(); + } + + // void _onStatTimer(Timer _) async { + // // + // final stats = await publisher?.pc.getStats(); + // if (stats == null || stats.isEmpty) return; + + // for (final s in stats) { + // logger.fine('STATS ${s.values}'); + // } + // } + Future join( String url, String token, { @@ -111,18 +98,16 @@ class RTCEngine with SignalClientDelegate { this.url = url; this.token = token; - final completer = Completer(); - _joinCompleter = completer; + // connect to rtc server + await signalClient.connect(url, token, options: options); - await client.join(url, token, options: options); + // wait for join response + final event = await _signalListener.waitFor( + duration: Timeouts.connection, + onTimeout: () => throw ConnectException(), + ); - // if it's not complete after 5 seconds, fail - Timer(_connectionTimeout, () { - _joinCompleter?.completeError(ConnectException()); - _joinCompleter = null; - }); - - return completer.future; + return event.response; } Future close() async { @@ -133,12 +118,12 @@ class RTCEngine with SignalClientDelegate { } isClosed = true; + // _statsTimer.cancel(); + // cancel events await _primaryIceStateListener?.call(); _primaryIceStateListener = null; - await events.dispose(); - // cancel all ongoing delays await delays.dispose(); @@ -149,7 +134,7 @@ class RTCEngine with SignalClientDelegate { await subscriber?.dispose(); subscriber = null; - client.close(); + signalClient.close(); } Future addTrack({ @@ -158,16 +143,17 @@ class RTCEngine with SignalClientDelegate { required lk_models.TrackType kind, TrackDimension? dimension, }) async { - if (_pendingTrackResolvers[cid] != null) { - throw TrackPublishException('a track with the same CID has already been published'); - } + // send request to add track + signalClient.sendAddTrack(cid: cid, name: name, type: kind, dimension: dimension); - final completer = Completer(); - _pendingTrackResolvers[cid] = completer; + // wait for response, or timeout + final event = await _signalListener.waitFor( + filter: (event) => event.cid == cid, + duration: Timeouts.publish, + onTimeout: () => throw TrackPublishException(), + ); - client.sendAddTrack(cid: cid, name: name, type: kind, dimension: dimension); - - return completer.future; + return event.track; } Future negotiate({bool? iceRestart}) async { @@ -214,7 +200,7 @@ class RTCEngine with SignalClientDelegate { await events.waitFor( filter: (event) => event.iceState.isConnected(), - duration: _maxICEConnectTimeout, + duration: Timeouts.iceConnection, ); logger.fine('[PUBLISHER] connected'); @@ -234,14 +220,13 @@ class RTCEngine with SignalClientDelegate { } if (_reconnectAttempts == 0) { - onReconnecting?.call(); - events.emit(EngineReconnectingEvent()); + events.emit(const EngineReconnectingEvent()); } _reconnectAttempts++; try { isReconnecting = true; - await client.reconnect(url, token); + await signalClient.reconnect(url, token); if (publisher == null || subscriber == null) { throw UnexpectedStateException('publisher or subscribers is null'); @@ -260,12 +245,12 @@ class RTCEngine with SignalClientDelegate { await events.waitFor( filter: (event) => event.isPrimary && event.iceState.isConnected(), - duration: _iceRestartTimeout, + duration: Timeouts.iceRestart, ); } logger.fine('reconnect: success'); - events.emit(EngineReconnectedEvent()); + events.emit(const EngineReconnectedEvent()); _reconnectAttempts = 0; // don't catch and pass up any exception @@ -293,17 +278,17 @@ class RTCEngine with SignalClientDelegate { publisher?.pc.onIceCandidate = (rtc.RTCIceCandidate candidate) { logger.fine('publisher onIceCandidate'); - client.sendIceCandidate(candidate, lk_rtc.SignalTarget.PUBLISHER); + signalClient.sendIceCandidate(candidate, lk_rtc.SignalTarget.PUBLISHER); }; subscriber?.pc.onIceCandidate = (rtc.RTCIceCandidate candidate) { logger.fine('subscriber onIceCandidate'); - client.sendIceCandidate(candidate, lk_rtc.SignalTarget.SUBSCRIBER); + signalClient.sendIceCandidate(candidate, lk_rtc.SignalTarget.SUBSCRIBER); }; publisher?.onOffer = (offer) { logger.fine('publisher onOffer'); - client.sendOffer(offer); + signalClient.sendOffer(offer); }; // in subscriber primary mode, server side opens sub data channels. @@ -336,10 +321,9 @@ class RTCEngine with SignalClientDelegate { if (!iceConnected) { iceConnected = true; if (isReconnecting) { - onReconnected?.call(); + events.emit(const EngineReconnectedEvent()); } else { - onICEConnected?.call(); - events.emit(EngineConnectedEvent()); + events.emit(const EngineConnectedEvent()); } } } else if (event.iceState == rtc.RTCIceConnectionState.RTCIceConnectionStateFailed) { @@ -352,10 +336,16 @@ class RTCEngine with SignalClientDelegate { }); subscriber?.pc.onTrack = (rtc.RTCTrackEvent event) { - onTrack?.call(event.track, event.streams.firstOrNull, event.receiver); - events.emit(EngineMediaTrackAddedEvent( + final stream = event.streams.firstOrNull; + if (stream == null) { + // we need the stream to get the track's id + logger.severe('received track without mediastream'); + return; + } + + events.emit(EngineTrackAddedEvent( track: event.track, - stream: event.streams.firstOrNull, + stream: stream, receiver: event.receiver, )); }; @@ -405,11 +395,9 @@ class RTCEngine with SignalClientDelegate { final dp = lk_models.DataPacket.fromBuffer(message.binary); if (dp.whichValue() == lk_models.DataPacket_Value.speaker) { // Speaker packet - onActiveSpeakerUpdated?.call(dp.speaker.speakers); events.emit(EngineSpeakersUpdateEvent(speakers: dp.speaker.speakers)); } else if (dp.whichValue() == lk_models.DataPacket_Value.user) { // User packet - onDataMessage?.call(dp.user, dp.kind); events.emit(EngineDataPacketReceivedEvent( packet: dp.user, kind: dp.kind, @@ -424,8 +412,7 @@ class RTCEngine with SignalClientDelegate { if (_reconnectAttempts >= _maxReconnectAttempts) { logger.info('could not connect after $_reconnectAttempts, giving up'); await close(); - onDisconnected?.call(); - events.emit(EngineDisconnectedEvent()); + events.emit(const EngineDisconnectedEvent()); return; } @@ -447,104 +434,82 @@ class RTCEngine with SignalClientDelegate { //------------------ SignalClient Delegate methods -------------------------// - @override - Future onConnected(lk_rtc.JoinResponse response) async { - // create peer connections - isClosed = false; - _subscriberPrimary = response.subscriberPrimary; - _providedIceServers = response.iceServers; + void _setUpListeners() => _signalListener + ..on((event) async { + // create peer connections + isClosed = false; + _subscriberPrimary = event.response.subscriberPrimary; + _providedIceServers = event.response.iceServers; - logger.fine('onConnected subscriberPrimary: ${_subscriberPrimary}, ' - 'serverVersion: ${response.serverVersion}, ' - 'iceServers: ${response.iceServers}'); + logger.fine('onConnected subscriberPrimary: ${_subscriberPrimary}, ' + 'serverVersion: ${event.response.serverVersion}, ' + 'iceServers: ${event.response.iceServers}'); - await _configurePeerConnections(); + await _configurePeerConnections(); - if (!_subscriberPrimary) { - // for subscriberPrimary, we negotiate when necessary (lazy) - await negotiate(); - } + if (!_subscriberPrimary) { + // for subscriberPrimary, we negotiate when necessary (lazy) + await negotiate(); + } - _joinCompleter?.complete(Future.value(response)); - _joinCompleter = null; - } + // _joinCompleter?.complete(Future.value(event.response)); + // _joinCompleter = null; + }) + ..on((_) async { + await _onDisconnected('signal'); + }) + ..on((event) async { + if (subscriber == null) { + return; + } - @override - Future onClose([String? reason]) async { - await _onDisconnected('signal'); - } + logger.fine('received server offer(type: ${event.sd.type}, ' + '${subscriber!.pc.signalingState})'); - @override - Future onOffer(rtc.RTCSessionDescription sd) async { - if (subscriber == null) { - return; - } + await subscriber!.setRemoteDescription(event.sd); - logger.fine('received server offer(type: ${sd.type}, ${subscriber!.pc.signalingState})'); - - await subscriber!.setRemoteDescription(sd); - - final answer = await subscriber!.pc.createAnswer(); - logger.fine('Created answer'); - logger.finer('sdp: ${answer.sdp}'); - await subscriber!.pc.setLocalDescription(answer); - client.sendAnswer(answer); - } - - @override - Future onAnswer(rtc.RTCSessionDescription sd) async { - if (publisher == null) { - return; - } - logger.fine('received answer (type: ${sd.type})'); - logger.finer('sdp: ${sd.sdp}'); - await publisher!.setRemoteDescription(sd); - } - - @override - Future onTrickle(rtc.RTCIceCandidate candidate, lk_rtc.SignalTarget target) async { - if (publisher == null || subscriber == null) { - return; - } - logger.fine('got ICE candidate from peer'); - if (target == lk_rtc.SignalTarget.SUBSCRIBER) { - await subscriber!.addIceCandidate(candidate); - } else if (target == lk_rtc.SignalTarget.PUBLISHER) { - await publisher!.addIceCandidate(candidate); - } - } - - @override - Future onParticipantUpdate(List updates) async { - onParticipantUpdated?.call(updates); - events.emit(EngineParticipantUpdateEvent(participants: updates)); - } - - @override - Future onLocalTrackPublished(lk_rtc.TrackPublishedResponse response) async { - final completer = _pendingTrackResolvers.remove(response.cid); - completer?.complete(Future.value(response.track)); - } - - @override - Future onActiveSpeakersChanged(List speakers) async { - onActiveSpeakerUpdated?.call(speakers); - events.emit(EngineSpeakersUpdateEvent(speakers: speakers)); - } - - @override - Future onLeave(lk_rtc.LeaveRequest req) async { - await close(); - onDisconnected?.call(); - events.emit(EngineDisconnectedEvent()); - } - - @override - Future onMuteTrack(lk_rtc.MuteTrackRequest req) async { - onRemoteMute?.call(req.sid, req.muted); - events.emit(EngineRemoteMuteChangedEvent( - sid: req.sid, - muted: req.muted, - )); - } + final answer = await subscriber!.pc.createAnswer(); + logger.fine('Created answer'); + logger.finer('sdp: ${answer.sdp}'); + await subscriber!.pc.setLocalDescription(answer); + signalClient.sendAnswer(answer); + }) + ..on((event) async { + if (publisher == null) { + return; + } + logger.fine('received answer (type: ${event.sd.type})'); + logger.finer('sdp: ${event.sd.sdp}'); + await publisher!.setRemoteDescription(event.sd); + }) + ..on((event) async { + if (publisher == null || subscriber == null) { + logger.warning('Received ${SignalTrickleEvent} but publisher or subscriber was null.'); + return; + } + logger.fine('got ICE candidate from peer'); + if (event.target == lk_rtc.SignalTarget.SUBSCRIBER) { + await subscriber!.addIceCandidate(event.candidate); + } else if (event.target == lk_rtc.SignalTarget.PUBLISHER) { + await publisher!.addIceCandidate(event.candidate); + } + }) + ..on((event) async { + events.emit(EngineParticipantUpdateEvent(participants: event.updates)); + }) + // ..on((event) async { + // final completer = _pendingTrackResolvers.remove(event.cid); + // completer?.complete(event.track); + // }) + ..on((event) async { + events.emit(EngineSpeakersUpdateEvent(speakers: event.speakers)); + }) + ..on((event) async { + await close(); + events.emit(const EngineDisconnectedEvent()); + }) + ..on((event) => events.emit(EngineRemoteMuteChangedEvent( + sid: event.sid, + muted: event.muted, + ))); } diff --git a/lib/src/signal_client.dart b/lib/src/signal_client.dart index e6975a5..bb2a27d 100644 --- a/lib/src/signal_client.dart +++ b/lib/src/signal_client.dart @@ -1,59 +1,40 @@ import 'dart:async'; import 'dart:convert'; -import 'dart:developer'; +import 'package:flutter/foundation.dart'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'package:http/http.dart' as http; -import 'package:synchronized/synchronized.dart' as sync; import 'errors.dart'; +import 'events.dart'; import 'extensions.dart'; import 'logger.dart'; +import 'managers/event.dart'; import 'options.dart'; import 'proto/livekit_models.pb.dart' as lk_models; import 'proto/livekit_rtc.pb.dart' as lk_rtc; -import 'track/track.dart'; +import 'types.dart'; import 'utils.dart'; import 'ws/interface.dart'; -mixin SignalClientDelegate { - // initial connection established - Future onConnected(lk_rtc.JoinResponse response); - // websocket has closed - Future onClose([String? reason]); - // when a server offer is received - Future onOffer(rtc.RTCSessionDescription sd); - // when an answer from server is received - Future onAnswer(rtc.RTCSessionDescription sd); - // when server has a new ICE candidate - Future onTrickle(rtc.RTCIceCandidate candidate, lk_rtc.SignalTarget target); - // participant has changed - Future onParticipantUpdate(List updates); - // when a track has been added successfully - Future onLocalTrackPublished(lk_rtc.TrackPublishedResponse response); - // active speaker has changed - Future onActiveSpeakersChanged(List speakers); - // when server sends this client a leave message - Future onLeave(lk_rtc.LeaveRequest req); - // explicit mute track - Future onMuteTrack(lk_rtc.MuteTrackRequest req); -} - class SignalClient { - final _lock = sync.Lock(); + final events = EventsEmitter(); + final ProtocolVersion protocol; - ProtocolVersion protocol; - SignalClientDelegate? delegate; bool _connected = false; LiveKitWebSocket? _ws; SignalClient({ this.protocol = ProtocolVersion.protocol3, - }); + }) { + events.listen((event) { + logger.fine('[SignalEvent] $event'); + }); + } bool get connected => _connected; - Future join( + Future connect( String uriString, String token, { ConnectOptions? options, @@ -202,7 +183,7 @@ class SignalClient { void _sendRequest(lk_rtc.SignalRequest req) { if (_ws == null) { - log('could not send message, not connected'); + logger.warning('could not send message, not connected'); return; } @@ -214,48 +195,49 @@ class SignalClient { if (message is! List) return; final msg = lk_rtc.SignalResponse.fromBuffer(message); - // Ensure previous delegate method's future is completed - // before calling another method - await _lock.synchronized(() async { - // - switch (msg.whichMessage()) { - case lk_rtc.SignalResponse_Message.join: - if (!_connected) { - _connected = true; - await delegate?.onConnected(msg.join); - } - break; - case lk_rtc.SignalResponse_Message.answer: - await delegate?.onAnswer(msg.answer.toSDKType()); - break; - case lk_rtc.SignalResponse_Message.offer: - await delegate?.onOffer(msg.offer.toSDKType()); - break; - case lk_rtc.SignalResponse_Message.trickle: - await delegate?.onTrickle( - RTCIceCandidateExt.fromJson(msg.trickle.candidateInit), - msg.trickle.target, - ); - break; - case lk_rtc.SignalResponse_Message.update: - await delegate?.onParticipantUpdate(msg.update.participants); - break; - case lk_rtc.SignalResponse_Message.trackPublished: - await delegate?.onLocalTrackPublished(msg.trackPublished); - break; - case lk_rtc.SignalResponse_Message.speaker: - await delegate?.onActiveSpeakersChanged(msg.speaker.speakers); - break; - case lk_rtc.SignalResponse_Message.leave: - await delegate?.onLeave(msg.leave); - break; - case lk_rtc.SignalResponse_Message.mute: - await delegate?.onMuteTrack(msg.mute); - break; - default: - log('unsupported message: ' + json.encode(msg)); - } - }); + switch (msg.whichMessage()) { + case lk_rtc.SignalResponse_Message.join: + if (!_connected) { + _connected = true; + events.emit(SignalConnectedEvent(response: msg.join)); + } + break; + case lk_rtc.SignalResponse_Message.answer: + events.emit(SignalAnswerEvent(sd: msg.answer.toSDKType())); + break; + case lk_rtc.SignalResponse_Message.offer: + events.emit(SignalOfferEvent(sd: msg.offer.toSDKType())); + break; + case lk_rtc.SignalResponse_Message.trickle: + events.emit(SignalTrickleEvent( + candidate: RTCIceCandidateExt.fromJson(msg.trickle.candidateInit), + target: msg.trickle.target, + )); + break; + case lk_rtc.SignalResponse_Message.update: + events.emit(SignalParticipantUpdateEvent(updates: msg.update.participants)); + break; + case lk_rtc.SignalResponse_Message.trackPublished: + events.emit(SignalLocalTrackPublishedEvent( + cid: msg.trackPublished.cid, + track: msg.trackPublished.track, + )); + break; + case lk_rtc.SignalResponse_Message.speaker: + events.emit(SignalActiveSpeakersChangedEvent(speakers: msg.speaker.speakers)); + break; + case lk_rtc.SignalResponse_Message.leave: + events.emit(SignalLeaveEvent(canReconnect: msg.leave.canReconnect)); + break; + case lk_rtc.SignalResponse_Message.mute: + events.emit(SignalMuteTrackEvent( + sid: msg.mute.sid, + muted: msg.mute.muted, + )); + break; + default: + logger.warning('unsupported message: ' + json.encode(msg)); + } } void _handleError(dynamic error) { @@ -266,6 +248,6 @@ class SignalClient { if (!_connected) return; _ws = null; _connected = false; - delegate?.onClose(); + events.emit(const SignalCloseEvent()); } } diff --git a/lib/src/track/local_track_publication.dart b/lib/src/track/local_track_publication.dart index 0c9e725..fc2fb6f 100644 --- a/lib/src/track/local_track_publication.dart +++ b/lib/src/track/local_track_publication.dart @@ -1,4 +1,7 @@ +import '../events.dart'; +import '../extensions.dart'; import '../logger.dart'; +import '../managers/event.dart'; import '../participant/local_participant.dart'; import '../proto/livekit_models.pb.dart' as lk_models; import 'track.dart'; @@ -23,15 +26,20 @@ class LocalTrackPublication extends TrackPublication { super.muted = val; track?.mediaStreamTrack.enabled = !val; - _participant.engine.client.sendMuteTrack(sid, val); + _participant.engine.signalClient.sendMuteTrack(sid, val); if (val) { - _participant.delegate?.onTrackMuted(_participant, this); - _participant.roomDelegate?.onTrackMuted(_participant, this); + // Track muted + [_participant.events, _participant.roomEvents].emit(TrackMutedEvent( + participant: _participant, + track: this, + )); } else { - _participant.delegate?.onTrackUnmuted(_participant, this); - _participant.roomDelegate?.onTrackUnmuted(_participant, this); + // Track un-muted + [_participant.events, _participant.roomEvents].emit(TrackUnmutedEvent( + participant: _participant, + track: this, + )); } - _participant.muteChanged(); } } diff --git a/lib/src/track/remote_track_publication.dart b/lib/src/track/remote_track_publication.dart index 5b0e3f0..bd0792e 100644 --- a/lib/src/track/remote_track_publication.dart +++ b/lib/src/track/remote_track_publication.dart @@ -1,7 +1,11 @@ +import 'package:livekit_client/livekit_client.dart'; + import '../participant/remote_participant.dart'; import '../proto/livekit_models.pb.dart' as lk_models; import '../proto/livekit_rtc.pb.dart' as lk_rtc; import 'track.dart'; +import '../extensions.dart'; + import 'track_publication.dart'; /// Represents a track publication from a RemoteParticipant. Provides methods to @@ -50,16 +54,21 @@ class RemoteTrackPublication extends TrackPublication { } super.muted = val; if (val) { - _participant.delegate?.onTrackMuted(_participant, this); - _participant.roomDelegate?.onTrackMuted(_participant, this); + // Track muted + [_participant.events, _participant.roomEvents].emit(TrackMutedEvent( + participant: _participant, + track: this, + )); } else { - _participant.delegate?.onTrackUnmuted(_participant, this); - _participant.roomDelegate?.onTrackUnmuted(_participant, this); + // Track un-muted + [_participant.events, _participant.roomEvents].emit(TrackUnmutedEvent( + participant: _participant, + track: this, + )); } if (subscribed) { track?.mediaStreamTrack.enabled = !val; } - _participant.muteChanged(); } RemoteTrackPublication( diff --git a/lib/src/track/track.dart b/lib/src/track/track.dart index 4a89044..af7ea5b 100644 --- a/lib/src/track/track.dart +++ b/lib/src/track/track.dart @@ -1,28 +1,29 @@ import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; +import 'package:livekit_client/src/classes/change_notifier.dart'; import 'package:uuid/uuid.dart'; import '../proto/livekit_models.pb.dart' as lk_models; -class TrackDimension { - int width; - int height; - - TrackDimension(this.width, this.height); -} - /// Wrapper around a MediaStreamTrack with additional metadata. -class Track { +/// Base for [AudioTrack] and [VideoTrack], +/// can not be instantiated directly. +abstract class Track extends LKChangeNotifier { static const cameraName = 'camera'; static const screenShareName = 'screen'; - String name; - lk_models.TrackType kind; + final String name; + final lk_models.TrackType kind; rtc.MediaStreamTrack mediaStreamTrack; + String? sid; rtc.RTCRtpTransceiver? transceiver; String? _cid; - Track(this.kind, this.name, this.mediaStreamTrack); + Track( + this.kind, + this.name, + this.mediaStreamTrack, + ); bool get muted => mediaStreamTrack.muted == null ? false : mediaStreamTrack.muted!; diff --git a/lib/src/track/track_publication.dart b/lib/src/track/track_publication.dart index 775d058..4b51838 100644 --- a/lib/src/track/track_publication.dart +++ b/lib/src/track/track_publication.dart @@ -1,13 +1,19 @@ import '../proto/livekit_models.pb.dart' as lk_models; +import '../types.dart'; import 'track.dart'; /// Represents a track that's published to the server. This class contains /// metadata associated with tracks. -class TrackPublication { +/// +/// Base for [RemoteTrackPublication] and [LocalTrackPublication], +/// can not be instantiated directly. + +abstract class TrackPublication { + final String name; + final String sid; + final lk_models.TrackType kind; + Track? track; - String name; - String sid; - lk_models.TrackType kind; bool muted = false; bool simulcasted = false; TrackDimension? dimension; @@ -31,4 +37,12 @@ class TrackPublication { dimension = TrackDimension(info.width, info.height); } } + + // Equality operators + // Object is considered equal when sid is equal + @override + int get hashCode => sid.hashCode; + + @override + bool operator ==(Object other) => other is TrackPublication && sid == other.sid; } diff --git a/lib/src/track/video_track.dart b/lib/src/track/video_track.dart index bd0e7d5..5da706b 100644 --- a/lib/src/track/video_track.dart +++ b/lib/src/track/video_track.dart @@ -1,12 +1,10 @@ -import 'package:flutter/foundation.dart'; -import 'package:flutter/material.dart'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import '../proto/livekit_models.pb.dart' as lk_models; import 'track.dart'; /// A video track will notify when its mediaTrack has changed. -class VideoTrack extends Track with ChangeNotifier { +class VideoTrack extends Track { rtc.MediaStream _mediaStream; VideoTrack( diff --git a/lib/src/transport.dart b/lib/src/transport.dart index 62860de..e31f79d 100644 --- a/lib/src/transport.dart +++ b/lib/src/transport.dart @@ -2,10 +2,11 @@ import 'dart:async'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; +import 'constants.dart'; +import 'extensions.dart'; import 'logger.dart'; import 'types.dart'; import 'utils.dart'; -import 'extensions.dart'; typedef PCTransportOnOffer = void Function(rtc.RTCSessionDescription offer); @@ -31,7 +32,7 @@ class PCTransport { late final negotiate = Utils.createDebounceFunc( () => createAndSendOffer(), cancelFunc: (f) => _cancelDebounce = f, - wait: const Duration(milliseconds: 100), + wait: Timeouts.debounce, ); Future dispose() async { diff --git a/lib/src/types.dart b/lib/src/types.dart index ef1e6b9..ae0726d 100644 --- a/lib/src/types.dart +++ b/lib/src/types.dart @@ -1,18 +1,37 @@ -// -// LiveKit -// - import 'package:flutter/material.dart'; import 'extensions.dart'; typedef CancelListenFunc = Function(); +enum ProtocolVersion { + protocol2, + protocol3, +} + +enum ConnectionState { + disconnected, + connected, + reconnecting, +} + enum Reliability { reliable, lossy, } +enum CloseReason { + network, + // ... +} + +enum TrackSubscribeFailReason { + invalidServerResponse, + notTrackMetadataFound, + unsupportedTrackType, + // ... +} + enum RTCIceTransportPolicy { all, relay, @@ -89,3 +108,14 @@ class RTCIceServer { if (credential?.isNotEmpty ?? false) 'credential': credential, }; } + +@immutable +class TrackDimension { + final int width; + final int height; + + const TrackDimension( + this.width, + this.height, + ); +} diff --git a/lib/src/utils.dart b/lib/src/utils.dart index 66539e2..f5aae38 100644 --- a/lib/src/utils.dart +++ b/lib/src/utils.dart @@ -1,25 +1,11 @@ -// -// -// - import 'dart:async'; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; +import 'extensions.dart'; import 'options.dart'; import 'track/options.dart'; - -enum ProtocolVersion { - protocol2, - protocol3, -} - -extension ProtocolVersionExt on ProtocolVersion { - String toStringValue() => { - ProtocolVersion.protocol2: '2', - ProtocolVersion.protocol3: '3', - }[this]!; -} +import 'types.dart'; extension UriExt on Uri { bool get isSecureScheme => ['https', 'wss'].contains(scheme); diff --git a/lib/src/ws/platform/io.dart b/lib/src/ws/platform/io.dart index 41d614d..417569e 100644 --- a/lib/src/ws/platform/io.dart +++ b/lib/src/ws/platform/io.dart @@ -33,7 +33,19 @@ class LiveKitWebSocketIO implements LiveKitWebSocket { } @override - void send(List data) => _ws.add(data); + void send(List data) { + // 0 CONNECTING + // 1 OPEN + // 2 CLOSING + // 3 CLOSED + if (_ws.readyState == 1) { + try { + _ws.add(data); + } catch (e) { + // + } + } + } static Future connect( Uri uri, [ diff --git a/pubspec.lock b/pubspec.lock index d60c3a1..0d7a855 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -98,9 +98,11 @@ packages: flutter_webrtc: dependency: "direct main" description: - name: flutter_webrtc - url: "https://pub.dartlang.org" - source: hosted + path: "." + ref: use-custom-webrtc-build + resolved-ref: "4942e7faec2e5775d35c42e22c2929ca6ca53769" + url: "https://github.com/livekit/flutter-webrtc" + source: git version: "0.6.7" http: dependency: "direct main" @@ -157,7 +159,7 @@ packages: name: path_provider url: "https://pub.dartlang.org" source: hosted - version: "2.0.4" + version: "2.0.5" path_provider_linux: dependency: transitive description: @@ -311,5 +313,5 @@ packages: source: hosted version: "0.2.0" sdks: - dart: ">=2.13.0 <3.0.0" - flutter: ">=2.0.0" + dart: ">=2.14.0 <3.0.0" + flutter: ">=2.5.0" diff --git a/pubspec.yaml b/pubspec.yaml index a31feb4..5eb9b4a 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -10,13 +10,23 @@ environment: dependencies: flutter: sdk: flutter - flutter_webrtc: ^0.6.7 + http: ^0.13.3 logging: ^1.0.2 uuid: ^3.0.4 synchronized: ^3.0.0 protobuf: ^2.0.0 + flutter_webrtc: + git: + url: https://github.com/livekit/flutter-webrtc + ref: use-custom-webrtc-build + + # ^0.6.7 + # path: ../../repos_livekit/flutter-webrtc/ + # This will use custom webrtc build from + # https://github.com/webrtc-sdk/Specs/releases + # protobuf: # git: # url: https://github.com/google/protobuf.dart.git