From 6445b8d2a8ac863b3d59d48d4620fbd0c8f7b6dd Mon Sep 17 00:00:00 2001
From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com>
Date: Fri, 24 Sep 2021 23:37:26 +0900
Subject: [PATCH] Switch to new event system (#9)
* `SignalEvents` and `synchronized` mode for `EventsListenable`
* cascade syntax
* clean up
* `RoomEvents`
* `ParticipantEvents`
* all events implemented
* make it compile with M1 macs
* dispose logic
* cleaner connect logic
* ask to publish
* fix initial EngineTrackAddedEvent glitch
* fix unpublishTrack bug
* clean up
* better events docs
* organize event types
* clean up
* cleaner wait logic
* wait for event instead of using Completer
* notifyListeners
---
example/ios/Flutter/AppFrameworkInfo.plist | 2 +-
example/ios/Podfile | 4 +
example/ios/Podfile.lock | 20 +-
example/ios/Runner.xcodeproj/project.pbxproj | 5 +-
example/lib/exts.dart | 20 +-
example/lib/main.dart | 2 +
example/lib/pages/connect.dart | 10 +-
example/lib/pages/room.dart | 121 ++---
example/lib/widgets/controls.dart | 4 +-
example/lib/widgets/participant.dart | 2 +-
example/pubspec.lock | 22 +-
lib/livekit_client.dart | 2 +
lib/src/classes/change_notifier.dart | 32 ++
lib/src/constants.dart | 7 +
lib/src/events.dart | 380 ++++++++++++----
lib/src/extensions.dart | 88 ++--
lib/src/livekit.dart | 14 +-
lib/src/managers/delay.dart | 3 -
lib/src/managers/event.dart | 63 ++-
lib/src/participant/local_participant.dart | 38 +-
lib/src/participant/participant.dart | 136 +++---
lib/src/participant/remote_participant.dart | 168 +++----
lib/src/room.dart | 443 ++++++++-----------
lib/src/rtc_engine.dart | 311 ++++++-------
lib/src/signal_client.dart | 132 +++---
lib/src/track/local_track_publication.dart | 20 +-
lib/src/track/remote_track_publication.dart | 19 +-
lib/src/track/track.dart | 23 +-
lib/src/track/track_publication.dart | 22 +-
lib/src/track/video_track.dart | 4 +-
lib/src/transport.dart | 5 +-
lib/src/types.dart | 38 +-
lib/src/utils.dart | 18 +-
lib/src/ws/platform/io.dart | 14 +-
pubspec.lock | 14 +-
pubspec.yaml | 12 +-
36 files changed, 1270 insertions(+), 948 deletions(-)
create mode 100644 lib/src/classes/change_notifier.dart
create mode 100644 lib/src/constants.dart
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