From b18c37fd1ab6dee4a6c50c01851f3588dfe0cab3 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Wed, 21 Jul 2021 23:30:24 -0700 Subject: [PATCH] implemented tracks, local & remote participants --- lib/src/errors.dart | 4 + lib/src/participant/local_participant.dart | 137 ++++++++++++++++++++ lib/src/participant/participant.dart | 38 ++++-- lib/src/participant/remote_participant.dart | 135 ++++++++++++++++++- lib/src/rtc_engine.dart | 1 + lib/src/track/local_audio_track.dart | 8 ++ lib/src/track/local_track_publication.dart | 13 ++ lib/src/track/local_video_track.dart | 8 ++ lib/src/track/remote_track_publication.dart | 66 ++++++++++ lib/src/track/track.dart | 41 ++++++ lib/src/track/track_publication.dart | 8 +- lib/src/track/video_track.dart | 17 +++ pubspec.lock | 7 + pubspec.yaml | 1 + 14 files changed, 468 insertions(+), 16 deletions(-) create mode 100644 lib/src/track/local_audio_track.dart create mode 100644 lib/src/track/local_track_publication.dart create mode 100644 lib/src/track/local_video_track.dart create mode 100644 lib/src/track/remote_track_publication.dart create mode 100644 lib/src/track/video_track.dart diff --git a/lib/src/errors.dart b/lib/src/errors.dart index 6cb762f..91867f9 100644 --- a/lib/src/errors.dart +++ b/lib/src/errors.dart @@ -16,3 +16,7 @@ class ConnectError extends LiveKitError { class TrackPublishError extends LiveKitError { TrackPublishError([String msg = 'Failed to publish track']) : super(msg); } + +class DataPublishError extends LiveKitError { + DataPublishError([String msg = 'Failed to publish data']) : super(msg); +} diff --git a/lib/src/participant/local_participant.dart b/lib/src/participant/local_participant.dart index 2d54945..f77bcce 100644 --- a/lib/src/participant/local_participant.dart +++ b/lib/src/participant/local_participant.dart @@ -1,5 +1,142 @@ +import 'package:flutter_webrtc/flutter_webrtc.dart'; + +import '../errors.dart'; +import '../proto/livekit_models.pb.dart'; +import '../proto/livekit_rtc.pbserver.dart'; +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 'participant.dart'; class LocalParticipant extends Participant { + RTCEngine _engine; + MediaStream? _mediaStream; + LocalParticipant({ + required RTCEngine engine, + required ParticipantInfo info, + }) : _engine = engine, + super(info.sid, info.identity) { + updateFromInfo(info); + } + + RTCEngine get engine => _engine; + + Future getMediaStream() async { + var stream = _mediaStream; + if (stream == null) { + stream = await createLocalMediaStream(sid); + _mediaStream = stream; + } + return stream; + } + + Future publishAudioTrack(LocalAudioTrack track) async { + if (audioTracks.values.any( + (element) => element.track?.mediaTrack.id == track.mediaTrack.id)) { + throw new TrackPublishError('track already exists'); + } + + var trackInfo = await _engine.addTrack( + cid: track.getCid(), name: track.name, kind: track.kind); + var stream = await getMediaStream(); + var transceiverInit = new RTCRtpTransceiverInit( + direction: TransceiverDirection.SendOnly, + streams: [stream], + ); + track.transceiver = await _engine.publisher?.pc.addTransceiver( + track: track.mediaTrack, + kind: track.mediaType, + init: transceiverInit, + ); + + var pub = new LocalTrackPublication(trackInfo, track, this); + addTrackPublication(pub); + + return pub; + } + + Future publishVideoTrack(LocalVideoTrack track) async { + if (audioTracks.values.any( + (element) => element.track?.mediaTrack.id == track.mediaTrack.id)) { + throw new TrackPublishError('track already exists'); + } + + var trackInfo = await _engine.addTrack( + cid: track.getCid(), name: track.name, kind: track.kind); + var stream = await getMediaStream(); + var transceiverInit = new RTCRtpTransceiverInit( + direction: TransceiverDirection.SendOnly, + streams: [stream], + ); + // TODO: video encodings and simulcast + track.transceiver = await _engine.publisher?.pc.addTransceiver( + track: track.mediaTrack, + kind: track.mediaType, + init: transceiverInit, + ); + + var pub = new LocalTrackPublication(trackInfo, track, this); + addTrackPublication(pub); + + return pub; + } + + unpublishTrack(Track track) { + var pub = tracks.values.firstWhere((element) => element.track == track); + if (pub == null) { + return; + } + + track.stop(); + var sender = track.transceiver?.sender; + if (sender != null) { + engine.publisher?.pc.removeTrack(sender); + } + + tracks.remove(pub.sid); + switch (pub.kind) { + case TrackType.AUDIO: + audioTracks.remove(pub.sid); + break; + case TrackType.VIDEO: + videoTracks.remove(pub.sid); + break; + } + } + + publishData(List data, DataPacket_Kind reliability, + {List? destinationSids}) { + RTCDataChannel? channel; + switch (reliability) { + case DataPacket_Kind.RELIABLE: + channel = engine.reliableDC; + break; + case DataPacket_Kind.LOSSY: + channel = engine.lossyDC; + break; + } + if (channel == null) { + return; + } + + var packet = new DataPacket( + kind: reliability, + user: new UserPacket( + payload: data, + participantSid: sid, + destinationSids: destinationSids, + ), + ); + + var buffer = packet.writeToBuffer(); + channel.send(RTCDataChannelMessage.fromBinary(buffer)); + } + + updateFromInfo(ParticipantInfo info) { + super.updateFromInfo(info); + } } diff --git a/lib/src/participant/participant.dart b/lib/src/participant/participant.dart index e8b1209..3ea0e5b 100644 --- a/lib/src/participant/participant.dart +++ b/lib/src/participant/participant.dart @@ -1,9 +1,25 @@ +import 'remote_participant.dart'; import '../proto/livekit_models.pb.dart'; +import '../track/remote_track_publication.dart'; +import '../track/track.dart'; import '../track/track_publication.dart'; mixin ParticipantDelegate { - void onMetadataChanged(Participant participant); - void onSpeakingChanged(Participant participant, bool speaking); + void onMetadataChanged(Participant participant) {} + void onSpeakingChanged(Participant participant, bool speaking) {} + void onTrackMuted(Participant participant, TrackPublication publication) {} + void onTrackUnmuted(Participant participant, TrackPublication publication) {} + void onTrackPublished( + RemoteParticipant participant, RemoteTrackPublication publication) {} + void onTrackUnpublished( + RemoteParticipant participant, RemoteTrackPublication publication) {} + void onTrackSubscribed(RemoteParticipant participant, Track track, + RemoteTrackPublication publication) {} + void onTrackUnsubscribed(RemoteParticipant participant, Track track, + RemoteTrackPublication publication) {} + void onDataReceived(RemoteParticipant participant, List data) {} + void onTrackSubscriptionFailed( + RemoteParticipant participant, String sid, String? message) {} } class Participant { @@ -28,9 +44,7 @@ class Participant { /// when the participant had last spoken DateTime? lastSpokeAt; - Participant(this.sid, this.identity); - - ParticipantDelegate? _roomDelegate; + ParticipantDelegate? roomDelegate; ParticipantDelegate? delegate; ParticipantInfo? _participantInfo; @@ -48,6 +62,10 @@ class Participant { /// if participant is currently speaking bool get isSpeaking => _isSpeaking; + bool get hasInfo => _participantInfo != null; + + Participant(this.sid, this.identity); + set isSpeaking(bool speaking) { if (_isSpeaking != speaking) { return; @@ -57,7 +75,7 @@ class Participant { lastSpokeAt = DateTime.now(); } delegate?.onSpeakingChanged(this, speaking); - _roomDelegate?.onSpeakingChanged(this, speaking); + roomDelegate?.onSpeakingChanged(this, speaking); } _setMetadata(String md) { @@ -65,11 +83,11 @@ class Participant { this.metadata = md; if (changed) { delegate?.onMetadataChanged(this); - _roomDelegate?.onMetadataChanged(this); + roomDelegate?.onMetadataChanged(this); } } - _updateInfo(ParticipantInfo info) { + updateFromInfo(ParticipantInfo info) { this.identity = info.identity; this.sid = info.sid; if (info.metadata.isNotEmpty) { @@ -78,7 +96,7 @@ class Participant { this._participantInfo = info; } - _addTrackPublication(TrackPublication pub) { + addTrackPublication(TrackPublication pub) { pub.track?.sid = pub.sid; tracks[pub.sid] = pub; switch (pub.kind) { @@ -89,7 +107,7 @@ class Participant { videoTracks[pub.sid] = pub; break; default: - // nothing + // nothing } } } diff --git a/lib/src/participant/remote_participant.dart b/lib/src/participant/remote_participant.dart index eaa60dc..0341322 100644 --- a/lib/src/participant/remote_participant.dart +++ b/lib/src/participant/remote_participant.dart @@ -1,3 +1,136 @@ +import 'package:flutter_webrtc/flutter_webrtc.dart'; +import '../proto/livekit_models.pb.dart'; +import '../signal_client.dart'; +import '../track/remote_track_publication.dart'; +import '../track/track.dart'; +import '../track/video_track.dart'; import 'participant.dart'; -class RemoteParticipant extends Participant {} +class RemoteParticipant extends Participant { + SignalClient _client; + + SignalClient get client => _client; + + RemoteParticipant(this._client, String sid, String identity) + : super(sid, identity); + + RemoteParticipant.fromInfo(this._client, ParticipantInfo info) + : super(info.sid, info.identity) { + updateFromInfo(info); + } + + RemoteTrackPublication? getTrackPublication(String sid) { + var pub = tracks[sid]; + if (pub is RemoteTrackPublication) { + return pub; + } + } + + addSubscribedMediaTrack(MediaStreamTrack mediaTrack, String sid) async { + var pub = getTrackPublication(sid); + if (pub == null) { + // we may have received the track prior to metadata. wait up to 3s + pub = await _waitForTrackPublication(sid, Duration(seconds: 3)); + if (pub == null) { + var msg = 'no track metadata found'; + delegate?.onTrackSubscriptionFailed(this, sid, msg); + roomDelegate?.onTrackSubscriptionFailed(this, sid, msg); + return; + } + } + + Track? track; + if (pub.kind == TrackType.AUDIO) { + track = new Track(pub.kind, pub.name, mediaTrack); + } else if (pub.kind == TrackType.VIDEO) { + track = new VideoTrack(pub.name, mediaTrack); + } else { + var msg = 'unsupported track type ${pub.kind}'; + delegate?.onTrackSubscriptionFailed(this, sid, msg); + roomDelegate?.onTrackSubscriptionFailed(this, sid, msg); + return; + } + + pub.track = track; + addTrackPublication(pub); + + delegate?.onTrackSubscribed(this, track, pub); + roomDelegate?.onTrackSubscribed(this, track, pub); + } + + @override + void updateFromInfo(ParticipantInfo info) { + var hadInfo = hasInfo; + super.updateFromInfo(info); + + // figuring out deltas between tracks + var validPubs = Map(); + var newPubs = Map(); + + for (var info in info.tracks) { + var sid = info.sid; + var pub = getTrackPublication(sid); + + if (pub == null) { + pub = RemoteTrackPublication(info, this); + newPubs[sid] = pub; + addTrackPublication(pub); + } else { + pub.updateFromInfo(info); + } + + validPubs[sid] = pub; + } + + // notify listeners when it's not a new participant + if (hadInfo) { + for (var pub in newPubs.values) { + delegate?.onTrackPublished(this, pub); + roomDelegate?.onTrackPublished(this, pub); + } + return; + } + + // remove tracks + for (var pub in tracks.values) { + if (!validPubs.containsKey(pub.sid)) { + unpublishTrack(sid, true); + } + } + } + + unpublishTrack(String sid, [bool sendUnpublish = false]) { + var pub = tracks.remove(sid); + if (pub == null || !(pub is RemoteTrackPublication)) { + return; + } + + audioTracks.remove(sid); + videoTracks.remove(sid); + + var track = pub.track; + if (track != null) { + track.stop(); + delegate?.onTrackUnsubscribed(this, track, pub); + roomDelegate?.onTrackUnsubscribed(this, track, pub); + } + if (sendUnpublish) { + delegate?.onTrackUnpublished(this, pub); + roomDelegate?.onTrackUnpublished(this, pub); + } + } + + Future _waitForTrackPublication( + String sid, Duration delay) async { + var endTime = DateTime.now().add(delay); + while (DateTime.now().isBefore(endTime)) { + var pub = await Future.delayed( + Duration(milliseconds: 100), () { + return getTrackPublication(sid); + }); + if (pub != null) { + return pub; + } + } + } +} diff --git a/lib/src/rtc_engine.dart b/lib/src/rtc_engine.dart index 9e3e456..ee38db3 100644 --- a/lib/src/rtc_engine.dart +++ b/lib/src/rtc_engine.dart @@ -289,6 +289,7 @@ class RTCEngine with SignalClientDelegate { var completer = pendingTrackResolvers[response.cid]; if (completer != null) { completer.complete(Future.value(response.track)); + pendingTrackResolvers.remove(response.cid); } } diff --git a/lib/src/track/local_audio_track.dart b/lib/src/track/local_audio_track.dart new file mode 100644 index 0000000..a67c829 --- /dev/null +++ b/lib/src/track/local_audio_track.dart @@ -0,0 +1,8 @@ +import 'package:flutter_webrtc/flutter_webrtc.dart'; + +import '../proto/livekit_models.pb.dart'; +import 'track.dart'; + +class LocalAudioTrack extends Track { + LocalAudioTrack(String name, MediaStreamTrack track) : super(TrackType.AUDIO, name, track); +} diff --git a/lib/src/track/local_track_publication.dart b/lib/src/track/local_track_publication.dart new file mode 100644 index 0000000..a47d2c1 --- /dev/null +++ b/lib/src/track/local_track_publication.dart @@ -0,0 +1,13 @@ +import '../participant/local_participant.dart'; +import '../proto/livekit_models.pb.dart'; +import 'track.dart'; +import 'track_publication.dart'; + +class LocalTrackPublication extends TrackPublication { + LocalParticipant _participant; + + LocalTrackPublication(TrackInfo info, Track track, this._participant) + : super.fromInfo(info) { + this.track = track; + } +} diff --git a/lib/src/track/local_video_track.dart b/lib/src/track/local_video_track.dart new file mode 100644 index 0000000..b01b467 --- /dev/null +++ b/lib/src/track/local_video_track.dart @@ -0,0 +1,8 @@ +import 'package:flutter_webrtc/flutter_webrtc.dart'; + +import 'video_track.dart'; + +class LocalVideoTrack extends VideoTrack { + LocalVideoTrack(String name, MediaStreamTrack mediaTrack) + : super(name, mediaTrack); +} diff --git a/lib/src/track/remote_track_publication.dart b/lib/src/track/remote_track_publication.dart new file mode 100644 index 0000000..a640de0 --- /dev/null +++ b/lib/src/track/remote_track_publication.dart @@ -0,0 +1,66 @@ +import '../proto/livekit_models.pb.dart'; +import '../proto/livekit_rtc.pbserver.dart'; +import '../participant/remote_participant.dart'; +import 'track.dart'; +import 'track_publication.dart'; + +class RemoteTrackPublication extends TrackPublication { + RemoteParticipant _participant; + bool _unsubscribed = false; + bool _disabled = false; + VideoQuality _videoQuality = VideoQuality.HIGH; + + VideoQuality get videoQuality => _videoQuality; + set videoQuality(VideoQuality val) { + _videoQuality = val; + _sendUpdateTrackSettings(); + } + + bool get enabled => !_disabled; + set enabled(bool val) { + _disabled = !val; + _sendUpdateTrackSettings(); + } + + bool get subscribed { + if (_unsubscribed) { + return false; + } + return super.subscribed; + } + + set subscribed(bool val) { + _unsubscribed = !val; + _sendUpdateTrackSettings(); + } + + set muted(bool val) { + if (val == muted) { + return; + } + super.muted = val; + if (val) { + _participant.delegate?.onTrackMuted(_participant, this); + _participant.roomDelegate?.onTrackMuted(_participant, this); + } else { + _participant.delegate?.onTrackUnmuted(_participant, this); + _participant.roomDelegate?.onTrackUnmuted(_participant, this); + } + } + + RemoteTrackPublication(TrackInfo info, this._participant, [Track? track]) + : super.fromInfo(info) { + this.track = track; + } + + _sendUpdateTrackSettings() { + var settings = new UpdateTrackSettings( + trackSids: [sid], + disabled: _disabled, + ); + if (kind == TrackType.VIDEO) { + settings.quality = _videoQuality; + } + _participant.client.sendUpdateTrackSettings(settings); + } +} diff --git a/lib/src/track/track.dart b/lib/src/track/track.dart index 7357de4..3d13f5b 100644 --- a/lib/src/track/track.dart +++ b/lib/src/track/track.dart @@ -1,3 +1,8 @@ +import 'package:flutter_webrtc/flutter_webrtc.dart'; +import 'package:uuid/uuid.dart'; + +import '../proto/livekit_models.pb.dart'; + class TrackDimension { int width; int height; @@ -6,5 +11,41 @@ class TrackDimension { } class Track { + String name; + TrackType kind; + MediaStreamTrack mediaTrack; String? sid; + RTCRtpTransceiver? transceiver; + String? _cid; + + Track(this.kind, this.name, this.mediaTrack); + + RTCRtpMediaType get mediaType { + switch (kind) { + case TrackType.AUDIO: + return RTCRtpMediaType.RTCRtpMediaTypeAudio; + case TrackType.VIDEO: + return RTCRtpMediaType.RTCRtpMediaTypeVideo; + // this should never happen + default: + return RTCRtpMediaType.RTCRtpMediaTypeAudio; + } + } + + String getCid() { + var cid = _cid; + if (cid == null) { + cid = mediaTrack.id; + } + if (cid == null) { + var uuid = Uuid(); + cid = uuid.v4(); + _cid = cid; + } + return cid; + } + + stop() { + mediaTrack.stop(); + } } diff --git a/lib/src/track/track_publication.dart b/lib/src/track/track_publication.dart index a498cb0..2c1b289 100644 --- a/lib/src/track/track_publication.dart +++ b/lib/src/track/track_publication.dart @@ -10,18 +10,16 @@ class TrackPublication { bool simulcasted = false; TrackDimension? dimension; - bool get isSubscribed => track != null; - - TrackPublication({required this.sid, required this.name, required this.kind}); + bool get subscribed => track != null; TrackPublication.fromInfo(TrackInfo info) : sid = info.sid, name = info.name, kind = info.type { - _updateFromInfo(info); + updateFromInfo(info); } - _updateFromInfo(TrackInfo info) { + updateFromInfo(TrackInfo info) { muted = info.muted; simulcasted = info.simulcast; if (info.type == TrackType.VIDEO) { diff --git a/lib/src/track/video_track.dart b/lib/src/track/video_track.dart new file mode 100644 index 0000000..8bd13c4 --- /dev/null +++ b/lib/src/track/video_track.dart @@ -0,0 +1,17 @@ +import 'package:flutter_webrtc/flutter_webrtc.dart'; + +import '../proto/livekit_models.pb.dart'; +import 'track.dart'; + +class VideoTrack extends Track { + VideoTrack(String name, MediaStreamTrack mediaTrack) + : super(TrackType.VIDEO, name, mediaTrack); + + // TODO: keep list of video renderers + + @override + stop() { + super.stop(); + // TODO: remove renderer + } +} diff --git a/pubspec.lock b/pubspec.lock index 481fbd5..807fd1f 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -240,6 +240,13 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "1.3.0" + uuid: + dependency: "direct main" + description: + name: uuid + url: "https://pub.dartlang.org" + source: hosted + version: "3.0.4" vector_math: dependency: transitive description: diff --git a/pubspec.yaml b/pubspec.yaml index ef69574..b67fc47 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -14,6 +14,7 @@ dependencies: logging: ^1.0.1 protobuf: ^2.0.0 web_socket_channel: ^2.1.0 + uuid: ^3.0.4 dev_dependencies: flutter_test: