connectivity handling, audio publishing

lib/src/_websocket_api.dart

@@ -0,0 +1,5 @@
+import 'package:web_socket_channel/web_socket_channel.dart';
+
+Future<WebSocketChannel> connectToWebSocket(Uri uri) {
+  throw UnsupportedError('no implementations found');
+}

lib/src/_websocket_html.dart

@@ -0,0 +1,18 @@
+import 'dart:async';
+// ignore: avoid_web_libraries_in_flutter
+import 'dart:html';
+
+import 'package:web_socket_channel/html.dart';
+import 'package:web_socket_channel/web_socket_channel.dart';
+
+Future<WebSocketChannel> connectToWebSocket(Uri uri) {
+  var ws = WebSocket(uri.toString());
+  var completer = Completer<WebSocketChannel>();
+  ws.onOpen.first.then((_) {
+    completer.complete(HtmlWebSocketChannel(ws));
+  });
+  ws.onError.first.then((e) {
+    completer.completeError('could not connect');
+  });
+  return completer.future;
+}

lib/src/_websocket_io.dart

@@ -0,0 +1,14 @@
+import 'dart:io';
+
+import 'package:web_socket_channel/io.dart';
+import 'package:web_socket_channel/web_socket_channel.dart';
+
+Future<WebSocketChannel> connectToWebSocket(Uri uri) async {
+  try {
+    // ignore: close_sinks
+    var ws = await WebSocket.connect(uri.toString());
+    return IOWebSocketChannel(ws);
+  } catch (e) {
+    return Future.error(e);
+  }
+}

lib/src/errors.dart

@@ -13,6 +13,11 @@ class ConnectError extends LiveKitError {
   ConnectError([String msg = 'Failed to connect to server']) : super(msg);
 }
 
+class UnexpectedConnectionState extends LiveKitError {
+  UnexpectedConnectionState([String msg = 'Unexpected connection state'])
+      : super(msg);
+}
+
 class TrackCreateError extends LiveKitError {
   TrackCreateError([String msg = 'Failed to create track']) : super(msg);
 }

lib/src/participant/local_participant.dart

@@ -71,6 +71,9 @@ class LocalParticipant extends Participant {
     var trackInfo = await _engine.addTrack(
         cid: track.getCid(), name: track.name, kind: track.kind);
     var stream = await getMediaStream();
+    if (stream == null) {
+      return Future.error(TrackPublishError());
+    }
     var transceiverInit = new RTCRtpTransceiverInit(
       direction: TransceiverDirection.SendOnly,
       streams: [stream],

lib/src/room.dart

@@ -100,8 +100,16 @@ class Room extends ChangeNotifier with ParticipantDelegate {
     _engine.onParticipantUpdateCallback = _handleParticipantUpdate;
     _engine.onActiveSpeakerchangedCallback = _handleSpeakerUpdate;
     _engine.onDataMessageCallback = _handleDataPacket;
-
-    // TODO: handle reconnecting & reconnected events
+    _engine.onReconnected = () {
+      _state = RoomState.Connected;
+      delegate?.onReconnected();
+      notifyListeners();
+    };
+    _engine.onReconnecting = () {
+      _state = RoomState.Reconnecting;
+      delegate?.onReconnecting();
+      notifyListeners();
+    };
   }
 
   Future<Room> connect(String url, String token, [JoinOptions? opts]) async {
@@ -128,6 +136,9 @@ class Room extends ChangeNotifier with ParticipantDelegate {
     // 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(Duration(seconds: 5), () {
+      if (_state != RoomState.Disconnected) {
+        return;
+      }
       _state = RoomState.Disconnected;
       _connectCompleter?.completeError(ConnectError());
       _connectCompleter = null;
@@ -173,7 +184,8 @@ class Room extends ChangeNotifier with ParticipantDelegate {
     }
 
     for (var p in _participants.values) {
-      for (var pub in p.tracks.values) {
+      var tracks = List<TrackPublication>.from(p.tracks.values);
+      for (var pub in tracks) {
         p.unpublishTrack(pub.sid);
       }
     }

lib/src/rtc_engine.dart

@@ -3,6 +3,7 @@ import 'package:flutter_webrtc/flutter_webrtc.dart';
 
 import 'errors.dart';
 import 'extensions.dart';
+import 'logger.dart';
 import 'proto/livekit_rtc.pb.dart';
 import 'proto/livekit_models.pb.dart';
 import 'signal_client.dart';
@@ -12,6 +13,8 @@ import 'transport.dart';
 const lossyDataChannel = '_lossy';
 const reliableDataChannel = '_reliable';
 final connectionTimeout = new Duration(seconds: 5);
+final maxReconnectAttempts = 5;
+final iceRestartTimeout = new Duration(seconds: 10);
 
 typedef GenericCallback = void Function();
 typedef TrackCallback = void Function(
@@ -33,8 +36,10 @@ class RTCEngine with SignalClientDelegate {
   RTCDataChannel? reliableDC;
   RTCDataChannel? lossyDC;
   bool iceConnected = false;
+  bool isReconnecting = false;
   bool isClosed = true;
   Map<String, Completer<TrackInfo>> pendingTrackResolvers = {};
+  int reconnectAttempts = 0;
   // to complete join request
   Completer<JoinResponse>? joinCompleter;
   // remember url and token for reconnect
@@ -47,6 +52,8 @@ class RTCEngine with SignalClientDelegate {
   ParticipantUpdateCallback? onParticipantUpdateCallback;
   ActiveSpeakerChangedCallback? onActiveSpeakerchangedCallback;
   DataPacketCallback? onDataMessageCallback;
+  GenericCallback? onReconnecting;
+  GenericCallback? onReconnected;
   GenericCallback? onDisconnected;
 
   RTCEngine(this.client, RTCConfiguration? rtcConfig) {
@@ -112,17 +119,13 @@ class RTCEngine with SignalClientDelegate {
     return completer.future;
   }
 
-  negotiate([Map<String, dynamic>? constraints]) async {
+  negotiate({bool? iceRestart}) async {
     var pub = this.publisher;
     if (pub == null) {
       return;
     }
 
-    RTCSessionDescription? remoteDesc;
-    if (pub.pc.iceConnectionState != null) {
-      // when not initially connected, this crashes on iOS
-      remoteDesc = await pub.pc.getRemoteDescription();
-    }
+    var remoteDesc = await pub.getRemoteDescription();
 
     // handle cases that we couldn't create a new offer due to a pending answer
     // that's lost in transit
@@ -132,14 +135,64 @@ class RTCEngine with SignalClientDelegate {
       await pub.pc.setRemoteDescription(remoteDesc);
     }
 
-    if (constraints == null) {
-      constraints = {};
+    var constraints = <String, dynamic>{};
+    if (iceRestart != null && iceRestart) {
+      constraints['mandatory'] = {
+        'IceRestart': true,
+      };
     }
     var offer = await pub.pc.createOffer(constraints);
     await pub.pc.setLocalDescription(offer);
     client.sendOffer(offer);
   }
 
+  Future<void> reconnect() async {
+    if (isClosed) {
+      return;
+    }
+    var url = this.url;
+    var token = this.token;
+    if (url == null || token == null) {
+      throw ConnectError("could not reconnect without url and token");
+    }
+    if (reconnectAttempts == 0) {
+      onReconnecting?.call();
+    }
+    reconnectAttempts++;
+
+    try {
+      isReconnecting = true;
+      await client.reconnect(url, token);
+
+      var pub = this.publisher;
+      var sub = this.subscriber;
+      if (pub == null || sub == null) {
+        throw UnexpectedConnectionState('publisher or subscribers is null');
+      }
+
+      pub.restartingIce = true;
+      sub.restartingIce = true;
+
+      await negotiate(iceRestart: true);
+    } catch (e) {
+      isReconnecting = false;
+      return Future.error(e);
+    }
+
+    // wait for connectivity to change
+    var startTime = DateTime.now();
+    while (DateTime.now().difference(startTime) < iceRestartTimeout) {
+      if (iceConnected) {
+        isReconnecting = false;
+        return;
+      }
+      await Future.delayed(Duration(milliseconds: 100));
+    }
+
+    isReconnecting = false;
+    return Future.error(ConnectError('could not reconnect ICE'));
+  }
+
   _configurePeerConnections() async {
     if (publisher != null) {
       return;
@@ -174,11 +227,16 @@ class RTCEngine with SignalClientDelegate {
         case RTCIceConnectionState.RTCIceConnectionStateConnected:
           if (!iceConnected) {
             iceConnected = true;
-            onICEConnected?.call();
+            if (isReconnecting) {
+              onReconnected?.call();
+            } else {
+              onICEConnected?.call();
+            }
           }
           break;
 
         case RTCIceConnectionState.RTCIceConnectionStateFailed:
+          iceConnected = false;
           // trigger reconnect sequence
           _handleDisconnect('peerconnection');
           break;
@@ -230,7 +288,25 @@ class RTCEngine with SignalClientDelegate {
   }
 
   _handleDisconnect(String reason) {
-    // TODO: implement method
+    if (isClosed) {
+      return;
+    }
+    logger.fine('disconnected $reason');
+    if (this.reconnectAttempts >= maxReconnectAttempts) {
+      logger.info('could not connect after $reconnectAttempts, giving up');
+      this.close();
+      onDisconnected?.call();
+      return;
+    }
+
+    var delay = (reconnectAttempts * reconnectAttempts) * 300;
+    Future.delayed(Duration(milliseconds: delay), () {
+      reconnect().then((_) {
+        reconnectAttempts = 0;
+      }).catchError((e) {
+        _handleDisconnect(reason);
+      });
+    });
   }
 
   //------------------ SignalClient Delegate methods -------------------------//
@@ -263,7 +339,7 @@ class RTCEngine with SignalClientDelegate {
   }
 
   void onClose([String? reason]) {
-    // TODO: handle reconnect when signal interrupted
+    _handleDisconnect("signal");
   }
 
   void onOffer(RTCSessionDescription sd) async {

lib/src/signal_client.dart

@@ -7,6 +7,9 @@ import './track/track.dart';
 import './version.dart';
 import './proto/livekit_models.pb.dart';
 import './proto/livekit_rtc.pb.dart';
+import '_websocket_api.dart'
+    if (dart.library.io) '_websocket_io.dart'
+    if (dart.library.html) '_websocket_html.dart' as platform;
 
 class JoinOptions {
   final bool? autoSubscribe;
@@ -53,18 +56,30 @@ class SignalClient {
     }
     var uri = Uri.parse(url + params);
 
-    try {
-      var ws = WebSocketChannel.connect(uri);
+    platform.connectToWebSocket(uri).then((ws) {
       ws.stream
           .listen(_handleMessage, onError: _handleError, onDone: _handleDone);
       _ws = ws;
-    } catch (e) {
-      // failed before error handler is installed, fail immediately
-      _handleError(e);
-    }
+    }).catchError((error) {
+      // TODO: ping api endpoint
+      _handleError(error);
+    });
   }
 
-  Future<void> reconnect(String url, String token) async {}
+  Future<void> reconnect(String url, String token) async {
+    _connected = false;
+    _ws?.sink.close();
+    _ws = null;
+
+    url += '/rtc';
+    var params = _paramsForToken(token);
+    params += '&reconnect=1';
+    var uri = Uri.parse(url + params);
+
+    var ws = await platform.connectToWebSocket(uri);
+    _ws = ws;
+    _connected = true;
+  }
 
   close() {
     this._connected = false;
@@ -147,7 +162,7 @@ class SignalClient {
 
   _sendRequest(SignalRequest req) {
     if (this._ws == null) {
-      log('could not send message, not connected: ' + jsonEncode(req));
+      log('could not send message, not connected');
       return;
     }
 

lib/src/track/local_audio_track.dart

@@ -6,12 +6,13 @@ import 'options.dart';
 import 'track.dart';
 
 class LocalAudioTrack extends Track {
-  MediaStream mediaStream;
+  MediaStream? mediaStream;
 
   LocalAudioTrack(String name, MediaStreamTrack track, this.mediaStream)
       : super(TrackType.AUDIO, name, track);
 
-  Future<LocalAudioTrack> createTrack(LocalAudioTrackOptions? options) async {
+  static Future<LocalAudioTrack> createTrack(
+      [LocalAudioTrackOptions? options]) async {
     try {
       var stream = await navigator.mediaDevices.getUserMedia({
         "audio": true,
@@ -22,9 +23,16 @@ class LocalAudioTrack extends Track {
         return Future.error(TrackCreateError());
       }
 
-      return LocalAudioTrack("", stream.getVideoTracks().first, stream);
+      return LocalAudioTrack("", stream.getAudioTracks().first, stream);
     } catch (e) {
       return Future.error(e);
     }
   }
+
+  @override
+  stop() {
+    super.stop();
+    mediaStream?.dispose();
+    mediaStream = null;
+  }
 }

lib/src/track/local_video_track.dart

@@ -8,8 +8,8 @@ class LocalVideoTrack extends VideoTrack {
   LocalVideoTrack(String name, MediaStreamTrack mediaTrack, MediaStream stream)
       : super(name, mediaTrack, stream);
 
-  Future<LocalVideoTrack> createCameraTrack(
-      LocalVideoTrackOptions? options) async {
+  static Future<LocalVideoTrack> createCameraTrack(
+      [LocalVideoTrackOptions? options]) async {
     if (options == null) {
       options = LocalVideoTrackOptions(params: VideoPresets.qhd);
     }
@@ -24,7 +24,7 @@ class LocalVideoTrack extends VideoTrack {
         return Future.error(TrackCreateError());
       }
 
-      return LocalVideoTrack("", stream.getVideoTracks().first, stream);
+      return LocalVideoTrack("camera", stream.getVideoTracks().first, stream);
     } catch (e) {
       return Future.error(e);
     }

lib/src/track/video_track.dart

@@ -4,8 +4,15 @@ import '../proto/livekit_models.pb.dart';
 import 'track.dart';
 
 class VideoTrack extends Track {
-  MediaStream mediaStream;
+  MediaStream? mediaStream;
 
   VideoTrack(String name, MediaStreamTrack mediaTrack, this.mediaStream)
       : super(TrackType.VIDEO, name, mediaTrack);
+
+  @override
+  stop() {
+    super.stop();
+    mediaStream?.dispose();
+    mediaStream = null;
+  }
 }

lib/src/transport.dart

@@ -11,16 +11,26 @@ class PCTransport {
   Future<void> setRemoteDescription(RTCSessionDescription sd) async {
     await pc.setRemoteDescription(sd);
 
-    Future.forEach<RTCIceCandidate>(pendingCandidates, (candidate) async {
+    await Future.forEach<RTCIceCandidate>(pendingCandidates, (candidate) async {
       await pc.addCandidate(candidate);
     });
+
+    pendingCandidates.clear();
+    restartingIce = false;
   }
 
   Future<void> addIceCandidate(RTCIceCandidate candidate) async {
-    var desc = await pc.getRemoteDescription();
+    var desc = await getRemoteDescription();
     if (desc != null && !restartingIce) {
       return pc.addCandidate(candidate);
     }
     pendingCandidates.add(candidate);
   }
+
+  Future<RTCSessionDescription?> getRemoteDescription() async {
+    if (pc.iceConnectionState == null) {
+      return null;
+    }
+    return pc.getRemoteDescription();
+  }
 }

lib/src/widget/video_track_renderer.dart

@@ -34,7 +34,9 @@ class _VideoTrackRendererState extends State<VideoTrackRenderer> {
 
   _initRenderer() async {
     await _renderer.initialize();
-    _renderer.srcObject = widget.track.mediaStream;
+    setState(() {
+      _renderer.srcObject = widget.track.mediaStream;
+    });
   }
 
   @override

pubspec.yaml

@@ -1,6 +1,6 @@
 name: livekit_client
 description: Flutter client for LiveKit
-version: 0.0.1
+version: 0.1.0
 homepage: https://livekit.io
 
 environment: