RTCEngine and SignalClient

lib/src/errors.dart

@@ -0,0 +1,18 @@
+class LiveKitError extends Error {
+  String message;
+
+  LiveKitError(this.message);
+
+  @override
+  String toString() {
+    return message;
+  }
+}
+
+class ConnectError extends LiveKitError {
+  ConnectError([String msg = 'Failed to connect to server']) : super(msg);
+}
+
+class TrackPublishError extends LiveKitError {
+  TrackPublishError([String msg = 'Failed to publish track']) : super(msg);
+}

lib/src/rtc_engine.dart

@@ -0,0 +1,262 @@
+import 'dart:async';
+import 'package:flutter_webrtc/flutter_webrtc.dart';
+import 'package:livekit_client_flutter/src/track/track.dart';
+import './errors.dart';
+import './proto/livekit_rtc.pbserver.dart';
+import './proto/livekit_models.pb.dart';
+import './signal_client.dart';
+import './transport.dart';
+
+const lossyDataChannel = '_lossy';
+const reliableDataChannel = '_reliable';
+final connectionTimeout = new Duration(seconds: 5);
+
+typedef GenericCallback = void Function();
+typedef TrackCallback = void Function(
+    MediaStreamTrack track, MediaStream? stream, RTCRtpReceiver? receiver);
+typedef ParticipantUpdateCallback = void Function(
+    List<ParticipantInfo> participants);
+typedef ActiveSpeakerChangedCallback = void Function(
+    List<SpeakerInfo> speakers);
+
+class RTCEngine with SignalClientDelegate {
+  PCTransport? publisher;
+  PCTransport? subscriber;
+  SignalClient client;
+  // config for RTCPeerConnection
+  Map<String, dynamic> rtcConfig = {};
+  // data channels for packets
+  RTCDataChannel? reliableDC;
+  RTCDataChannel? lossyDC;
+  bool iceConnected = false;
+  bool isClosed = true;
+  Map<String, Completer<TrackInfo>> pendingTrackResolvers = {};
+  // to complete join request
+  Completer<JoinResponse>? joinCompleter;
+  // remember url and token for reconnect
+  String? url;
+  String? token;
+
+  // delegate methods
+  GenericCallback? onICEConnected;
+  TrackCallback? onTrack;
+  ParticipantUpdateCallback? onParticipantUpdateCallback;
+  ActiveSpeakerChangedCallback? onActiveSpeakerchangedCallback;
+  GenericCallback? onDisconnected;
+
+  RTCEngine(this.client, Map<String, dynamic>? rtcConfig) {
+    if (rtcConfig != null) {
+      this.rtcConfig = rtcConfig;
+    }
+
+    this.client.delegate = this;
+  }
+
+  Future<JoinResponse> join(String url, String token, JoinOptions? opts) {
+    this.url = url;
+    this.token = token;
+
+    var completer = new Completer<JoinResponse>();
+    joinCompleter = completer;
+
+    // if it's not complete after 5 seconds, fail
+    new Timer(connectionTimeout, () {
+      joinCompleter?.completeError(new ConnectError());
+      joinCompleter = null;
+    });
+
+    return completer.future;
+  }
+
+  close() async {
+    isClosed = true;
+
+    if (publisher != null) {
+      var senders = await publisher?.pc.getSenders();
+      senders?.forEach((element) async {
+        await publisher?.pc.removeTrack(element);
+      });
+
+      publisher?.pc.close();
+      publisher = null;
+    }
+    if (subscriber != null) {
+      subscriber?.pc.close();
+      subscriber = null;
+    }
+    client.close();
+  }
+
+  Future<TrackInfo> addTrack(
+      {required String cid,
+      required String name,
+      required TrackType kind,
+      TrackDimension? dimension}) async {
+    if (pendingTrackResolvers[cid] != null) {
+      throw new TrackPublishError(
+          'a track with the same CID has already been published');
+    }
+
+    var completer = new Completer<TrackInfo>();
+    pendingTrackResolvers[cid] = completer;
+
+    client.sendAddTrack(cid: cid, name: name, type: kind, dimension: dimension);
+
+    return completer.future;
+  }
+
+  negotiate() async {}
+
+  _configurePeerConnections() async {
+    if (publisher != null) {
+      return;
+    }
+
+    var pubPC = await createPeerConnection(rtcConfig);
+    publisher = new PCTransport(pubPC);
+    var subPC = await createPeerConnection(rtcConfig);
+    subscriber = new PCTransport(subPC);
+
+    pubPC.onIceCandidate = (RTCIceCandidate candidate) {
+      client.sendIceCandidate(candidate, SignalTarget.PUBLISHER);
+    };
+    subPC.onIceCandidate = (RTCIceCandidate candidate) {
+      client.sendIceCandidate(candidate, SignalTarget.SUBSCRIBER);
+    };
+
+    pubPC.onRenegotiationNeeded = () {
+      if (pubPC.iceConnectionState ==
+          RTCIceConnectionState.RTCIceConnectionStateNew) {
+        return;
+      }
+      negotiate();
+    };
+
+    pubPC.onIceConnectionState = (RTCIceConnectionState state) {
+      if (publisher == null) {
+        return;
+      }
+      switch (state) {
+        case RTCIceConnectionState.RTCIceConnectionStateConnected:
+          if (!iceConnected) {
+            iceConnected = true;
+            onICEConnected?.call();
+          }
+          break;
+
+        case RTCIceConnectionState.RTCIceConnectionStateFailed:
+          // trigger reconnect sequence
+          _handleDisconnect('peerconnection');
+          break;
+
+        default:
+        // do nothing
+      }
+    };
+
+    subPC.onTrack = (RTCTrackEvent event) {
+      onTrack?.call(event.track, event.streams.first, event.receiver);
+    };
+
+    // create data channels
+    var lossyInit = new RTCDataChannelInit();
+    lossyInit.ordered = true;
+    lossyInit.maxRetransmits = 1;
+    lossyDC = await pubPC.createDataChannel(lossyDataChannel, lossyInit);
+
+    var reliableInit = new RTCDataChannelInit();
+    reliableInit.ordered = true;
+    reliableDC =
+        await pubPC.createDataChannel(reliableDataChannel, reliableInit);
+
+    lossyDC?.onMessage = _handleDataMessage;
+    reliableDC?.onMessage = _handleDataMessage;
+  }
+
+  _handleDataMessage(RTCDataChannelMessage message) {}
+
+  _handleDisconnect(String reason) {
+    // TODO: implement method
+  }
+
+  //------------------ SignalClient Delegate methods -------------------------//
+
+  void onConnected(JoinResponse response) {
+    // create peer connections
+    this.isClosed = false;
+
+    if (rtcConfig['iceServers'] == null && response.iceServers.length > 0) {
+      var iceServers = [];
+      response.iceServers.forEach((item) {
+        Map<String, dynamic> iceServer = {
+          '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;
+    }
+
+    _configurePeerConnections();
+
+    negotiate();
+
+    joinCompleter?.complete(Future.value(response));
+    joinCompleter = null;
+  }
+
+  void onClose(String? reason) {}
+
+  void onOffer(RTCSessionDescription sd) async {
+    var sub = subscriber;
+    if (sub == null) {
+      return;
+    }
+    await sub.setRemoteDescription(sd);
+
+    var answer = await sub.pc.createAnswer();
+    await sub.pc.setLocalDescription(answer);
+    client.sendAnswer(answer);
+  }
+
+  void onAnswer(RTCSessionDescription sd) {
+    if (publisher == null) {
+      return;
+    }
+
+    publisher?.setRemoteDescription(sd);
+  }
+
+  void onTrickle(RTCIceCandidate candidate, SignalTarget target) {
+    if (target == SignalTarget.SUBSCRIBER) {
+      subscriber?.addIceCandidate(candidate);
+    } else if (target == SignalTarget.PUBLISHER) {
+      publisher?.addIceCandidate(candidate);
+    }
+  }
+
+  void onParticipantUpdate(List<ParticipantInfo> updates) {
+    onParticipantUpdateCallback?.call(updates);
+  }
+
+  void onLocalTrackPublished(TrackPublishedResponse response) {
+    var completer = pendingTrackResolvers[response.cid];
+    if (completer != null) {
+      completer.complete(Future.value(response.track));
+    }
+  }
+
+  void onActiveSpeakersChanged(List<SpeakerInfo> speakers) {
+    onActiveSpeakerchangedCallback?.call(speakers);
+  }
+
+  void onLeave(LeaveRequest req) {
+    close();
+    onDisconnected?.call();
+  }
+}

lib/src/signal_client.dart

@@ -0,0 +1,225 @@
+import 'dart:convert';
+import 'dart:developer';
+import 'dart:typed_data';
+
+import 'package:flutter_webrtc/flutter_webrtc.dart';
+import 'package:web_socket_channel/web_socket_channel.dart';
+import './track/track.dart';
+import './version.dart';
+import './proto/livekit_models.pb.dart';
+import './proto/livekit_rtc.pb.dart';
+
+class JoinOptions {
+  final bool? autoSubscribe;
+
+  const JoinOptions({this.autoSubscribe});
+}
+
+mixin SignalClientDelegate {
+  // initial connection established
+  void onConnected(JoinResponse response);
+  // websocket has closed
+  void onClose(String? reason);
+  // when a server offer is received
+  void onOffer(RTCSessionDescription sd);
+  // when an answer from server is received
+  void onAnswer(RTCSessionDescription sd);
+  // when server has a new ICE candidate
+  void onTrickle(RTCIceCandidate candidate, SignalTarget target);
+  // participant has changed
+  void onParticipantUpdate(List<ParticipantInfo> updates);
+  // when a track has been added successfully
+  void onLocalTrackPublished(TrackPublishedResponse response);
+  // active speaker has changed
+  void onActiveSpeakersChanged(List<SpeakerInfo> speakers);
+  // when server sends this client a leave message
+  void onLeave(LeaveRequest req);
+}
+
+class SignalClient {
+  SignalClientDelegate delegate;
+
+  bool _connected = false;
+  WebSocketChannel? _ws;
+
+  SignalClient(this.delegate);
+
+  bool get connected => this._connected;
+
+  join(String url, String token, JoinOptions options) {
+    url += '/rtc';
+    var params = _paramsForToken(token);
+    if (options.autoSubscribe != null) {
+      params += '&auto_subscribe=${options.autoSubscribe! ? '1' : '0'}';
+    }
+    var uri = Uri.parse(url + params);
+
+    try {
+      var ws = WebSocketChannel.connect(uri);
+      ws.stream
+          .listen(_handleMessage, onError: _handleError, onDone: _handleDone);
+      _ws = ws;
+    } catch (e) {
+      // failed before error handler is installed, fail immediately
+    }
+  }
+
+  Future<void> reconnect(String url, String token) async {}
+
+  close() {
+    this._connected = false;
+    this._ws?.sink.close();
+  }
+
+  sendOffer(RTCSessionDescription offer) {
+    this._sendRequest(new SignalRequest(
+      offer: fromRTCSessionDescription(offer),
+    ));
+  }
+
+  sendAnswer(RTCSessionDescription answer) {
+    this._sendRequest(new SignalRequest(
+      answer: fromRTCSessionDescription(answer),
+    ));
+  }
+
+  sendIceCandidate(RTCIceCandidate candidate, SignalTarget target) {
+    this._sendRequest(new SignalRequest(
+        trickle: new TrickleRequest(
+      candidateInit: fromRTCIceCandidate(candidate),
+      target: target,
+    )));
+  }
+
+  sendMuteTrack(String trackSid, bool muted) {
+    this._sendRequest(new SignalRequest(
+      mute: new MuteTrackRequest(
+        sid: trackSid,
+        muted: muted,
+      ),
+    ));
+  }
+
+  sendAddTrack(
+      {required String cid,
+      required String name,
+      required TrackType type,
+      TrackDimension? dimension}) {
+    var req = new AddTrackRequest(
+      cid: cid,
+      name: name,
+      type: type,
+    );
+    if (dimension != null) {
+      req.width = dimension.width;
+      req.height = dimension.height;
+    }
+    this._sendRequest(new SignalRequest(
+      addTrack: req,
+    ));
+  }
+
+  sendUpdateTrackSettings(UpdateTrackSettings settings) {
+    this._sendRequest(new SignalRequest(
+      trackSetting: settings,
+    ));
+  }
+
+  sendUpdateSubscription(UpdateSubscription subscription) {
+    this._sendRequest(new SignalRequest(
+      subscription: subscription,
+    ));
+  }
+
+  sendSetSimulcastLayers(String trackSid, List<VideoQuality> layers) {
+    this._sendRequest(new SignalRequest(
+        simulcast: new SetSimulcastLayers(
+      trackSid: trackSid,
+      layers: layers,
+    )));
+  }
+
+  sendLeave() {
+    this._sendRequest(new SignalRequest(
+      leave: new LeaveRequest(),
+    ));
+  }
+
+  _sendRequest(SignalRequest req) {
+    if (this._ws == null) {
+      log('could not send message, not connected: ' + jsonEncode(req));
+      return;
+    }
+
+    var buf = req.writeToBuffer();
+    this._ws?.sink.add(buf);
+  }
+
+  _handleMessage(dynamic message) {
+    if (!(message is List<int>)) {
+      return;
+    }
+    var msg = SignalResponse.fromBuffer(message);
+    switch (msg.whichMessage()) {
+      case SignalResponse_Message.join:
+        if (!_connected) {
+          _connected = true;
+          delegate.onConnected(msg.join);
+        }
+        break;
+      case SignalResponse_Message.answer:
+        delegate.onAnswer(toRTCSessionDescription(msg.answer));
+        break;
+      case SignalResponse_Message.offer:
+        delegate.onOffer(toRTCSessionDescription(msg.offer));
+        break;
+      case SignalResponse_Message.trickle:
+        delegate.onTrickle(toRTCIceCandidate(msg.trickle), msg.trickle.target);
+        break;
+      case SignalResponse_Message.update:
+        delegate.onParticipantUpdate(msg.update.participants);
+        break;
+      case SignalResponse_Message.trackPublished:
+        delegate.onLocalTrackPublished(msg.trackPublished);
+        break;
+      case SignalResponse_Message.speaker:
+        delegate.onActiveSpeakersChanged(msg.speaker.speakers);
+        break;
+      case SignalResponse_Message.leave:
+        delegate.onLeave(msg.leave);
+        break;
+      default:
+        log('unsupported message: ' + jsonEncode(msg));
+    }
+  }
+
+  _handleError(Object error) {
+    // TODO: test HTTP endpoint
+  }
+
+  _handleDone() {
+    _ws = null;
+  }
+}
+
+String _paramsForToken(String token) {
+  return '?access_token=$token&protocol=$protocolVersion';
+}
+
+RTCSessionDescription toRTCSessionDescription(SessionDescription sd) {
+  return new RTCSessionDescription(sd.sdp, sd.type);
+}
+
+SessionDescription fromRTCSessionDescription(RTCSessionDescription rsd) {
+  return new SessionDescription(type: rsd.type, sdp: rsd.sdp);
+}
+
+RTCIceCandidate toRTCIceCandidate(String candidateInit) {
+  var candInit = jsonDecode(candidateInit);
+  return new RTCIceCandidate(
+      candInit['candidate'], candInit['sdpMid'], candInit['sdpMLineIndex']);
+}
+
+String fromRTCIceCandidate(RTCIceCandidate candidate) {
+  return jsonEncode(candidate.toMap());
+}

lib/src/track/track.dart

@@ -0,0 +1,6 @@
+class TrackDimension {
+  int width;
+  int height;
+
+  TrackDimension(this.width, this.height);
+}

lib/src/transport.dart

@@ -0,0 +1,26 @@
+import 'package:flutter_webrtc/flutter_webrtc.dart';
+
+/// a wrapper around PeerConnection
+class PCTransport {
+  RTCPeerConnection pc;
+  List<RTCIceCandidate> pendingCandidates = [];
+  bool restartingIce = false;
+
+  PCTransport(this.pc);
+
+  Future<void> setRemoteDescription(RTCSessionDescription sd) async {
+    await pc.setRemoteDescription(sd);
+
+    Future.forEach<RTCIceCandidate>(pendingCandidates, (candidate) async {
+      await pc.addCandidate(candidate);
+    });
+  }
+
+  Future<void> addIceCandidate(RTCIceCandidate candidate) async {
+    var desc = await pc.getRemoteDescription();
+    if (desc != null && !restartingIce) {
+      return pc.addCandidate(candidate);
+    }
+    pendingCandidates.add(candidate);
+  }
+}

lib/src/version.dart

@@ -0,0 +1,2 @@
+const version = '0.1.0';
+const protocolVersion = 2;

pubspec.lock

@@ -1,27 +1,6 @@
 # Generated by pub
 # See https://dart.dev/tools/pub/glossary#lockfile
 packages:
-  _fe_analyzer_shared:
-    dependency: transitive
-    description:
-      name: _fe_analyzer_shared
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "22.0.0"
-  analyzer:
-    dependency: transitive
-    description:
-      name: analyzer
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "1.7.1"
-  args:
-    dependency: transitive
-    description:
-      name: args
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "2.1.1"
   async:
     dependency: transitive
     description:
@@ -50,13 +29,6 @@ packages:
       url: "https://pub.dartlang.org"
     source: hosted
     version: "1.2.0"
-  cli_util:
-    dependency: transitive
-    description:
-      name: cli_util
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "0.3.3"
   clock:
     dependency: transitive
     description:
@@ -71,13 +43,6 @@ packages:
       url: "https://pub.dartlang.org"
     source: hosted
     version: "1.15.0"
-  convert:
-    dependency: transitive
-    description:
-      name: convert
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "3.0.1"
   crypto:
     dependency: transitive
     description:
@@ -85,13 +50,6 @@ packages:
       url: "https://pub.dartlang.org"
     source: hosted
     version: "3.0.1"
-  dart_style:
-    dependency: transitive
-    description:
-      name: dart_style
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "1.3.14"
   fake_async:
     dependency: transitive
     description:
@@ -137,13 +95,6 @@ packages:
       url: "https://pub.dartlang.org"
     source: hosted
     version: "0.6.5"
-  glob:
-    dependency: transitive
-    description:
-      name: glob
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "2.0.1"
   matcher:
     dependency: transitive
     description:
@@ -158,13 +109,6 @@ packages:
       url: "https://pub.dartlang.org"
     source: hosted
     version: "1.3.0"
-  package_config:
-    dependency: transitive
-    description:
-      name: package_config
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "2.0.0"
   path:
     dependency: transitive
     description:
@@ -207,13 +151,6 @@ packages:
       url: "https://pub.dartlang.org"
     source: hosted
     version: "2.0.1"
-  pedantic:
-    dependency: transitive
-    description:
-      name: pedantic
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "1.11.1"
   platform:
     dependency: transitive
     description:
@@ -242,20 +179,6 @@ packages:
       url: "https://pub.dartlang.org"
     source: hosted
     version: "2.0.0"
-  protoc_plugin:
-    dependency: "direct main"
-    description:
-      name: protoc_plugin
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "20.0.0"
-  pub_semver:
-    dependency: transitive
-    description:
-      name: pub_semver
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "2.0.0"
   sky_engine:
     dependency: transitive
     description: flutter
@@ -317,13 +240,13 @@ packages:
       url: "https://pub.dartlang.org"
     source: hosted
     version: "2.1.0"
-  watcher:
+  web_socket_channel:
-    dependency: transitive
+    dependency: "direct main"
     description:
-      name: watcher
+      name: web_socket_channel
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "1.0.0"
+    version: "2.1.0"
   win32:
     dependency: transitive
     description:
@@ -338,13 +261,6 @@ packages:
       url: "https://pub.dartlang.org"
     source: hosted
     version: "0.2.0"
-  yaml:
-    dependency: transitive
-    description:
-      name: yaml
-      url: "https://pub.dartlang.org"
-    source: hosted
-    version: "3.1.0"
 sdks:
   dart: ">=2.13.0 <3.0.0"
   flutter: ">=1.22.0"

pubspec.yaml

@@ -12,7 +12,7 @@ dependencies:
     sdk: flutter
   flutter_webrtc: ^0.6.4
   protobuf: ^2.0.0
-  protoc_plugin: ^20.0.0
+  web_socket_channel: ^2.1.0
 
 dev_dependencies:
   flutter_test: