implemented tracks, local & remote participants

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<MediaStream> getMediaStream() async {
+    var stream = _mediaStream;
+    if (stream == null) {
+      stream = await createLocalMediaStream(sid);
+      _mediaStream = stream;
+    }
+    return stream;
+  }
+
+  Future<TrackPublication> 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<TrackPublication> 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<int> data, DataPacket_Kind reliability,
+      {List<String>? 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);
+  }
 }

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<int> 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
     }
   }
 }

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<String, RemoteTrackPublication>();
+    var newPubs = Map<String, RemoteTrackPublication>();
+
+    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<RemoteTrackPublication?> _waitForTrackPublication(
+      String sid, Duration delay) async {
+    var endTime = DateTime.now().add(delay);
+    while (DateTime.now().isBefore(endTime)) {
+      var pub = await Future<RemoteTrackPublication?>.delayed(
+          Duration(milliseconds: 100), () {
+        return getTrackPublication(sid);
+      });
+      if (pub != null) {
+        return pub;
+      }
+    }
+  }
+}