diff --git a/example/lib/exts.dart b/example/lib/exts.dart index 55308f3..0cfd4c5 100644 --- a/example/lib/exts.dart +++ b/example/lib/exts.dart @@ -15,4 +15,72 @@ extension LKExampleExt on BuildContext { ], ), ); + + Future showDisconnectDialog() => showDialog( + context: this, + builder: (ctx) => AlertDialog( + title: const Text('Disconnect'), + content: const Text('Are you sure to disconnect?'), + actions: [ + TextButton( + onPressed: () => Navigator.pop(ctx, false), + child: const Text('Cancel'), + ), + TextButton( + onPressed: () => Navigator.pop(ctx, true), + child: const Text('Disconnect'), + ), + ], + ), + ); + + Future showReconnectDialog() => showDialog( + context: this, + builder: (ctx) => AlertDialog( + title: const Text('Reconnect'), + content: const Text('This will force a reconnection'), + actions: [ + TextButton( + onPressed: () => Navigator.pop(ctx, false), + child: const Text('Cancel'), + ), + TextButton( + onPressed: () => Navigator.pop(ctx, true), + child: const Text('Reconnect'), + ), + ], + ), + ); + + Future showSendDataDialog() => showDialog( + context: this, + builder: (ctx) => AlertDialog( + title: const Text('Send data'), + content: const Text('This will send a sample data to all participants in the room'), + actions: [ + TextButton( + onPressed: () => Navigator.pop(ctx, false), + child: const Text('Cancel'), + ), + TextButton( + onPressed: () => Navigator.pop(ctx, true), + child: const Text('Send'), + ), + ], + ), + ); + + Future showDataReceivedDialog(String data) => showDialog( + context: this, + builder: (ctx) => AlertDialog( + title: const Text('Received data'), + content: Text('"${data}"'), + actions: [ + TextButton( + onPressed: () => Navigator.pop(ctx, true), + child: const Text('OK'), + ), + ], + ), + ); } diff --git a/example/lib/pages/room.dart b/example/lib/pages/room.dart index 78128c2..236fa43 100644 --- a/example/lib/pages/room.dart +++ b/example/lib/pages/room.dart @@ -1,3 +1,4 @@ +import 'dart:convert'; import 'dart:math' as math; import 'package:flutter/material.dart'; @@ -6,6 +7,7 @@ import 'package:provider/provider.dart'; import '../widgets/controls.dart'; import '../widgets/participant.dart'; +import '../exts.dart'; class RoomPage extends StatefulWidget { // @@ -17,13 +19,10 @@ class RoomPage extends StatefulWidget { }) : super(key: key); @override - State createState() { - return _RoomPageState(); - } + State createState() => _RoomPageState(); } class _RoomPageState extends State with RoomDelegate { - // BuildContext? _lastContext; // List participants = []; @@ -107,13 +106,15 @@ class _RoomPageState extends State with RoomDelegate { }); } + @override + void onDataReceived(RemoteParticipant participant, List data) async { + await context.showDataReceivedDialog(utf8.decode(data)); + } + @override void onDisconnected() { - // final context = _lastContext; print('disconnected: $context'); - // if (context != null) { Navigator.pop(context); - // } } @override diff --git a/example/lib/widgets/controls.dart b/example/lib/widgets/controls.dart index 11cdc99..194b9af 100644 --- a/example/lib/widgets/controls.dart +++ b/example/lib/widgets/controls.dart @@ -1,7 +1,10 @@ +import 'dart:convert'; + import 'package:eva_icons_flutter/eva_icons_flutter.dart'; import 'package:flutter/material.dart'; import 'package:livekit_client/livekit_client.dart'; import 'package:collection/collection.dart'; +import '../exts.dart'; class ControlsWidget extends StatefulWidget { // @@ -113,8 +116,23 @@ class _ControlsWidgetState extends State { } } - void _exit() { - widget.room.disconnect(); + void _onTapDisconnect() async { + final result = await context.showDisconnectDialog(); + if (result == true) await widget.room.disconnect(); + } + + void _onTapReconnect() async { + final result = await context.showReconnectDialog(); + if (result == true) await widget.room.reconnect(); + } + + void _onTapSendData() async { + final result = await context.showSendDataDialog(); + if (result == true) { + await widget.room.localParticipant.publishData( + utf8.encode('This is a sample data message'), + ); + } } @override @@ -157,9 +175,17 @@ class _ControlsWidgetState extends State { onPressed: () => _shareScreen(), ), IconButton( - onPressed: _exit, + onPressed: _onTapDisconnect, icon: const Icon(EvaIcons.closeCircle), - ) + ), + IconButton( + onPressed: _onTapSendData, + icon: const Icon(EvaIcons.paperPlane), + ), + IconButton( + onPressed: _onTapReconnect, + icon: const Icon(EvaIcons.refresh), + ), ], ); } diff --git a/example/pubspec.lock b/example/pubspec.lock index 3f0d7b6..b2c64ef 100644 --- a/example/pubspec.lock +++ b/example/pubspec.lock @@ -162,7 +162,7 @@ packages: name: logging url: "https://pub.dartlang.org" source: hosted - version: "1.0.1" + version: "1.0.2" matcher: dependency: transitive description: @@ -268,13 +268,6 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "5.0.0" - quiver: - dependency: transitive - description: - name: quiver - url: "https://pub.dartlang.org" - source: hosted - version: "3.0.1" shared_preferences: dependency: "direct main" description: @@ -371,13 +364,6 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "0.4.2" - tuple: - dependency: transitive - description: - name: tuple - url: "https://pub.dartlang.org" - source: hosted - version: "2.0.0" typed_data: dependency: transitive description: diff --git a/lib/livekit_client.dart b/lib/livekit_client.dart index 365ff00..4b11945 100644 --- a/lib/livekit_client.dart +++ b/lib/livekit_client.dart @@ -19,4 +19,5 @@ export 'src/track/remote_track_publication.dart'; export 'src/track/track.dart'; export 'src/track/track_publication.dart'; export 'src/track/video_track.dart'; +export 'src/types.dart' show RTCConfiguration, RTCIceServer, RTCIceTransportPolicy, Reliability; export 'src/widget/video_track_renderer.dart'; diff --git a/lib/src/errors.dart b/lib/src/errors.dart index 6f3f7d7..382ac9c 100644 --- a/lib/src/errors.dart +++ b/lib/src/errors.dart @@ -1,32 +1,34 @@ // -// `Exception` implies runtime errors while, an `Error` object -// represents a program failure that the programmer -// should have avoided. +// // class LiveKitException implements Exception { final String message; const LiveKitException._(this.message); @override - String toString() => 'LiveKitException $runtimeType $message'; + String toString() => 'LiveKit Exception $runtimeType $message'; } -class ConnectError extends LiveKitException { - ConnectError([String msg = 'Failed to connect to server']) : super._(msg); +class ConnectException extends LiveKitException { + ConnectException([String msg = 'Failed to connect to server']) : super._(msg); } -class UnexpectedConnectionState extends LiveKitException { - UnexpectedConnectionState([String msg = 'Unexpected connection state']) : super._(msg); +class UnexpectedStateException extends LiveKitException { + UnexpectedStateException([String msg = 'Unexpected connection state']) : super._(msg); } -class TrackCreateError extends LiveKitException { - TrackCreateError([String msg = 'Failed to create track']) : super._(msg); +class TrackCreateException extends LiveKitException { + TrackCreateException([String msg = 'Failed to create track']) : super._(msg); } -class TrackPublishError extends LiveKitException { - TrackPublishError([String msg = 'Failed to publish track']) : super._(msg); +class TrackPublishException extends LiveKitException { + TrackPublishException([String msg = 'Failed to publish track']) : super._(msg); } -class DataPublishError extends LiveKitException { - DataPublishError([String msg = 'Failed to publish data']) : super._(msg); +class DataPublishException extends LiveKitException { + DataPublishException([String msg = 'Failed to publish data']) : super._(msg); +} + +class TimeoutException extends LiveKitException { + TimeoutException([String msg = 'Timeout']) : super._(msg); } diff --git a/lib/src/events.dart b/lib/src/events.dart new file mode 100644 index 0000000..f11fffa --- /dev/null +++ b/lib/src/events.dart @@ -0,0 +1,181 @@ +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; + +import 'proto/livekit_models.pb.dart' as lk_models; + +abstract class LiveKitEvent {} + +abstract class RoomEvent implements LiveKitEvent { + const RoomEvent(); +} + +abstract class ParticipantEvent implements LiveKitEvent { + const ParticipantEvent(); +} + +abstract class EngineEvent implements LiveKitEvent { + const EngineEvent(); +} + +abstract class TrackEvent implements LiveKitEvent { + const TrackEvent(); +} + +// +// Room events +// +class RoomReconnectingEvent extends RoomEvent {} + +class RoomReconnectedEvent extends RoomEvent {} + +class RoomDisconnectedEvent extends RoomEvent {} + +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 {} + +// +// Engine events +// +class EngineConnectedEvent extends EngineEvent {} + +class EngineDisconnectedEvent extends EngineEvent {} + +class EngineReconnectingEvent extends EngineEvent {} + +class EngineReconnectedEvent extends EngineEvent {} + +class EngineParticipantUpdateEvent extends EngineEvent { + final List participants; + const EngineParticipantUpdateEvent({ + required this.participants, + }); +} + +class EngineMediaTrackAddedEvent extends EngineEvent { + final rtc.MediaStreamTrack track; + final rtc.MediaStream? stream; + final rtc.RTCRtpReceiver? receiver; + const EngineMediaTrackAddedEvent({ + required this.track, + required this.stream, + required this.receiver, + }); +} + +class EngineSpeakersUpdateEvent extends EngineEvent { + final List speakers; + const EngineSpeakersUpdateEvent({ + required this.speakers, + }); +} + +class EngineDataPacketReceivedEvent extends EngineEvent { + final lk_models.UserPacket packet; + final lk_models.DataPacket_Kind kind; + const EngineDataPacketReceivedEvent({ + required this.packet, + required this.kind, + }); +} + +class EngineRemoteMuteChangedEvent extends EngineEvent { + final String sid; + final bool muted; + const EngineRemoteMuteChangedEvent({ + required this.sid, + required this.muted, + }); +} + +// added +abstract class EngineIceStateUpdatedEvent implements EngineEvent { + final rtc.RTCIceConnectionState iceState; + final bool isPrimary; + const EngineIceStateUpdatedEvent({ + required this.iceState, + required this.isPrimary, + }); +} + +class EngineSubscriberIceStateUpdatedEvent extends EngineIceStateUpdatedEvent { + const EngineSubscriberIceStateUpdatedEvent({ + required rtc.RTCIceConnectionState state, + required bool isPrimary, + }) : super( + iceState: state, + isPrimary: isPrimary, + ); +} + +class EnginePublisherIceStateUpdatedEvent extends EngineIceStateUpdatedEvent { + const EnginePublisherIceStateUpdatedEvent({ + required rtc.RTCIceConnectionState state, + required bool isPrimary, + }) : super( + iceState: state, + isPrimary: isPrimary, + ); +} + +// +// Track events +// + +class TrackMessageEvent extends TrackEvent {} + +class TrackMutedEvent extends TrackEvent {} + +class TrackUnmutedEvent extends TrackEvent {} + +class TrackUpdateSettingsEvent extends TrackEvent {} + +class TrackUpdateSubscriptionEvent extends TrackEvent {} + +class TrackAudioPlaybackStartedEvent extends TrackEvent {} + +class TrackAudioPlaybackFailedEvent extends TrackEvent {} diff --git a/lib/src/extensions.dart b/lib/src/extensions.dart index 365fbfa..21f5f3a 100644 --- a/lib/src/extensions.dart +++ b/lib/src/extensions.dart @@ -1,6 +1,25 @@ -enum RTCIceTransportPolicy { - all, - relay, +import 'dart:convert'; + +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; + +import 'proto/livekit_rtc.pb.dart' as lk_rtc; +import 'proto/livekit_models.pb.dart' as lk_models; + +import 'types.dart'; + +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 { @@ -10,41 +29,51 @@ extension RTCIceTransportPolicyExt on RTCIceTransportPolicy { }[this]!; } -class RTCConfiguration { - int? iceCandidatePoolSize; - List? iceServers; - RTCIceTransportPolicy? iceTransportPolicy; - - Map toMap() { - final iceServersMap = >[ - if (iceServers != null) - for (final element in iceServers!) element.toMap() - ]; - - return { - // only supports unified plan - 'sdpSemantics': 'unified-plan', - if (iceServersMap.isNotEmpty) 'iceServers': iceServersMap, - if (iceCandidatePoolSize != null) 'iceCandidatePoolSize': iceCandidatePoolSize, - if (iceTransportPolicy != null) 'iceTransportPolicy': iceTransportPolicy!.toStringValue(), - }; +extension SessionDescriptionExt on lk_rtc.SessionDescription { + rtc.RTCSessionDescription toSDKType() { + return rtc.RTCSessionDescription(sdp, type); } } -class RTCIceServer { - List urls; - String? username; - String? credential; +extension RTCSessionDescriptionExt on rtc.RTCSessionDescription { + lk_rtc.SessionDescription toSDKType() { + return lk_rtc.SessionDescription(type: type, sdp: sdp); + } +} - RTCIceServer({ - required this.urls, - this.username, - this.credential, - }); +extension RTCIceCandidateExt on rtc.RTCIceCandidate { + static rtc.RTCIceCandidate fromJson(String jsonString) { + final map = json.decode(jsonString) as Map; + return rtc.RTCIceCandidate( + map['candidate'] as String?, + map['sdpMid'] as String?, + map['sdpMLineIndex'] as int?, + ); + } - Map toMap() => { - 'urls': urls, - if (username != null) 'username': username, - if (credential != null) 'credential': credential, - }; + 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, + ); +} + +// 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 ReliabilityExt on Reliability { + lk_models.DataPacket_Kind toPBType() => { + Reliability.reliable: lk_models.DataPacket_Kind.RELIABLE, + Reliability.lossy: lk_models.DataPacket_Kind.LOSSY, + }[this]!; } diff --git a/lib/src/managers/delay.dart b/lib/src/managers/delay.dart new file mode 100644 index 0000000..dca4e1a --- /dev/null +++ b/lib/src/managers/delay.dart @@ -0,0 +1,34 @@ +// +// +// +import 'package:async/async.dart'; + +class CancelableDelayManager { + // + final _delays = >[]; + + // delay but cancelable + Future waitFor( + Duration wait, { + Function? ifNotCancelled, + }) async { + final op = CancelableOperation.fromFuture( + Future.delayed(wait), + ); + _delays.add(op); + await op.valueOrCancellation(); + _delays.remove(op); + // if it was cancelled we probably don't want to execute it + if (!op.isCanceled) ifNotCancelled?.call(); + } + + Future dispose() async { + // cancel all delays + if (_delays.isEmpty) return; + // make a copy so we don't mutate while iterating + final snapshot = List>.from(_delays); + for (final op in snapshot) { + await op.cancel(); + } + } +} diff --git a/lib/src/managers/event.dart b/lib/src/managers/event.dart new file mode 100644 index 0000000..a24e4fd --- /dev/null +++ b/lib/src/managers/event.dart @@ -0,0 +1,114 @@ +import 'dart:async'; + +import 'package:flutter/material.dart'; + +import '../errors.dart'; +import '../events.dart'; +import '../extensions.dart'; +import '../logger.dart'; +import '../types.dart'; + +// Type-safe, multi-listenable, dispose safe event handling + +class EventsEmitter extends EventsListenable { + // suppport for multiple event listeners + final streamCtrl = StreamController.broadcast(sync: false); + + @override + EventsEmitter get emitter => this; + + void emit(T event) { + // do nothing if already closed + if (streamCtrl.isClosed) return; + // emit the event + streamCtrl.add(event); + } + + @override + Future dispose() async { + await streamCtrl.close(); + await super.dispose(); + } +} + +// for listening only +class EventsListener extends EventsListenable { + @override + final EventsEmitter emitter; + + EventsListener({ + required this.emitter, + }); +} + +// ensures all listeners will close on dispose +abstract class EventsListenable { + // the emitter to listen to + EventsEmitter get emitter; + // keep track of listeners to cancel later + final _listeners = >[]; + + @mustCallSuper + Future dispose() async { + // Stop listening to all events + logger.fine('${objectId} dispose() cancelling ${_listeners.length} event(s)'); + for (final listener in _listeners) { + await listener.cancel(); + } + } + + // listens to all events, guaranteed to be cancelled on dispose + CancelListenFunc listen(Function(T) onEvent) { + final listener = emitter.streamCtrl.stream.listen(onEvent); + _listeners.add(listener); + + // make a cancel func to cancel listening and remove from list in 1 call + _cancelFunc() async { + await listener.cancel(); + _listeners.remove(listener); + logger.fine('${objectId} event was cancelled by func'); + } + + return _cancelFunc; + } + + // convenience method to listen & filter a specific event type + CancelListenFunc on( + Function(E) then, { + bool Function(E)? filter, + }) => + listen((event) { + // 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; + // cast to E + then(event as E); + }); + + // waits for a specific event type + Future waitFor({ + required Duration duration, + bool Function(E)? filter, + FutureOr Function()? onTimeout, + }) async { + final completer = Completer(); + + final _cancelFunc = on( + (event) => completer.complete(), + filter: filter, + ); + + try { + // wait to complete with timeout + await completer.future.timeout( + duration, + onTimeout: onTimeout ?? () => throw TimeoutException(), + ); + // do not catch exceptions and pass it up + } finally { + // always clean-up listener + await _cancelFunc.call(); + } + } +} diff --git a/lib/src/participant/local_participant.dart b/lib/src/participant/local_participant.dart index cd9c55f..2cea886 100644 --- a/lib/src/participant/local_participant.dart +++ b/lib/src/participant/local_participant.dart @@ -1,7 +1,8 @@ import 'package:flutter/foundation.dart'; -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import '../errors.dart'; +import '../extensions.dart'; import '../logger.dart'; import '../options.dart'; import '../proto/livekit_models.pb.dart' as lk_models; @@ -11,6 +12,7 @@ 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'; import 'participant.dart'; @@ -35,7 +37,7 @@ class LocalParticipant extends Participant { /// publish an audio track to the room Future publishAudioTrack(LocalAudioTrack track) async { if (audioTracks.any((e) => e.track?.mediaStreamTrack.id == track.mediaStreamTrack.id)) { - throw TrackPublishError('track already exists'); + throw TrackPublishException('track already exists'); } // try { @@ -45,14 +47,16 @@ class LocalParticipant extends Participant { kind: track.kind, ); - final transceiverInit = RTCRtpTransceiverInit( - direction: TransceiverDirection.SendOnly, + final transceiverInit = rtc.RTCRtpTransceiverInit( + direction: rtc.TransceiverDirection.SendOnly, ); // addTransceiver cannot pass in a kind parameter due to a bug in flutter-webrtc (web) track.transceiver = await _engine.publisher?.pc.addTransceiver( track: track.mediaStreamTrack, + kind: rtc.RTCRtpMediaType.RTCRtpMediaTypeAudio, init: transceiverInit, ); + await _engine.negotiate(); final pub = LocalTrackPublication(trackInfo, track, this); addTrackPublication(pub); @@ -67,7 +71,7 @@ class LocalParticipant extends Participant { TrackPublishOptions? options, }) async { if (videoTracks.any((e) => e.track?.mediaStreamTrack.id == track.mediaStreamTrack.id)) { - throw TrackPublishError('track already exists'); + throw TrackPublishException('track already exists'); } // Use default options from `ConnectOptions` if options is null @@ -78,10 +82,9 @@ class LocalParticipant extends Participant { name: track.name, kind: track.kind, ); + logger.fine('publishVideoTrack addTrack response: ${trackInfo}'); - // // Video encodings and simulcasts - // // use constraints passed to getUserMedia by default int? width = track.currentOptions.params.width; @@ -94,8 +97,6 @@ class LocalParticipant extends Participant { final settings = track.mediaStreamTrack.getSettings(); width = settings['width'] as int?; height = settings['height'] as int?; - // TODO: Get actual video dimensions to compute more accurately - // mediaTrack.getConsstraints() is not implemented for mobile } catch (_) { logger.warning('Failed to call `mediaStreamTrack.getSettings()`'); } @@ -111,19 +112,20 @@ class LocalParticipant extends Participant { logger.fine('Using encodings: ${encodings?.map((e) => e.toMap())}'); - final transceiverInit = RTCRtpTransceiverInit( - direction: TransceiverDirection.SendOnly, + final transceiverInit = rtc.RTCRtpTransceiverInit( + direction: rtc.TransceiverDirection.SendOnly, sendEncodings: encodings, streams: [track.mediaStream], ); - // - // addTransceiver cannot pass in a kind parameter due to a bug in flutter-webrtc (web) - // + logger.fine('publishVideoTrack publisher: ${_engine.publisher}'); + track.transceiver = await _engine.publisher?.pc.addTransceiver( track: track.mediaStreamTrack, + kind: rtc.RTCRtpMediaType.RTCRtpMediaTypeVideo, init: transceiverInit, ); + await _engine.negotiate(); final pub = LocalTrackPublication(trackInfo, track, this); addTrackPublication(pub); @@ -144,6 +146,7 @@ class LocalParticipant extends Participant { final sender = track.transceiver?.sender; if (sender != null) { await engine.publisher?.pc.removeTrack(sender); + await engine.negotiate(); } tracks.remove(pub.sid); @@ -151,26 +154,13 @@ class LocalParticipant extends Participant { /// Publish a new data payload to the room. /// @param destinationSids When empty, data will be forwarded to each participant in the room. - void publishData( - List data, - lk_models.DataPacket_Kind reliability, { + Future publishData( + List data, { + Reliability reliability = Reliability.reliable, List? destinationSids, - }) { - RTCDataChannel? channel; - switch (reliability) { - case lk_models.DataPacket_Kind.RELIABLE: - channel = engine.reliableDC; - break; - case lk_models.DataPacket_Kind.LOSSY: - channel = engine.lossyDC; - break; - } - if (channel == null) { - return; - } - + }) async { final packet = lk_models.DataPacket( - kind: reliability, + kind: reliability.toPBType(), user: lk_models.UserPacket( payload: data, participantSid: sid, @@ -178,8 +168,7 @@ class LocalParticipant extends Participant { ), ); - final buffer = packet.writeToBuffer(); - channel.send(RTCDataChannelMessage.fromBinary(buffer)); + await engine.sendDataPacket(packet); } /// for internal use diff --git a/lib/src/participant/participant.dart b/lib/src/participant/participant.dart index f1e4b0d..bf4c1fa 100644 --- a/lib/src/participant/participant.dart +++ b/lib/src/participant/participant.dart @@ -1,5 +1,7 @@ import 'package:flutter/foundation.dart'; +import '../events.dart'; +import '../managers/event.dart'; import '../proto/livekit_models.pb.dart' as lk_models; import '../track/remote_track_publication.dart'; import '../track/track.dart'; @@ -77,6 +79,9 @@ class Participant extends ChangeNotifier { lk_models.ParticipantInfo? _participantInfo; bool _isSpeaking = false; + // suppport for multiple event listeners + final events = EventsEmitter(); + /// when the participant joined the room DateTime get joinedAt { final pi = _participantInfo; @@ -159,7 +164,7 @@ class Participant extends ChangeNotifier { } // Convenience extension -extension LKParticipantExt on Participant { +extension ParticipantExt on Participant { List get videoTracks => tracks.values.where((e) => e.kind == lk_models.TrackType.VIDEO).toList(); diff --git a/lib/src/participant/remote_participant.dart b/lib/src/participant/remote_participant.dart index caa9719..88b8c7b 100644 --- a/lib/src/participant/remote_participant.dart +++ b/lib/src/participant/remote_participant.dart @@ -1,4 +1,4 @@ -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import '../logger.dart'; import '../proto/livekit_models.pb.dart' as lk_models; @@ -35,7 +35,11 @@ class RemoteParticipant extends Participant { /// for internal use /// {@nodoc} - void addSubscribedMediaTrack(MediaStreamTrack mediaTrack, MediaStream stream, String? sid) async { + void addSubscribedMediaTrack( + rtc.MediaStreamTrack mediaTrack, + rtc.MediaStream stream, + String? sid, + ) async { if (sid == null) { const msg = 'addSubscribedMediaTrack received null sid'; delegate?.onTrackSubscriptionFailed(this, '', msg); diff --git a/lib/src/proto/livekit_rtc.pb.dart b/lib/src/proto/livekit_rtc.pb.dart index 06f7f67..42dab15 100644 --- a/lib/src/proto/livekit_rtc.pb.dart +++ b/lib/src/proto/livekit_rtc.pb.dart @@ -935,6 +935,8 @@ class JoinResponse extends $pb.GeneratedMessage { const $core.bool.fromEnvironment('protobuf.omit_field_names') ? '' : 'iceServers', $pb.PbFieldType.PM, subBuilder: ICEServer.create) + ..aOB( + 6, const $core.bool.fromEnvironment('protobuf.omit_field_names') ? '' : 'subscriberPrimary') ..hasRequiredFields = false; JoinResponse._() : super(); @@ -944,6 +946,7 @@ class JoinResponse extends $pb.GeneratedMessage { $core.Iterable<$0.ParticipantInfo>? otherParticipants, $core.String? serverVersion, $core.Iterable? iceServers, + $core.bool? subscriberPrimary, }) { final _result = create(); if (room != null) { @@ -961,6 +964,9 @@ class JoinResponse extends $pb.GeneratedMessage { if (iceServers != null) { _result.iceServers.addAll(iceServers); } + if (subscriberPrimary != null) { + _result.subscriberPrimary = subscriberPrimary; + } return _result; } factory JoinResponse.fromBuffer($core.List<$core.int> i, @@ -1034,6 +1040,18 @@ class JoinResponse extends $pb.GeneratedMessage { @$pb.TagNumber(5) $core.List get iceServers => $_getList(4); + + @$pb.TagNumber(6) + $core.bool get subscriberPrimary => $_getBF(5); + @$pb.TagNumber(6) + set subscriberPrimary($core.bool v) { + $_setBool(5, v); + } + + @$pb.TagNumber(6) + $core.bool hasSubscriberPrimary() => $_has(5); + @$pb.TagNumber(6) + void clearSubscriberPrimary() => clearField(6); } class TrackPublishedResponse extends $pb.GeneratedMessage { diff --git a/lib/src/proto/livekit_rtc.pbjson.dart b/lib/src/proto/livekit_rtc.pbjson.dart index 01e77d4..36b572d 100644 --- a/lib/src/proto/livekit_rtc.pbjson.dart +++ b/lib/src/proto/livekit_rtc.pbjson.dart @@ -304,12 +304,13 @@ const JoinResponse$json = const { '6': '.livekit.ICEServer', '10': 'iceServers' }, + const {'1': 'subscriber_primary', '3': 6, '4': 1, '5': 8, '10': 'subscriberPrimary'}, ], }; /// Descriptor for `JoinResponse`. Decode as a `google.protobuf.DescriptorProto`. final $typed_data.Uint8List joinResponseDescriptor = $convert.base64Decode( - 'CgxKb2luUmVzcG9uc2USIQoEcm9vbRgBIAEoCzINLmxpdmVraXQuUm9vbVIEcm9vbRI6CgtwYXJ0aWNpcGFudBgCIAEoCzIYLmxpdmVraXQuUGFydGljaXBhbnRJbmZvUgtwYXJ0aWNpcGFudBJHChJvdGhlcl9wYXJ0aWNpcGFudHMYAyADKAsyGC5saXZla2l0LlBhcnRpY2lwYW50SW5mb1IRb3RoZXJQYXJ0aWNpcGFudHMSJQoOc2VydmVyX3ZlcnNpb24YBCABKAlSDXNlcnZlclZlcnNpb24SMwoLaWNlX3NlcnZlcnMYBSADKAsyEi5saXZla2l0LklDRVNlcnZlclIKaWNlU2VydmVycw=='); + 'CgxKb2luUmVzcG9uc2USIQoEcm9vbRgBIAEoCzINLmxpdmVraXQuUm9vbVIEcm9vbRI6CgtwYXJ0aWNpcGFudBgCIAEoCzIYLmxpdmVraXQuUGFydGljaXBhbnRJbmZvUgtwYXJ0aWNpcGFudBJHChJvdGhlcl9wYXJ0aWNpcGFudHMYAyADKAsyGC5saXZla2l0LlBhcnRpY2lwYW50SW5mb1IRb3RoZXJQYXJ0aWNpcGFudHMSJQoOc2VydmVyX3ZlcnNpb24YBCABKAlSDXNlcnZlclZlcnNpb24SMwoLaWNlX3NlcnZlcnMYBSADKAsyEi5saXZla2l0LklDRVNlcnZlclIKaWNlU2VydmVycxItChJzdWJzY3JpYmVyX3ByaW1hcnkYBiABKAhSEXN1YnNjcmliZXJQcmltYXJ5'); @$core.Deprecated('Use trackPublishedResponseDescriptor instead') const TrackPublishedResponse$json = const { '1': 'TrackPublishedResponse', diff --git a/lib/src/room.dart b/lib/src/room.dart index b86b1a5..22e70eb 100644 --- a/lib/src/room.dart +++ b/lib/src/room.dart @@ -2,12 +2,13 @@ import 'dart:async'; import 'dart:collection'; import 'package:flutter/foundation.dart'; -import 'package:flutter_webrtc/flutter_webrtc.dart'; -import 'package:tuple/tuple.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 'participant/local_participant.dart'; import 'participant/participant.dart'; @@ -18,6 +19,7 @@ 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, @@ -100,10 +102,10 @@ mixin RoomDelegate { /// * active speakers are different /// {@category Room} class Room extends ChangeNotifier with ParticipantDelegate { - RoomState _state = RoomState.disconnected; + RoomState _connectionState = RoomState.disconnected; /// connection state of the room - RoomState get state => _state; + RoomState get state => _connectionState; final Map _participants = {}; @@ -131,7 +133,9 @@ class Room extends ChangeNotifier with ParticipantDelegate { final RTCEngine _engine; - Completer? _connectCompleter; + // suppport for multiple event listeners + final events = EventsEmitter(); + late final _engineListener = EventsListener(emitter: _engine.events); /// internal use /// {@nodoc} @@ -144,25 +148,29 @@ class Room extends ChangeNotifier with ParticipantDelegate { _engine.onDataMessage = _handleDataPacket; _engine.onRemoteMute = _onRemoteMuteChanged; _engine.onReconnected = () { - _state = RoomState.connected; + _connectionState = RoomState.connected; delegate?.onReconnected(); notifyListeners(); }; _engine.onReconnecting = () { - _state = RoomState.reconnecting; + _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 completer = Completer(); - _connectCompleter = completer; - final joinResponse = await _engine.join( url, token, @@ -185,19 +193,24 @@ class Room extends ChangeNotifier with ParticipantDelegate { _getOrCreateRemoteParticipant(info.sid, info); } - // room is not ready until ICE is connected. so we would return a completer for now - // if it times out, we'll fail the completer - Timer(const Duration(seconds: 5), () { - if (_state != RoomState.disconnected) { - return; - } - _state = RoomState.disconnected; - _connectCompleter?.completeError(ConnectError()); - _connectCompleter = null; - notifyListeners(); - }); + // room is not ready until ICE is connected. + try { + await _engineListener.waitFor( + filter: (event) => event.iceState.isConnected(), + duration: const Duration(seconds: 5), + onTimeout: () => throw ConnectException(), + ); - return completer.future; + // catch any exception + } catch (_) { + _connectionState = RoomState.disconnected; + notifyListeners(); + + // pass on the exception + rethrow; + } + + return this; } /// Disconnects from the room, notifying server of disconnection. @@ -206,6 +219,10 @@ class Room extends ChangeNotifier with ParticipantDelegate { await _handleDisconnect(); } + Future reconnect() async { + await _engine.reconnect(); + } + RemoteParticipant _getOrCreateRemoteParticipant(String sid, lk_models.ParticipantInfo? info) { var participant = _participants[sid]; if (participant != null) { @@ -224,16 +241,21 @@ class Room extends ChangeNotifier with ParticipantDelegate { } void _handleICEConnected() { - _connectCompleter?.complete(this); - _connectCompleter = null; - _state = RoomState.connected; + // _connectCompleter?.complete(this); + // _connectCompleter = null; + _connectionState = RoomState.connected; notifyListeners(); } Future _handleDisconnect() async { - if (_state == RoomState.disconnected) { + if (_connectionState == RoomState.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; for (final p in _participants.values) { final tracks = List.from(p.tracks.values); @@ -248,7 +270,7 @@ class Room extends ChangeNotifier with ParticipantDelegate { await _engine.close(); _participants.clear(); _activeSpeakers.clear(); - _state = RoomState.disconnected; + notifyListeners(); delegate?.onDisconnected(); } @@ -340,17 +362,23 @@ class Room extends ChangeNotifier with ParticipantDelegate { track?.muted = mute; } - void _onTrackAdded(MediaStreamTrack track, MediaStream? stream, RTCRtpReceiver? receiver) { + 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 parsed = _unpackStreamId(stream.id); - final trackSid = parsed.item2 ?? track.id; + final idParts = stream.id.split('|'); - final participant = _getOrCreateRemoteParticipant(parsed.item1, null); + final participantSid = idParts[0]; + final trackSid = idParts.elementAtOrNull(1) ?? track.id; + + final participant = _getOrCreateRemoteParticipant(participantSid, null); participant.addSubscribedMediaTrack(track, stream, trackSid); } @@ -415,11 +443,3 @@ class Room extends ChangeNotifier with ParticipantDelegate { delegate?.onTrackSubscriptionFailed(participant, sid, message); } } - -Tuple2 _unpackStreamId(String streamId) { - var parts = streamId.split('|'); - if (parts.length != 2) { - return Tuple2(parts[0], null); - } - return Tuple2(parts[0], parts[1]); -} diff --git a/lib/src/rtc_engine.dart b/lib/src/rtc_engine.dart index 3ce461c..221e4da 100644 --- a/lib/src/rtc_engine.dart +++ b/lib/src/rtc_engine.dart @@ -1,28 +1,28 @@ import 'dart:async'; -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:collection/collection.dart'; +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/delay.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 'signal_client.dart'; import 'track/track.dart'; import 'transport.dart'; - -const lossyDataChannel = '_lossy'; -const reliableDataChannel = '_reliable'; -const connectionTimeout = Duration(seconds: 5); -const maxReconnectAttempts = 5; -const iceRestartTimeout = Duration(seconds: 10); +import 'types.dart'; typedef GenericCallback = void Function(); typedef TrackCallback = void Function( - MediaStreamTrack track, - MediaStream? stream, - RTCRtpReceiver? receiver, + rtc.MediaStreamTrack track, + rtc.MediaStream? stream, + rtc.RTCRtpReceiver? receiver, ); typedef ParticipantUpdateCallback = void Function(List participants); typedef ActiveSpeakerChangedCallback = void Function(List speakers); @@ -31,25 +31,42 @@ typedef DataPacketCallback = void Function( typedef RemoteMuteCallback = void Function(String sid, bool mute); class RTCEngine with SignalClientDelegate { + 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; + // config for RTCPeerConnection + final RTCConfiguration? rtcConfig; + PCTransport? publisher; PCTransport? subscriber; - SignalClient client; - // config for RTCPeerConnection - RTCConfiguration rtcConfig = RTCConfiguration(); + PCTransport? get primary => _subscriberPrimary ? subscriber : publisher; + + // used for ice state notifications + CancelListenFunc? _primaryIceStateListener; + // data channels for packets - RTCDataChannel? reliableDC; - RTCDataChannel? lossyDC; + rtc.RTCDataChannel? reliableDC; + rtc.RTCDataChannel? lossyDC; bool iceConnected = false; bool isReconnecting = false; bool isClosed = true; - Map> pendingTrackResolvers = {}; - int reconnectAttempts = 0; - // to complete join request - Completer? joinCompleter; + // true if publisher connection has already been established. + // this is helpful to know if we need to restart ICE on the publisher connection + bool _hasPublished = false; + // remember url and token for reconnect String? url; String? token; + bool _subscriberPrimary = false; + // server-provided ice servers + List _providedIceServers = []; + // delegate methods GenericCallback? onICEConnected; TrackCallback? onTrack; @@ -61,12 +78,29 @@ class RTCEngine with SignalClientDelegate { GenericCallback? onReconnected; GenericCallback? onDisconnected; - RTCEngine(this.client, RTCConfiguration? rtcConfig) { - if (rtcConfig != null) { - this.rtcConfig = rtcConfig; - } + // + // internal + // + final Map> _pendingTrackResolvers = {}; + int _reconnectAttempts = 0; + // to complete join request + Completer? _joinCompleter; + final events = EventsEmitter(); + + final delays = CancelableDelayManager(); + + RTCEngine( + this.client, + 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')); + } } Future join( @@ -78,22 +112,36 @@ class RTCEngine with SignalClientDelegate { this.token = token; final completer = Completer(); - joinCompleter = completer; + _joinCompleter = completer; await client.join(url, token, options: options); // if it's not complete after 5 seconds, fail - Timer(connectionTimeout, () { - joinCompleter?.completeError(ConnectError()); - joinCompleter = null; + Timer(_connectionTimeout, () { + _joinCompleter?.completeError(ConnectException()); + _joinCompleter = null; }); return completer.future; } Future close() async { + logger.fine('${objectId} close()'); + if (isClosed) { + logger.fine('${objectId} close() already closed'); + return; + } isClosed = true; + // cancel events + await _primaryIceStateListener?.call(); + _primaryIceStateListener = null; + + await events.dispose(); + + // cancel all ongoing delays + await delays.dispose(); + // PCTransport is responsible for disposing RTCPeerConnection await publisher?.dispose(); publisher = null; @@ -110,12 +158,12 @@ class RTCEngine with SignalClientDelegate { required lk_models.TrackType kind, TrackDimension? dimension, }) async { - if (pendingTrackResolvers[cid] != null) { - throw TrackPublishError('a track with the same CID has already been published'); + if (_pendingTrackResolvers[cid] != null) { + throw TrackPublishException('a track with the same CID has already been published'); } final completer = Completer(); - pendingTrackResolvers[cid] = completer; + _pendingTrackResolvers[cid] = completer; client.sendAddTrack(cid: cid, name: name, type: kind, dimension: dimension); @@ -123,187 +171,277 @@ class RTCEngine with SignalClientDelegate { } Future negotiate({bool? iceRestart}) async { - final pub = publisher; - if (pub == null) return; - - final remoteDesc = await pub.getRemoteDescription(); - - // handle cases that we couldn't create a new offer due to a pending answer - // that's lost in transit - if (remoteDesc != null && - pub.pc.signalingState == RTCSignalingState.RTCSignalingStateHaveLocalOffer) { - await pub.pc.setRemoteDescription(remoteDesc); + if (publisher == null) { + return; } - final constraints = {}; - if (iceRestart != null && iceRestart) { - constraints['mandatory'] = { - 'IceRestart': true, - }; + _hasPublished = true; + publisher!.negotiate(); + } + + /* @internal */ + Future sendDataPacket( + lk_models.DataPacket packet, + ) async { + // make sure we do have a data connection + await _ensurePublisherConnected(); + + final dcMessage = rtc.RTCDataChannelMessage.fromBinary(packet.writeToBuffer()); + + if (packet.kind == lk_models.DataPacket_Kind.LOSSY && lossyDC != null) { + await lossyDC?.send(dcMessage); + } else if (packet.kind == lk_models.DataPacket_Kind.RELIABLE && reliableDC != null) { + await reliableDC?.send(dcMessage); } - final offer = await pub.pc.createOffer(constraints); - logger.fine('Created offer'); - logger.finer('sdp: ${offer.sdp}'); - await pub.pc.setLocalDescription(offer); - client.sendOffer(offer); + } + + Future _ensurePublisherConnected() async { + logger.fine('ensurePublisherConnected()'); + if (!_subscriberPrimary) { + return; + } + + if (publisher?.pc.iceConnectionState?.isConnected() == true) { + logger.warning('publisher is already connected'); + return; + } + + // start negotiation + await negotiate(); + + logger.fine('[PUBLISHER] waiting for to ice-connect ' + '(current: ${publisher?.pc.iceConnectionState})'); + + await events.waitFor( + filter: (event) => event.iceState.isConnected(), + duration: _maxICEConnectTimeout, + ); + + logger.fine('[PUBLISHER] connected'); } Future reconnect() async { - if (isClosed) return; + if (isClosed) { + logger.fine('$objectId reconnect() already closed'); + return; + } final url = this.url; final token = this.token; + if (url == null || token == null) { - throw ConnectError('could not reconnect without url and token'); + throw ConnectException('could not reconnect without url and token'); } - if (reconnectAttempts == 0) { + + if (_reconnectAttempts == 0) { onReconnecting?.call(); + events.emit(EngineReconnectingEvent()); } - reconnectAttempts++; + _reconnectAttempts++; try { isReconnecting = true; await client.reconnect(url, token); - final pub = publisher; - final sub = subscriber; - if (pub == null || sub == null) { - throw UnexpectedConnectionState('publisher or subscribers is null'); + if (publisher == null || subscriber == null) { + throw UnexpectedStateException('publisher or subscribers is null'); } - pub.restartingIce = true; - sub.restartingIce = true; + subscriber!.restartingIce = true; - await negotiate(iceRestart: true); - } catch (error) { + // await negotiate(iceRestart: true); + if (_hasPublished) { + logger.fine('reconnect: publisher.createAndSendOffer'); + await publisher!.createAndSendOffer(const RTCOfferOptions(iceRestart: true)); + } + + if (!(primary?.pc.iceConnectionState?.isConnected() ?? false)) { + logger.fine('reconnect: waiting for primary to ice-connect...'); + + await events.waitFor( + filter: (event) => event.isPrimary && event.iceState.isConnected(), + duration: _iceRestartTimeout, + ); + } + + logger.fine('reconnect: success'); + events.emit(EngineReconnectedEvent()); + _reconnectAttempts = 0; + + // don't catch and pass up any exception + } finally { + // always set reconnecting to false isReconnecting = false; - return Future.error(error); } - - // wait for connectivity to change - final startTime = DateTime.now(); - while (DateTime.now().difference(startTime) < iceRestartTimeout) { - if (iceConnected) { - isReconnecting = false; - return; - } - await Future.delayed(const Duration(milliseconds: 100)); - } - - isReconnecting = false; - throw ConnectError('could not reconnect ICE'); } Future _configurePeerConnections() async { - if (publisher != null) { + if (publisher != null || subscriber != null) { + logger.warning('Already configured'); return; } - final pubPC = await createPeerConnection(rtcConfig.toMap()); - publisher = PCTransport(pubPC); - final subPC = await createPeerConnection(rtcConfig.toMap()); - subscriber = PCTransport(subPC); + RTCConfiguration? config; + // use server-provided iceServers if not provided by user + if ((rtcConfig?.iceServers?.isEmpty ?? true) && _providedIceServers.isNotEmpty) { + final iceServers = _providedIceServers.map((e) => e.toSDKType()).toList(); + config = (rtcConfig ?? const RTCConfiguration()).copyWith(iceServers: iceServers); + } - pubPC.onIceCandidate = (RTCIceCandidate candidate) { + publisher = await PCTransport.create(config); + subscriber = await PCTransport.create(config); + + publisher?.pc.onIceCandidate = (rtc.RTCIceCandidate candidate) { + logger.fine('publisher onIceCandidate'); client.sendIceCandidate(candidate, lk_rtc.SignalTarget.PUBLISHER); }; - subPC.onIceCandidate = (RTCIceCandidate candidate) { + + subscriber?.pc.onIceCandidate = (rtc.RTCIceCandidate candidate) { + logger.fine('subscriber onIceCandidate'); client.sendIceCandidate(candidate, lk_rtc.SignalTarget.SUBSCRIBER); }; - pubPC.onRenegotiationNeeded = () async { - if (pubPC.iceConnectionState == null || - pubPC.iceConnectionState == RTCIceConnectionState.RTCIceConnectionStateNew) { - return; - } - await negotiate(); + publisher?.onOffer = (offer) { + logger.fine('publisher onOffer'); + client.sendOffer(offer); }; - pubPC.onIceConnectionState = (RTCIceConnectionState state) { - if (publisher == null) { - return; - } - switch (state) { - case RTCIceConnectionState.RTCIceConnectionStateConnected: - if (!iceConnected) { - iceConnected = true; - if (isReconnecting) { - onReconnected?.call(); - } else { - onICEConnected?.call(); - } + // in subscriber primary mode, server side opens sub data channels. + if (_subscriberPrimary) { + subscriber?.pc.onDataChannel = _onDataChannel; + } + + // logger.fine('subscriber.pc: ${subscriber?.pc}'); + subscriber?.pc.onIceConnectionState = (state) { + // + events.emit(EngineSubscriberIceStateUpdatedEvent( + state: state, + isPrimary: _subscriberPrimary, + )); + }; + + publisher?.pc.onIceConnectionState = (state) { + // + events.emit(EnginePublisherIceStateUpdatedEvent( + state: state, + isPrimary: !_subscriberPrimary, + )); + }; + + _primaryIceStateListener ??= events.on((event) { + // only listen to primary ice events + if (!event.isPrimary) return; + + if (event.iceState == rtc.RTCIceConnectionState.RTCIceConnectionStateConnected) { + if (!iceConnected) { + iceConnected = true; + if (isReconnecting) { + onReconnected?.call(); + } else { + onICEConnected?.call(); + events.emit(EngineConnectedEvent()); } - break; - - case RTCIceConnectionState.RTCIceConnectionStateFailed: + } + } else if (event.iceState == rtc.RTCIceConnectionState.RTCIceConnectionStateFailed) { + // trigger reconnect sequence + if (iceConnected) { iceConnected = false; - // trigger reconnect sequence - _handleDisconnect('peerconnection'); - break; - - default: - // do nothing + _onDisconnected('peerconnection'); + } } + }); + + subscriber?.pc.onTrack = (rtc.RTCTrackEvent event) { + onTrack?.call(event.track, event.streams.firstOrNull, event.receiver); + events.emit(EngineMediaTrackAddedEvent( + track: event.track, + stream: event.streams.firstOrNull, + receiver: event.receiver, + )); }; - subPC.onTrack = (RTCTrackEvent event) { - onTrack?.call(event.track, event.streams.first, event.receiver); - }; - - // create data channels - final lossyInit = RTCDataChannelInit() - ..maxRetransmits = 1 + // data channels + final lossyInit = rtc.RTCDataChannelInit() + ..binaryType = 'binary' ..ordered = true - ..binaryType = 'binary'; - lossyDC = await pubPC.createDataChannel(lossyDataChannel, lossyInit); + ..maxRetransmits = 0; + lossyDC = await publisher?.pc.createDataChannel(_lossyDCLabel, lossyInit); - final reliableInit = RTCDataChannelInit() - ..ordered = true - ..maxRetransmits = 50 - ..binaryType = 'binary'; - reliableDC = await pubPC.createDataChannel(reliableDataChannel, reliableInit); + final reliableInit = rtc.RTCDataChannelInit() + ..binaryType = 'binary' + ..ordered = true; + reliableDC = await publisher?.pc.createDataChannel(_reliableDCLabel, reliableInit); - lossyDC?.onMessage = _handleDataMessage; - reliableDC?.onMessage = _handleDataMessage; + // also handle messages over the pub channel, for backwards compatibility + lossyDC?.onMessage = _onDCMessage; + reliableDC?.onMessage = _onDCMessage; } - void _handleDataMessage(RTCDataChannelMessage message) { + void _onDataChannel(rtc.RTCDataChannel dc) { + switch (dc.label) { + case _reliableDCLabel: + logger.fine('Server opened DC label: ${dc.label}'); + reliableDC = dc; + reliableDC?.onMessage = _onDCMessage; + break; + case _lossyDCLabel: + logger.fine('Server opened DC label: ${dc.label}'); + lossyDC = dc; + lossyDC?.onMessage = _onDCMessage; + break; + default: + logger.warning('Unknown DC label: ${dc.label}'); + break; + } + } + + void _onDCMessage(rtc.RTCDataChannelMessage message) { // always expect binary if (!message.isBinary) { + logger.warning('Data message is not binary'); return; } final dp = lk_models.DataPacket.fromBuffer(message.binary); - switch (dp.whichValue()) { - case lk_models.DataPacket_Value.speaker: - onActiveSpeakerUpdated?.call(dp.speaker.speakers); - break; - case lk_models.DataPacket_Value.user: - onDataMessage?.call(dp.user, dp.kind); - break; - default: - // do nothing + 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, + )); } } - Future _handleDisconnect(String reason) async { + Future _onDisconnected(String reason) async { if (isClosed) return; logger.fine('disconnected $reason'); - if (reconnectAttempts >= maxReconnectAttempts) { - logger.info('could not connect after $reconnectAttempts, giving up'); + if (_reconnectAttempts >= _maxReconnectAttempts) { + logger.info('could not connect after $_reconnectAttempts, giving up'); await close(); onDisconnected?.call(); + events.emit(EngineDisconnectedEvent()); return; } - final delay = (reconnectAttempts * reconnectAttempts) * 300; - Future.delayed(Duration(milliseconds: delay), () { - reconnect().then((_) { - reconnectAttempts = 0; - }).catchError((dynamic e) { - _handleDisconnect(reason); - }); + final delay = Duration(milliseconds: (_reconnectAttempts * _reconnectAttempts) * 300); + + // if this instance is disposed, we probably don't want to continue any more + // so the whole block will be canceled from being executed + await delays.waitFor(delay, ifNotCancelled: () async { + try { + await reconnect(); + _reconnectAttempts = 0; + } catch (_) { + // doesn't need to be awaited + // ignore: unawaited_futures + _onDisconnected(reason); + } }); } @@ -313,90 +451,100 @@ class RTCEngine with SignalClientDelegate { Future onConnected(lk_rtc.JoinResponse response) async { // create peer connections isClosed = false; + _subscriberPrimary = response.subscriberPrimary; + _providedIceServers = response.iceServers; - if (rtcConfig.iceServers == null && response.iceServers.isNotEmpty) { - List iceServers = []; - for (final item in response.iceServers) { - final iceServer = RTCIceServer(urls: item.urls); - if (item.username.isNotEmpty) { - iceServer.username = item.username; - } - if (item.credential.isNotEmpty) { - iceServer.credential = item.credential; - } - iceServers.add(iceServer); - } - rtcConfig.iceServers = iceServers; - } + logger.fine('onConnected subscriberPrimary: ${_subscriberPrimary}, ' + 'serverVersion: ${response.serverVersion}, ' + 'iceServers: ${response.iceServers}'); await _configurePeerConnections(); - await negotiate(); + if (!_subscriberPrimary) { + // for subscriberPrimary, we negotiate when necessary (lazy) + await negotiate(); + } - joinCompleter?.complete(Future.value(response)); - joinCompleter = null; + _joinCompleter?.complete(Future.value(response)); + _joinCompleter = null; } @override Future onClose([String? reason]) async { - await _handleDisconnect('signal'); + await _onDisconnected('signal'); } @override - Future onOffer(RTCSessionDescription sd) async { - final sub = subscriber; - if (sub == null) return; + Future onOffer(rtc.RTCSessionDescription sd) async { + if (subscriber == null) { + return; + } - await sub.setRemoteDescription(sd); + logger.fine('received server offer(type: ${sd.type}, ${subscriber!.pc.signalingState})'); - final answer = await sub.pc.createAnswer(); + await subscriber!.setRemoteDescription(sd); + + final answer = await subscriber!.pc.createAnswer(); logger.fine('Created answer'); logger.finer('sdp: ${answer.sdp}'); - await sub.pc.setLocalDescription(answer); + await subscriber!.pc.setLocalDescription(answer); client.sendAnswer(answer); } @override - Future onAnswer(RTCSessionDescription sd) async { - if (publisher == null) return; - logger.fine('Received answer'); + 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(RTCIceCandidate candidate, lk_rtc.SignalTarget target) async { + 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); + await subscriber!.addIceCandidate(candidate); } else if (target == lk_rtc.SignalTarget.PUBLISHER) { - await publisher?.addIceCandidate(candidate); + 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); + 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, + )); } } diff --git a/lib/src/signal_client.dart b/lib/src/signal_client.dart index 34ccd42..e6975a5 100644 --- a/lib/src/signal_client.dart +++ b/lib/src/signal_client.dart @@ -2,17 +2,19 @@ import 'dart:async'; import 'dart:convert'; import 'dart:developer'; -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'package:http/http.dart' as http; -import 'package:livekit_client/src/ws/interface.dart'; import 'package:synchronized/synchronized.dart' as sync; import 'errors.dart'; +import 'extensions.dart'; import 'logger.dart'; import 'options.dart'; import 'proto/livekit_models.pb.dart' as lk_models; import 'proto/livekit_rtc.pb.dart' as lk_rtc; import 'track/track.dart'; +import 'utils.dart'; +import 'ws/interface.dart'; mixin SignalClientDelegate { // initial connection established @@ -20,11 +22,11 @@ mixin SignalClientDelegate { // websocket has closed Future onClose([String? reason]); // when a server offer is received - Future onOffer(RTCSessionDescription sd); + Future onOffer(rtc.RTCSessionDescription sd); // when an answer from server is received - Future onAnswer(RTCSessionDescription sd); + Future onAnswer(rtc.RTCSessionDescription sd); // when server has a new ICE candidate - Future onTrickle(RTCIceCandidate candidate, lk_rtc.SignalTarget target); + Future onTrickle(rtc.RTCIceCandidate candidate, lk_rtc.SignalTarget target); // participant has changed Future onParticipantUpdate(List updates); // when a track has been added successfully @@ -37,48 +39,20 @@ mixin SignalClientDelegate { Future onMuteTrack(lk_rtc.MuteTrackRequest req); } -extension LKUriExt on Uri { - bool get isSecureScheme => ['https', 'wss'].contains(scheme); -} - class SignalClient { - static const protocolVersion = 2; - final _lock = sync.Lock(); + + ProtocolVersion protocol; SignalClientDelegate? delegate; bool _connected = false; - LKWebSocket? _ws; + LiveKitWebSocket? _ws; - SignalClient(); + SignalClient({ + this.protocol = ProtocolVersion.protocol3, + }); bool get connected => _connected; - Uri _buildUri( - String uriOrString, { - required String token, - ConnectOptions? options, - bool reconnect = false, - bool validate = false, - bool forceSecure = false, - }) { - final Uri uri = Uri.parse(uriOrString); - - final useSecure = uri.isSecureScheme || forceSecure; - final httpScheme = useSecure ? 'https' : 'http'; - final wsScheme = useSecure ? 'wss' : 'ws'; - - return uri.replace( - scheme: validate ? httpScheme : wsScheme, - path: validate ? 'validate' : 'rtc', - queryParameters: { - 'access_token': token, - if (options != null) 'auto_subscribe': options.autoSubscribe ? '1' : '0', - if (reconnect) 'reconnect': '1', - 'protocol': protocolVersion.toString(), - }, - ); - } - Future join( String uriString, String token, { @@ -87,16 +61,17 @@ class SignalClient { // Create default options if null options ??= const ConnectOptions(); - final rtcUri = _buildUri( + final rtcUri = Utils.buildUri( uriString, token: token, options: options, + protocol: protocol, ); try { - _ws = await LKWebSocket.connect( + _ws = await LiveKitWebSocket.connect( rtcUri, - LKWebSocketOptions( + WebSocketOptions( onData: _onSocketData, onDispose: _onSocketDone, onError: _handleError, @@ -104,24 +79,25 @@ class SignalClient { ); } catch (socketError) { // Re-build same uri for validate mode - final validateUri = _buildUri( + final validateUri = Utils.buildUri( uriString, token: token, options: options, validate: true, forceSecure: rtcUri.isSecureScheme, + protocol: protocol, ); // Attempt Validation try { final validateResponse = await http.get(validateUri); - if (validateResponse.statusCode != 200) throw ConnectError(validateResponse.body); - throw ConnectError(); + if (validateResponse.statusCode != 200) throw ConnectException(validateResponse.body); + throw ConnectException(); } catch (error) { // Pass it up if it's already a `ConnectError` - if (error is ConnectError) rethrow; + if (error is ConnectException) rethrow; // HTTP doesn't work either - throw ConnectError(); + throw ConnectException(); } } } @@ -134,15 +110,16 @@ class SignalClient { _ws?.dispose(); _ws = null; - final rtcUri = _buildUri( + final rtcUri = Utils.buildUri( uriString, token: token, reconnect: true, + protocol: protocol, ); - _ws = await LKWebSocket.connect( + _ws = await LiveKitWebSocket.connect( rtcUri, - LKWebSocketOptions( + WebSocketOptions( onData: _onSocketData, onDispose: _onSocketDone, onError: _handleError, @@ -157,18 +134,18 @@ class SignalClient { _ws?.dispose(); } - void sendOffer(RTCSessionDescription offer) => _sendRequest(lk_rtc.SignalRequest( - offer: fromRTCSessionDescription(offer), + void sendOffer(rtc.RTCSessionDescription offer) => _sendRequest(lk_rtc.SignalRequest( + offer: offer.toSDKType(), )); - void sendAnswer(RTCSessionDescription answer) => _sendRequest(lk_rtc.SignalRequest( - answer: fromRTCSessionDescription(answer), + void sendAnswer(rtc.RTCSessionDescription answer) => _sendRequest(lk_rtc.SignalRequest( + answer: answer.toSDKType(), )); - void sendIceCandidate(RTCIceCandidate candidate, lk_rtc.SignalTarget target) => _sendRequest( + void sendIceCandidate(rtc.RTCIceCandidate candidate, lk_rtc.SignalTarget target) => _sendRequest( lk_rtc.SignalRequest( trickle: lk_rtc.TrickleRequest( - candidateInit: fromRTCIceCandidate(candidate), + candidateInit: candidate.toJson(), target: target, ), ), @@ -249,14 +226,14 @@ class SignalClient { } break; case lk_rtc.SignalResponse_Message.answer: - await delegate?.onAnswer(toRTCSessionDescription(msg.answer)); + await delegate?.onAnswer(msg.answer.toSDKType()); break; case lk_rtc.SignalResponse_Message.offer: - await delegate?.onOffer(toRTCSessionDescription(msg.offer)); + await delegate?.onOffer(msg.offer.toSDKType()); break; case lk_rtc.SignalResponse_Message.trickle: await delegate?.onTrickle( - toRTCIceCandidate(msg.trickle.candidateInit), + RTCIceCandidateExt.fromJson(msg.trickle.candidateInit), msg.trickle.target, ); break; @@ -292,24 +269,3 @@ class SignalClient { delegate?.onClose(); } } - -RTCSessionDescription toRTCSessionDescription(lk_rtc.SessionDescription sd) { - return RTCSessionDescription(sd.sdp, sd.type); -} - -lk_rtc.SessionDescription fromRTCSessionDescription(RTCSessionDescription rsd) { - return lk_rtc.SessionDescription(type: rsd.type, sdp: rsd.sdp); -} - -RTCIceCandidate toRTCIceCandidate(String candidateInit) { - final candInit = json.decode(candidateInit) as Map; - return RTCIceCandidate( - candInit['candidate'] as String?, - candInit['sdpMid'] as String?, - candInit['sdpMLineIndex'] as int?, - ); -} - -String fromRTCIceCandidate(RTCIceCandidate candidate) { - return json.encode(candidate.toMap()); -} diff --git a/lib/src/track/_audio_api.dart b/lib/src/track/_audio_api.dart index c6a2923..599e4ae 100644 --- a/lib/src/track/_audio_api.dart +++ b/lib/src/track/_audio_api.dart @@ -1,6 +1,6 @@ -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; -void startAudio(String id, MediaStreamTrack stream) { +void startAudio(String id, rtc.MediaStreamTrack stream) { // do nothing } diff --git a/lib/src/track/_audio_html.dart b/lib/src/track/_audio_html.dart index ccc8ebe..aa0e0e4 100644 --- a/lib/src/track/_audio_html.dart +++ b/lib/src/track/_audio_html.dart @@ -1,14 +1,14 @@ // ignore: avoid_web_libraries_in_flutter import 'dart:html' as html; -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; // ignore: implementation_imports import 'package:flutter_webrtc/src/web/media_stream_track_impl.dart'; const audioContainerId = 'livekit_audio_container'; const audioPrefix = 'livekit_audio_'; -void startAudio(String id, MediaStreamTrack track) { +void startAudio(String id, rtc.MediaStreamTrack track) { if (track is! MediaStreamTrackWeb) { return; } diff --git a/lib/src/track/audio_track.dart b/lib/src/track/audio_track.dart index a196eda..057b3d6 100644 --- a/lib/src/track/audio_track.dart +++ b/lib/src/track/audio_track.dart @@ -1,4 +1,4 @@ -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import '../proto/livekit_models.pb.dart' as lk_models; import '_audio_api.dart' if (dart.library.html) '_audio_html.dart' as audio; @@ -6,9 +6,9 @@ import 'local_audio_track.dart'; import 'track.dart'; class AudioTrack extends Track { - MediaStream? mediaStream; + rtc.MediaStream? mediaStream; - AudioTrack(String name, MediaStreamTrack track, this.mediaStream) + AudioTrack(String name, rtc.MediaStreamTrack track, this.mediaStream) : super(lk_models.TrackType.AUDIO, name, track); /// Start playing audio track. On web platform, create an audio element and diff --git a/lib/src/track/local_audio_track.dart b/lib/src/track/local_audio_track.dart index e926302..830090c 100644 --- a/lib/src/track/local_audio_track.dart +++ b/lib/src/track/local_audio_track.dart @@ -1,6 +1,6 @@ import 'dart:async'; -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import '../errors.dart'; import 'audio_track.dart'; @@ -9,19 +9,19 @@ import 'options.dart'; class LocalAudioTrack extends AudioTrack { LocalAudioTrack( String name, - MediaStreamTrack track, - MediaStream stream, + rtc.MediaStreamTrack track, + rtc.MediaStream stream, ) : super(name, track, stream); /// Creates a new audio track from the default audio input device. static Future create([LocalAudioTrackOptions? options]) async { // try { - final stream = await navigator.mediaDevices.getUserMedia({ + final stream = await rtc.navigator.mediaDevices.getUserMedia({ 'audio': true, 'video': false, }); - if (stream.getAudioTracks().isEmpty) throw TrackCreateError(); + if (stream.getAudioTracks().isEmpty) throw TrackCreateException(); return LocalAudioTrack('', stream.getAudioTracks().first, stream); } diff --git a/lib/src/track/local_track_publication.dart b/lib/src/track/local_track_publication.dart index 6945b69..0c9e725 100644 --- a/lib/src/track/local_track_publication.dart +++ b/lib/src/track/local_track_publication.dart @@ -1,5 +1,4 @@ -import 'package:livekit_client/src/logger.dart'; - +import '../logger.dart'; import '../participant/local_participant.dart'; import '../proto/livekit_models.pb.dart' as lk_models; import 'track.dart'; diff --git a/lib/src/track/local_video_track.dart b/lib/src/track/local_video_track.dart index 8b45857..ae8e106 100644 --- a/lib/src/track/local_video_track.dart +++ b/lib/src/track/local_video_track.dart @@ -1,4 +1,4 @@ -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import '../errors.dart'; import '../logger.dart'; @@ -19,19 +19,19 @@ class LocalVideoTrack extends VideoTrack { // LocalVideoTrack._( String name, - MediaStreamTrack mediaTrack, - MediaStream stream, + rtc.MediaStreamTrack mediaTrack, + rtc.MediaStream stream, this.currentOptions, ) : super(name, mediaTrack, stream); - RTCRtpSender? get sender => transceiver?.sender; + rtc.RTCRtpSender? get sender => transceiver?.sender; /// Restarts the track with new options. This is useful when switching between /// front and back cameras. Future restartTrack([ LocalVideoTrackOptions? options, ]) async { - if (sender == null) throw TrackCreateError('could not restart track'); + if (sender == null) throw TrackCreateException('could not restart track'); if (options != null && currentOptions.runtimeType != options.runtimeType) { throw Exception('options must be a ${currentOptions.runtimeType}'); } @@ -73,7 +73,7 @@ class LocalVideoTrack extends VideoTrack { ); } - static Future _createStream( + static Future _createStream( LocalVideoTrackOptions options, ) async { final constraints = { @@ -81,15 +81,15 @@ class LocalVideoTrack extends VideoTrack { 'video': options.toMediaConstraintsMap(), }; - final MediaStream stream; + final rtc.MediaStream stream; if (options is ScreenTrackOptions) { - stream = await navigator.mediaDevices.getDisplayMedia(constraints); + stream = await rtc.navigator.mediaDevices.getDisplayMedia(constraints); } else { // options is CameraVideoTrackOptions - stream = await navigator.mediaDevices.getUserMedia(constraints); + stream = await rtc.navigator.mediaDevices.getUserMedia(constraints); } - if (stream.getVideoTracks().isEmpty) throw TrackCreateError(); + if (stream.getVideoTracks().isEmpty) throw TrackCreateException(); return stream; } } @@ -97,7 +97,7 @@ class LocalVideoTrack extends VideoTrack { // // Convenience extensions // -extension LKLocalVideoTrackExt on LocalVideoTrack { +extension LocalVideoTrackExt on LocalVideoTrack { // Calls restartTrack under the hood Future setCameraPosition(CameraPosition position) async { final options = currentOptions; diff --git a/lib/src/track/options.dart b/lib/src/track/options.dart index 4a8b907..7ed6ca8 100644 --- a/lib/src/track/options.dart +++ b/lib/src/track/options.dart @@ -1,4 +1,4 @@ -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; enum LocalVideoTrackType { camera, @@ -10,7 +10,7 @@ enum CameraPosition { back, } -extension LKCameraPositionExt on CameraPosition { +extension CameraPositionExt on CameraPosition { CameraPosition swap() => { CameraPosition.front: CameraPosition.back, CameraPosition.back: CameraPosition.front, @@ -74,12 +74,12 @@ class VideoEncoding { } extension VideoEncodingExt on VideoEncoding { - RTCRtpEncoding toRTCRtpEncoding({ + rtc.RTCRtpEncoding toRTCRtpEncoding({ String? rid, double? scaleResolutionDownBy = 1.0, int? numTemporalLayers, }) => - RTCRtpEncoding( + rtc.RTCRtpEncoding( rid: rid, scaleResolutionDownBy: scaleResolutionDownBy, maxFramerate: maxFramerate, diff --git a/lib/src/track/track.dart b/lib/src/track/track.dart index 15ec331..4a89044 100644 --- a/lib/src/track/track.dart +++ b/lib/src/track/track.dart @@ -1,4 +1,4 @@ -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'package:uuid/uuid.dart'; import '../proto/livekit_models.pb.dart' as lk_models; @@ -17,24 +17,24 @@ class Track { String name; lk_models.TrackType kind; - MediaStreamTrack mediaStreamTrack; + rtc.MediaStreamTrack mediaStreamTrack; String? sid; - RTCRtpTransceiver? transceiver; + rtc.RTCRtpTransceiver? transceiver; String? _cid; Track(this.kind, this.name, this.mediaStreamTrack); bool get muted => mediaStreamTrack.muted == null ? false : mediaStreamTrack.muted!; - RTCRtpMediaType get mediaType { + rtc.RTCRtpMediaType get mediaType { switch (kind) { case lk_models.TrackType.AUDIO: - return RTCRtpMediaType.RTCRtpMediaTypeAudio; + return rtc.RTCRtpMediaType.RTCRtpMediaTypeAudio; case lk_models.TrackType.VIDEO: - return RTCRtpMediaType.RTCRtpMediaTypeVideo; + return rtc.RTCRtpMediaType.RTCRtpMediaTypeVideo; // this should never happen default: - return RTCRtpMediaType.RTCRtpMediaTypeAudio; + return rtc.RTCRtpMediaType.RTCRtpMediaTypeAudio; } } diff --git a/lib/src/track/video_track.dart b/lib/src/track/video_track.dart index 960bfa9..bd0e7d5 100644 --- a/lib/src/track/video_track.dart +++ b/lib/src/track/video_track.dart @@ -1,17 +1,17 @@ import 'package:flutter/foundation.dart'; import 'package:flutter/material.dart'; -import 'package:flutter_webrtc/flutter_webrtc.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 { - MediaStream _mediaStream; + rtc.MediaStream _mediaStream; VideoTrack( String name, - MediaStreamTrack mediaTrack, + rtc.MediaStreamTrack mediaTrack, this._mediaStream, ) : super( lk_models.TrackType.VIDEO, @@ -19,11 +19,11 @@ class VideoTrack extends Track with ChangeNotifier { mediaTrack, ); - MediaStream get mediaStream => _mediaStream; + rtc.MediaStream get mediaStream => _mediaStream; /// internal use /// {@nodoc} - void setMediaStream(MediaStream stream) { + void setMediaStream(rtc.MediaStream stream) { _mediaStream = stream; notifyListeners(); } diff --git a/lib/src/transport.dart b/lib/src/transport.dart index 70c4b82..62860de 100644 --- a/lib/src/transport.dart +++ b/lib/src/transport.dart @@ -1,49 +1,120 @@ -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'dart:async'; + +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'logger.dart'; +import 'types.dart'; +import 'utils.dart'; +import 'extensions.dart'; + +typedef PCTransportOnOffer = void Function(rtc.RTCSessionDescription offer); /// a wrapper around PeerConnection class PCTransport { - final RTCPeerConnection pc; - final List _pendingCandidates = []; + final rtc.RTCPeerConnection pc; + final List _pendingCandidates = []; bool restartingIce = false; + bool renegotiate = false; + PCTransportOnOffer? onOffer; + Function? _cancelDebounce; - PCTransport(this.pc); + // private constructor + PCTransport._(this.pc); + + static Future create([RTCConfiguration? rtcConfig]) async { + rtcConfig ??= const RTCConfiguration(); + logger.fine('PCTransport creating ${rtcConfig.toMap()}'); + final _ = await rtc.createPeerConnection(rtcConfig.toMap()); + return PCTransport._(_); + } + + late final negotiate = Utils.createDebounceFunc( + () => createAndSendOffer(), + cancelFunc: (f) => _cancelDebounce = f, + wait: const Duration(milliseconds: 100), + ); Future dispose() async { + logger.fine('${objectId} dispose()'); + // Ensure debounce won't fire + _cancelDebounce?.call(); + _cancelDebounce = null; + // Ensure callbacks won't fire any more pc.onRenegotiationNeeded = null; pc.onIceCandidate = null; pc.onIceConnectionState = null; pc.onTrack = null; - List senders = []; + // Remove all senders + List senders = []; try { senders = await pc.getSenders(); - } catch (_) {} + } catch (_) { + logger.warning('getSenders() failed with error: $_'); + } for (final e in senders) { try { await pc.removeTrack(e); - } catch (_) {} + } catch (_) { + logger.warning('removeTrack() failed with error: $_'); + } } await pc.close(); await pc.dispose(); } - Future setRemoteDescription(RTCSessionDescription sd) async { + Future setRemoteDescription(rtc.RTCSessionDescription sd) async { await pc.setRemoteDescription(sd); - await Future.forEach(_pendingCandidates, (candidate) async { + for (final candidate in _pendingCandidates) { await pc.addCandidate(candidate); - }); + } _pendingCandidates.clear(); restartingIce = false; + + if (renegotiate) { + renegotiate = false; + await createAndSendOffer(); // await or un-awaited ? + } } - Future addIceCandidate(RTCIceCandidate candidate) async { + Future createAndSendOffer([RTCOfferOptions? options]) async { + if (onOffer == null) { + logger.warning('onOffer is null'); + return; + } + + if (options?.iceRestart ?? false) { + logger.fine('restarting ICE'); + restartingIce = true; + } + + if (pc.signalingState == rtc.RTCSignalingState.RTCSignalingStateHaveLocalOffer) { + // we're waiting for the peer to accept our offer, so we'll just wait + // the only exception to this is when ICE restart is needed + final currentSD = await getRemoteDescription(); + if ((options?.iceRestart ?? false) && currentSD != null) { + // TODO: handle when ICE restart is needed but we don't have a remote description + // the best thing to do is to recreate the peerconnection + await pc.setRemoteDescription(currentSD); + } else { + renegotiate = true; + return; + } + } + + // actually negotiate + logger.fine('starting to negotiate'); + final offer = await pc.createOffer(options?.toMap() ?? {}); + await pc.setLocalDescription(offer); + onOffer?.call(offer); + } + + Future addIceCandidate(rtc.RTCIceCandidate candidate) async { final desc = await getRemoteDescription(); if (desc != null && !restartingIce) { @@ -54,15 +125,16 @@ class PCTransport { _pendingCandidates.add(candidate); } - Future getRemoteDescription() async { + Future getRemoteDescription() async { // Checking agains null doesn't work as intended // if (pc.iceConnectionState == null) return null; + try { final result = await pc.getRemoteDescription(); logger.fine('pc.getRemoteDescription $result'); return result; } catch (_) { - logger.warning('pc.getRemoteDescription did throw: $_'); + logger.warning('pc.getRemoteDescription failed with error: $_'); } } } diff --git a/lib/src/types.dart b/lib/src/types.dart new file mode 100644 index 0000000..ef1e6b9 --- /dev/null +++ b/lib/src/types.dart @@ -0,0 +1,91 @@ +// +// LiveKit +// + +import 'package:flutter/material.dart'; + +import 'extensions.dart'; + +typedef CancelListenFunc = Function(); + +enum Reliability { + reliable, + lossy, +} + +enum RTCIceTransportPolicy { + all, + relay, +} + +@immutable +class RTCOfferOptions { + final bool iceRestart; + + const RTCOfferOptions({ + this.iceRestart = false, + }); + + Map toMap() => { + if (iceRestart) 'iceRestart': true, + }; +} + +@immutable +class RTCConfiguration { + final int? iceCandidatePoolSize; + final List? iceServers; + final RTCIceTransportPolicy? iceTransportPolicy; + + const RTCConfiguration({ + this.iceCandidatePoolSize, + this.iceServers, + this.iceTransportPolicy, + }); + + Map toMap() { + final iceServersMap = >[ + if (iceServers != null) + for (final e in iceServers!) e.toMap() + ]; + + return { + // only supports unified plan + 'sdpSemantics': 'unified-plan', + if (iceServersMap.isNotEmpty) 'iceServers': iceServersMap, + if (iceCandidatePoolSize != null) 'iceCandidatePoolSize': iceCandidatePoolSize, + if (iceTransportPolicy != null) 'iceTransportPolicy': iceTransportPolicy!.toStringValue(), + }; + } + + // Returns new options with updated properties + RTCConfiguration copyWith({ + int? iceCandidatePoolSize, + List? iceServers, + RTCIceTransportPolicy? iceTransportPolicy, + }) => + RTCConfiguration( + iceCandidatePoolSize: iceCandidatePoolSize ?? this.iceCandidatePoolSize, + iceServers: iceServers ?? this.iceServers, + iceTransportPolicy: iceTransportPolicy ?? this.iceTransportPolicy, + ); +} + +@immutable +class RTCIceServer { + final List? urls; + final String? username; + final String? credential; + + const RTCIceServer({ + this.urls, + this.username, + this.credential, + }); + + Map toMap() => { + if (urls?.isNotEmpty ?? false) 'urls': urls, + if (username?.isNotEmpty ?? false) 'username': username, + if (credential?.isNotEmpty ?? false) 'credential': credential, + }; +} diff --git a/lib/src/utils.dart b/lib/src/utils.dart index ee9798d..66539e2 100644 --- a/lib/src/utils.dart +++ b/lib/src/utils.dart @@ -2,12 +2,58 @@ // // -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'dart:async'; + +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'options.dart'; import 'track/options.dart'; +enum ProtocolVersion { + protocol2, + protocol3, +} + +extension ProtocolVersionExt on ProtocolVersion { + String toStringValue() => { + ProtocolVersion.protocol2: '2', + ProtocolVersion.protocol3: '3', + }[this]!; +} + +extension UriExt on Uri { + bool get isSecureScheme => ['https', 'wss'].contains(scheme); +} + +// Collection of state-less static methods class Utils { + static Uri buildUri( + String uriString, { + required String token, + ConnectOptions? options, + bool reconnect = false, + bool validate = false, + bool forceSecure = false, + required ProtocolVersion protocol, + }) { + final Uri uri = Uri.parse(uriString); + + final useSecure = uri.isSecureScheme || forceSecure; + final httpScheme = useSecure ? 'https' : 'http'; + final wsScheme = useSecure ? 'wss' : 'ws'; + + return uri.replace( + scheme: validate ? httpScheme : wsScheme, + path: validate ? 'validate' : 'rtc', + queryParameters: { + 'access_token': token, + if (options != null) 'auto_subscribe': options.autoSubscribe ? '1' : '0', + if (reconnect) 'reconnect': '1', + 'protocol': protocol.toStringValue(), + }, + ); + } + static List _presetsForResolution( int width, int height, @@ -31,7 +77,7 @@ class Utils { return result; } - static List? computeVideoEncodings({ + static List? computeVideoEncodings({ int? width, int? height, TrackPublishOptions? options, @@ -68,7 +114,7 @@ class Utils { ), // if resolution is high enough, we would send both h and q res.. // otherwise only send h - if (height * 0.7 >= midPreset.height) ...[ + if (width >= 960) ...[ midPreset.encoding.toRTCRtpEncoding( rid: 'h', scaleResolutionDownBy: height / midPreset.height, @@ -84,4 +130,22 @@ class Utils { ), ]; } + + // makes a debounce func + static Function createDebounceFunc( + Function f, { + Function(Function)? cancelFunc, + required Duration wait, + }) { + Timer? t; + return () { + t?.cancel(); + t = Timer(wait, () { + t = null; + f(); + }); + // pass back the cancel method so we can cancel it when no longer needed + cancelFunc?.call(t!.cancel); + }; + } } diff --git a/lib/src/widget/video_track_renderer.dart b/lib/src/widget/video_track_renderer.dart index 08717b2..987bcfd 100644 --- a/lib/src/widget/video_track_renderer.dart +++ b/lib/src/widget/video_track_renderer.dart @@ -1,6 +1,6 @@ import 'package:flutter/foundation.dart'; import 'package:flutter/material.dart'; -import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import '../track/local_video_track.dart'; import '../track/video_track.dart'; @@ -8,13 +8,13 @@ import '../track/video_track.dart'; /// Widget that renders a [VideoTrack]. class VideoTrackRenderer extends StatefulWidget { final VideoTrack track; - final RTCVideoRenderer renderer; - final RTCVideoViewObjectFit fit; + final rtc.RTCVideoRenderer renderer; + final rtc.RTCVideoViewObjectFit fit; VideoTrackRenderer( this.track, { - this.fit = RTCVideoViewObjectFit.RTCVideoViewObjectFitContain, - }) : renderer = RTCVideoRenderer(), + this.fit = rtc.RTCVideoViewObjectFit.RTCVideoViewObjectFitContain, + }) : renderer = rtc.RTCVideoRenderer(), super(key: ValueKey(track.sid)); @override @@ -22,7 +22,7 @@ class VideoTrackRenderer extends StatefulWidget { } class _VideoTrackRendererState extends State { - final _renderer = RTCVideoRenderer(); + final _renderer = rtc.RTCVideoRenderer(); @override void initState() { @@ -61,7 +61,7 @@ class _VideoTrackRendererState extends State { @override Widget build(BuildContext context) { final isLocal = widget.track is LocalVideoTrack; - return RTCVideoView( + return rtc.RTCVideoView( _renderer, mirror: isLocal, filterQuality: FilterQuality.medium, diff --git a/lib/src/ws/interface.dart b/lib/src/ws/interface.dart index 0041666..f82edf8 100644 --- a/lib/src/ws/interface.dart +++ b/lib/src/ws/interface.dart @@ -1,41 +1,41 @@ import 'platform/io.dart' if (dart.library.html) 'platform/web.dart'; -class LKWebSocketError implements Exception { +class WebSocketException implements Exception { final int code; - const LKWebSocketError._(this.code); + const WebSocketException._(this.code); - static LKWebSocketError unknown() => const LKWebSocketError._(0); - static LKWebSocketError connect() => const LKWebSocketError._(1); + static WebSocketException unknown() => const WebSocketException._(0); + static WebSocketException connect() => const WebSocketException._(1); @override String toString() => { - LKWebSocketError.unknown(): 'Unknown error', - LKWebSocketError.connect(): 'Failed to connect', + WebSocketException.unknown(): 'Unknown error', + WebSocketException.connect(): 'Failed to connect', }[this]!; } -typedef LKWebSocketOnData = Function(dynamic data); -typedef LKWebSocketOnError = Function(dynamic error); -typedef LKWebSocketOnDispose = Function(); +typedef WebSocketOnData = Function(dynamic data); +typedef WebSocketOnError = Function(dynamic error); +typedef WebSocketOnDispose = Function(); -class LKWebSocketOptions { - final LKWebSocketOnData? onData; - final LKWebSocketOnError? onError; - final LKWebSocketOnDispose? onDispose; - const LKWebSocketOptions({ +class WebSocketOptions { + final WebSocketOnData? onData; + final WebSocketOnError? onError; + final WebSocketOnDispose? onDispose; + const WebSocketOptions({ this.onData, this.onError, this.onDispose, }); } -abstract class LKWebSocket { +abstract class LiveKitWebSocket { void send(List data); void dispose(); - static Future connect( + static Future connect( Uri uri, [ - LKWebSocketOptions? options, + WebSocketOptions? options, ]) => lkWebSocketConnect(uri, options); } diff --git a/lib/src/ws/platform/io.dart b/lib/src/ws/platform/io.dart index c951541..41d614d 100644 --- a/lib/src/ws/platform/io.dart +++ b/lib/src/ws/platform/io.dart @@ -1,22 +1,21 @@ import 'dart:async'; import 'dart:io' as io; -import 'package:livekit_client/src/logger.dart'; - +import '../../logger.dart'; import '../interface.dart'; -Future lkWebSocketConnect( +Future lkWebSocketConnect( Uri uri, [ - LKWebSocketOptions? options, + WebSocketOptions? options, ]) => - LKWebSocketIO.connect(uri, options); + LiveKitWebSocketIO.connect(uri, options); -class LKWebSocketIO implements LKWebSocket { +class LiveKitWebSocketIO implements LiveKitWebSocket { final io.WebSocket _ws; - final LKWebSocketOptions? options; + final WebSocketOptions? options; late final StreamSubscription _subscription; - LKWebSocketIO._( + LiveKitWebSocketIO._( this._ws, [ this.options, ]) { @@ -36,18 +35,18 @@ class LKWebSocketIO implements LKWebSocket { @override void send(List data) => _ws.add(data); - static Future connect( + static Future connect( Uri uri, [ - LKWebSocketOptions? options, + WebSocketOptions? options, ]) async { - logger.fine('LKWebSocketIO connect (uri: ${uri.toString()})'); + logger.fine('WebSocketIO connect (uri: ${uri.toString()})'); try { final ws = await io.WebSocket.connect(uri.toString()); - logger.fine('LKWebSocketIO connected'); - return LKWebSocketIO._(ws, options); + logger.fine('WebSocketIO connected'); + return LiveKitWebSocketIO._(ws, options); } catch (_) { - logger.severe('LKWebSocketIO error ${_}'); - throw LKWebSocketError.connect(); + logger.severe('WebSocketIO error ${_}'); + throw WebSocketException.connect(); } } } diff --git a/lib/src/ws/platform/web.dart b/lib/src/ws/platform/web.dart index d4f7e8b..b1da3d7 100644 --- a/lib/src/ws/platform/web.dart +++ b/lib/src/ws/platform/web.dart @@ -6,19 +6,19 @@ import 'dart:typed_data'; import '../interface.dart'; -Future lkWebSocketConnect( +Future lkWebSocketConnect( Uri uri, [ - LKWebSocketOptions? options, + WebSocketOptions? options, ]) => - LKWebSocketWeb.connect(uri, options); + LiveKitWebSocketWeb.connect(uri, options); -class LKWebSocketWeb implements LKWebSocket { +class LiveKitWebSocketWeb implements LiveKitWebSocket { final html.WebSocket _ws; - final LKWebSocketOptions? options; + final WebSocketOptions? options; late final StreamSubscription _messageSubscription; late final StreamSubscription _closeSubscription; - LKWebSocketWeb._( + LiveKitWebSocketWeb._( this._ws, [ this.options, ]) { @@ -41,14 +41,14 @@ class LKWebSocketWeb implements LKWebSocket { _ws.close(); } - static Future connect( + static Future connect( Uri uri, [ - LKWebSocketOptions? options, + WebSocketOptions? options, ]) async { - final completer = Completer(); + final completer = Completer(); final ws = html.WebSocket(uri.toString()); - ws.onOpen.listen((_) => completer.complete(LKWebSocketWeb._(ws, options))); - ws.onError.listen((_) => completer.completeError(LKWebSocketError.connect())); + ws.onOpen.listen((_) => completer.complete(LiveKitWebSocketWeb._(ws, options))); + ws.onError.listen((_) => completer.completeError(WebSocketException.connect())); return completer.future; } } diff --git a/pubspec.lock b/pubspec.lock index 9f20fb9..d60c3a1 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -37,7 +37,7 @@ packages: source: hosted version: "1.1.0" collection: - dependency: "direct main" + dependency: transitive description: name: collection url: "https://pub.dartlang.org" @@ -72,7 +72,7 @@ packages: source: hosted version: "6.1.2" fixnum: - dependency: "direct main" + dependency: transitive description: name: fixnum url: "https://pub.dartlang.org" @@ -129,7 +129,7 @@ packages: name: logging url: "https://pub.dartlang.org" source: hosted - version: "1.0.1" + version: "1.0.2" matcher: dependency: transitive description: @@ -157,14 +157,14 @@ packages: name: path_provider url: "https://pub.dartlang.org" source: hosted - version: "2.0.3" + version: "2.0.4" 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: @@ -221,13 +221,6 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "2.0.0" - quiver: - dependency: transitive - description: - name: quiver - url: "https://pub.dartlang.org" - source: hosted - version: "3.0.1" sky_engine: dependency: transitive description: flutter @@ -282,13 +275,6 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "0.4.2" - tuple: - dependency: "direct main" - description: - name: tuple - url: "https://pub.dartlang.org" - source: hosted - version: "2.0.0" typed_data: dependency: transitive description: diff --git a/pubspec.yaml b/pubspec.yaml index f8200dd..a31feb4 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -10,72 +10,24 @@ environment: dependencies: flutter: sdk: flutter - - collection: ^1.15.0 - fixnum: ^1.0.0 - flutter_webrtc: ^0.6.7 - http: ^0.13.3 - logging: ^1.0.1 - protobuf: ^2.0.0 - - tuple: ^2.0.0 - + logging: ^1.0.2 uuid: ^3.0.4 synchronized: ^3.0.0 - - # + protobuf: ^2.0.0 + # protobuf: # git: # url: https://github.com/google/protobuf.dart.git # ref: master # path: protobuf/ - # # WebSocketChannel has design flaws # https://github.com/dart-lang/web_socket_channel/issues/25 - # # web_socket_channel: ^2.1.0 - + dev_dependencies: flutter_test: sdk: flutter flutter_lints: ^1.0.4 - -# For information on the generic Dart part of this file, see the -# following page: https://dart.dev/tools/pub/pubspec - -# The following section is specific to Flutter. -flutter: - - # To add assets to your package, add an assets section, like this: - # assets: - # - images/a_dot_burr.jpeg - # - images/a_dot_ham.jpeg - # - # For details regarding assets in packages, see - # https://flutter.dev/assets-and-images/#from-packages - # - # An image asset can refer to one or more resolution-specific "variants", see - # https://flutter.dev/assets-and-images/#resolution-aware. - - # To add custom fonts to your package, add a fonts section here, - # in this "flutter" section. Each entry in this list should have a - # "family" key with the font family name, and a "fonts" key with a - # list giving the asset and other descriptors for the font. For - # example: - # fonts: - # - family: Schyler - # fonts: - # - asset: fonts/Schyler-Regular.ttf - # - asset: fonts/Schyler-Italic.ttf - # style: italic - # - family: Trajan Pro - # fonts: - # - asset: fonts/TrajanPro.ttf - # - asset: fonts/TrajanPro_Bold.ttf - # weight: 700 - # - # For details regarding fonts in packages, see - # https://flutter.dev/custom-fonts/#from-packages