From 593890331ee3b314e352d365864465a386b8f15f Mon Sep 17 00:00:00 2001 From: David Zhao Date: Wed, 4 Aug 2021 10:59:41 -0700 Subject: [PATCH] connectivity handling, audio publishing --- lib/src/_websocket_api.dart | 5 ++ lib/src/_websocket_html.dart | 18 ++++ lib/src/_websocket_io.dart | 14 ++++ lib/src/errors.dart | 5 ++ lib/src/participant/local_participant.dart | 3 + lib/src/room.dart | 18 +++- lib/src/rtc_engine.dart | 98 +++++++++++++++++++--- lib/src/signal_client.dart | 31 +++++-- lib/src/track/local_audio_track.dart | 14 +++- lib/src/track/local_video_track.dart | 6 +- lib/src/track/video_track.dart | 9 +- lib/src/transport.dart | 14 +++- lib/src/widget/video_track_renderer.dart | 4 +- pubspec.yaml | 2 +- 14 files changed, 208 insertions(+), 33 deletions(-) create mode 100644 lib/src/_websocket_api.dart create mode 100644 lib/src/_websocket_html.dart create mode 100644 lib/src/_websocket_io.dart diff --git a/lib/src/_websocket_api.dart b/lib/src/_websocket_api.dart new file mode 100644 index 0000000..ab79b0b --- /dev/null +++ b/lib/src/_websocket_api.dart @@ -0,0 +1,5 @@ +import 'package:web_socket_channel/web_socket_channel.dart'; + +Future connectToWebSocket(Uri uri) { + throw UnsupportedError('no implementations found'); +} diff --git a/lib/src/_websocket_html.dart b/lib/src/_websocket_html.dart new file mode 100644 index 0000000..9f84dce --- /dev/null +++ b/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 connectToWebSocket(Uri uri) { + var ws = WebSocket(uri.toString()); + var completer = Completer(); + ws.onOpen.first.then((_) { + completer.complete(HtmlWebSocketChannel(ws)); + }); + ws.onError.first.then((e) { + completer.completeError('could not connect'); + }); + return completer.future; +} diff --git a/lib/src/_websocket_io.dart b/lib/src/_websocket_io.dart new file mode 100644 index 0000000..bfbe5ff --- /dev/null +++ b/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 connectToWebSocket(Uri uri) async { + try { + // ignore: close_sinks + var ws = await WebSocket.connect(uri.toString()); + return IOWebSocketChannel(ws); + } catch (e) { + return Future.error(e); + } +} diff --git a/lib/src/errors.dart b/lib/src/errors.dart index 0fe9c40..39e9ebf 100644 --- a/lib/src/errors.dart +++ b/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); } diff --git a/lib/src/participant/local_participant.dart b/lib/src/participant/local_participant.dart index 5be92d9..64f4fc8 100644 --- a/lib/src/participant/local_participant.dart +++ b/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], diff --git a/lib/src/room.dart b/lib/src/room.dart index 47c5569..b737f62 100644 --- a/lib/src/room.dart +++ b/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 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.from(p.tracks.values); + for (var pub in tracks) { p.unpublishTrack(pub.sid); } } diff --git a/lib/src/rtc_engine.dart b/lib/src/rtc_engine.dart index e053435..ffdf1ab 100644 --- a/lib/src/rtc_engine.dart +++ b/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> pendingTrackResolvers = {}; + int reconnectAttempts = 0; // to complete join request Completer? 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? 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 = {}; + if (iceRestart != null && iceRestart) { + constraints['mandatory'] = { + 'IceRestart': true, + }; } var offer = await pub.pc.createOffer(constraints); await pub.pc.setLocalDescription(offer); client.sendOffer(offer); } + Future 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 { diff --git a/lib/src/signal_client.dart b/lib/src/signal_client.dart index 311aa7b..3686256 100644 --- a/lib/src/signal_client.dart +++ b/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 reconnect(String url, String token) async {} + Future 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; } diff --git a/lib/src/track/local_audio_track.dart b/lib/src/track/local_audio_track.dart index 8c0606a..d633160 100644 --- a/lib/src/track/local_audio_track.dart +++ b/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 createTrack(LocalAudioTrackOptions? options) async { + static Future 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; + } } diff --git a/lib/src/track/local_video_track.dart b/lib/src/track/local_video_track.dart index 7619310..9523f6a 100644 --- a/lib/src/track/local_video_track.dart +++ b/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 createCameraTrack( - LocalVideoTrackOptions? options) async { + static Future 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); } diff --git a/lib/src/track/video_track.dart b/lib/src/track/video_track.dart index 1f77b0e..8423a98 100644 --- a/lib/src/track/video_track.dart +++ b/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; + } } diff --git a/lib/src/transport.dart b/lib/src/transport.dart index 2141dbf..2798bd4 100644 --- a/lib/src/transport.dart +++ b/lib/src/transport.dart @@ -11,16 +11,26 @@ class PCTransport { Future setRemoteDescription(RTCSessionDescription sd) async { await pc.setRemoteDescription(sd); - Future.forEach(pendingCandidates, (candidate) async { + await Future.forEach(pendingCandidates, (candidate) async { await pc.addCandidate(candidate); }); + + pendingCandidates.clear(); + restartingIce = false; } Future 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 getRemoteDescription() async { + if (pc.iceConnectionState == null) { + return null; + } + return pc.getRemoteDescription(); + } } diff --git a/lib/src/widget/video_track_renderer.dart b/lib/src/widget/video_track_renderer.dart index 273522b..65a0d1f 100644 --- a/lib/src/widget/video_track_renderer.dart +++ b/lib/src/widget/video_track_renderer.dart @@ -34,7 +34,9 @@ class _VideoTrackRendererState extends State { _initRenderer() async { await _renderer.initialize(); - _renderer.srcObject = widget.track.mediaStream; + setState(() { + _renderer.srcObject = widget.track.mediaStream; + }); } @override diff --git a/pubspec.yaml b/pubspec.yaml index cecdb2f..92dc4ce 100644 --- a/pubspec.yaml +++ b/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: