Simulcast, Screen sharing & Various improvements (#4)
* Respect `RTCIceTransportPolicy` enum and organize * Simplify syntax where possible etc. * Combine `VideoPreset` and `VideoPresets` * Default values for `ConnectOptions` * Build URI instead of String manipulation * Slight modifications to Exception * `LiveKitTheme` for example * `VideoEncoding` class * Organize imports * First simulcast implementation * Remove unnecessary try-catches * Update Android settings * Remember uri and token * example improvements * `fit` parameter for VideoTrackRenderer * Simulcast option for example * Pass `defaultPublishOptions` * Show only `VideoQuality` * Pass tests * Better buildUri logic * Named parameter to positional * Explicit imports * `VideoParameter` instead of `VideoPreset` * Use `mediaTrack.getSettings` when possible * Safer dispose logic * Safer `PCTransport` Update transport.dart * Synchronized events for `SignalClient` * Use logger instead of print * Make example compile for iOS * First screen share implementation * Make example work with screen share * Example improvement * Code optimization * Don't depend on web_socket_channel * Fix: Unpublish track bug * Show participant mute state & identity * Update protos * Remote mute/unmute * iOS Background mode * Separate `createCameraTrack` and `createScreenTrack` * Clean up * PB fix * format * Fix analyzer warning * Android clean up * Update README.md * Clean up
This commit is contained in:
Hiroshi Horie
+210
-153
@@ -3,134 +3,191 @@ import 'dart:convert';
|
||||
import 'dart:developer';
|
||||
|
||||
import 'package:flutter_webrtc/flutter_webrtc.dart';
|
||||
import 'package:web_socket_channel/web_socket_channel.dart';
|
||||
import 'package:http/http.dart' as http;
|
||||
import 'package:livekit_client/src/ws/interface.dart';
|
||||
import 'package:synchronized/synchronized.dart' as sync;
|
||||
|
||||
import 'errors.dart';
|
||||
import 'logger.dart';
|
||||
import 'options.dart';
|
||||
import 'proto/livekit_models.pb.dart' as lk_models;
|
||||
import 'proto/livekit_rtc.pb.dart' as lk_rtc;
|
||||
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;
|
||||
|
||||
mixin SignalClientDelegate {
|
||||
// initial connection established
|
||||
void onConnected(JoinResponse response);
|
||||
Future<void> onConnected(lk_rtc.JoinResponse response);
|
||||
// websocket has closed
|
||||
void onClose([String? reason]);
|
||||
Future<void> onClose([String? reason]);
|
||||
// when a server offer is received
|
||||
void onOffer(RTCSessionDescription sd);
|
||||
Future<void> onOffer(RTCSessionDescription sd);
|
||||
// when an answer from server is received
|
||||
void onAnswer(RTCSessionDescription sd);
|
||||
Future<void> onAnswer(RTCSessionDescription sd);
|
||||
// when server has a new ICE candidate
|
||||
void onTrickle(RTCIceCandidate candidate, SignalTarget target);
|
||||
Future<void> onTrickle(RTCIceCandidate candidate, lk_rtc.SignalTarget target);
|
||||
// participant has changed
|
||||
void onParticipantUpdate(List<ParticipantInfo> updates);
|
||||
Future<void> onParticipantUpdate(List<lk_models.ParticipantInfo> updates);
|
||||
// when a track has been added successfully
|
||||
void onLocalTrackPublished(TrackPublishedResponse response);
|
||||
Future<void> onLocalTrackPublished(lk_rtc.TrackPublishedResponse response);
|
||||
// active speaker has changed
|
||||
void onActiveSpeakersChanged(List<SpeakerInfo> speakers);
|
||||
Future<void> onActiveSpeakersChanged(List<lk_models.SpeakerInfo> speakers);
|
||||
// when server sends this client a leave message
|
||||
void onLeave(LeaveRequest req);
|
||||
Future<void> onLeave(lk_rtc.LeaveRequest req);
|
||||
// explicit mute track
|
||||
Future<void> onMuteTrack(lk_rtc.MuteTrackRequest req);
|
||||
}
|
||||
|
||||
extension LKUriExt on Uri {
|
||||
bool get isSecureScheme => ['https', 'wss'].contains(scheme);
|
||||
}
|
||||
|
||||
class SignalClient {
|
||||
SignalClientDelegate? delegate;
|
||||
static const protocolVersion = 2;
|
||||
|
||||
final _lock = sync.Lock();
|
||||
SignalClientDelegate? delegate;
|
||||
bool _connected = false;
|
||||
WebSocketChannel? _ws;
|
||||
LKWebSocket? _ws;
|
||||
|
||||
SignalClient();
|
||||
|
||||
bool get connected => _connected;
|
||||
|
||||
Future<void> join(String url, String token, JoinOptions? options) async {
|
||||
final rtcUrl = '$url/rtc';
|
||||
var params = _joinParams(token);
|
||||
if (options != null && options.autoSubscribe != null) {
|
||||
params += '&auto_subscribe=${options.autoSubscribe! ? '1' : '0'}';
|
||||
}
|
||||
Uri _buildUri(
|
||||
String uriOrString, {
|
||||
required String token,
|
||||
ConnectOptions? options,
|
||||
bool reconnect = false,
|
||||
bool validate = false,
|
||||
bool forceSecure = false,
|
||||
}) {
|
||||
final Uri uri = Uri.parse(uriOrString);
|
||||
|
||||
final useSecure = uri.isSecureScheme || forceSecure;
|
||||
final httpScheme = useSecure ? 'https' : 'http';
|
||||
final wsScheme = useSecure ? 'wss' : 'ws';
|
||||
|
||||
return uri.replace(
|
||||
scheme: validate ? httpScheme : wsScheme,
|
||||
path: validate ? 'validate' : 'rtc',
|
||||
queryParameters: <String, String>{
|
||||
'access_token': token,
|
||||
if (options != null) 'auto_subscribe': options.autoSubscribe ? '1' : '0',
|
||||
if (reconnect) 'reconnect': '1',
|
||||
'protocol': protocolVersion.toString(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> join(
|
||||
String uriString,
|
||||
String token, {
|
||||
ConnectOptions? options,
|
||||
}) async {
|
||||
// Create default options if null
|
||||
options ??= const ConnectOptions();
|
||||
|
||||
final rtcUri = _buildUri(
|
||||
uriString,
|
||||
token: token,
|
||||
options: options,
|
||||
);
|
||||
|
||||
try {
|
||||
final ws = await platform.connectToWebSocket(Uri.parse(rtcUrl + params));
|
||||
ws.stream.listen(_handleMessage, onError: _handleError, onDone: _handleDone);
|
||||
_ws = ws;
|
||||
} catch (e) {
|
||||
final completer = Completer<void>();
|
||||
final validateUri = Uri.parse('http${rtcUrl.substring(2)}/validate$params');
|
||||
http.get(validateUri).then((response) {
|
||||
if (response.statusCode != 200) {
|
||||
completer.completeError(ConnectError(response.body));
|
||||
} else {
|
||||
completer.completeError(ConnectError());
|
||||
}
|
||||
}).catchError((dynamic e) {
|
||||
completer.completeError(ConnectError());
|
||||
});
|
||||
_ws = await LKWebSocket.connect(
|
||||
rtcUri,
|
||||
LKWebSocketOptions(
|
||||
onData: _onSocketData,
|
||||
onDispose: _onSocketDone,
|
||||
onError: _handleError,
|
||||
),
|
||||
);
|
||||
} catch (socketError) {
|
||||
// Re-build same uri for validate mode
|
||||
final validateUri = _buildUri(
|
||||
uriString,
|
||||
token: token,
|
||||
options: options,
|
||||
validate: true,
|
||||
forceSecure: rtcUri.isSecureScheme,
|
||||
);
|
||||
|
||||
return completer.future;
|
||||
// Attempt Validation
|
||||
try {
|
||||
final validateResponse = await http.get(validateUri);
|
||||
if (validateResponse.statusCode != 200) throw ConnectError(validateResponse.body);
|
||||
throw ConnectError();
|
||||
} catch (error) {
|
||||
// Pass it up if it's already a `ConnectError`
|
||||
if (error is ConnectError) rethrow;
|
||||
// HTTP doesn't work either
|
||||
throw ConnectError();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> reconnect(String url, String token) async {
|
||||
Future<void> reconnect(
|
||||
String uriString,
|
||||
String token,
|
||||
) async {
|
||||
_connected = false;
|
||||
_ws?.sink.close();
|
||||
_ws?.dispose();
|
||||
_ws = null;
|
||||
|
||||
url += '/rtc';
|
||||
var params = _joinParams(token);
|
||||
params += '&reconnect=1';
|
||||
final uri = Uri.parse(url + params);
|
||||
final rtcUri = _buildUri(
|
||||
uriString,
|
||||
token: token,
|
||||
reconnect: true,
|
||||
);
|
||||
|
||||
_ws = await LKWebSocket.connect(
|
||||
rtcUri,
|
||||
LKWebSocketOptions(
|
||||
onData: _onSocketData,
|
||||
onDispose: _onSocketDone,
|
||||
onError: _handleError,
|
||||
),
|
||||
);
|
||||
|
||||
final ws = await platform.connectToWebSocket(uri);
|
||||
_ws = ws;
|
||||
_connected = true;
|
||||
}
|
||||
|
||||
void close() {
|
||||
_connected = false;
|
||||
_ws?.sink.close();
|
||||
_ws?.dispose();
|
||||
}
|
||||
|
||||
void sendOffer(RTCSessionDescription offer) {
|
||||
_sendRequest(SignalRequest(
|
||||
offer: fromRTCSessionDescription(offer),
|
||||
));
|
||||
}
|
||||
void sendOffer(RTCSessionDescription offer) => _sendRequest(lk_rtc.SignalRequest(
|
||||
offer: fromRTCSessionDescription(offer),
|
||||
));
|
||||
|
||||
void sendAnswer(RTCSessionDescription answer) {
|
||||
_sendRequest(SignalRequest(
|
||||
answer: fromRTCSessionDescription(answer),
|
||||
));
|
||||
}
|
||||
void sendAnswer(RTCSessionDescription answer) => _sendRequest(lk_rtc.SignalRequest(
|
||||
answer: fromRTCSessionDescription(answer),
|
||||
));
|
||||
|
||||
void sendIceCandidate(RTCIceCandidate candidate, SignalTarget target) {
|
||||
_sendRequest(SignalRequest(
|
||||
trickle: TrickleRequest(
|
||||
candidateInit: fromRTCIceCandidate(candidate),
|
||||
target: target,
|
||||
)));
|
||||
}
|
||||
void sendIceCandidate(RTCIceCandidate candidate, lk_rtc.SignalTarget target) => _sendRequest(
|
||||
lk_rtc.SignalRequest(
|
||||
trickle: lk_rtc.TrickleRequest(
|
||||
candidateInit: fromRTCIceCandidate(candidate),
|
||||
target: target,
|
||||
),
|
||||
),
|
||||
);
|
||||
|
||||
void sendMuteTrack(String trackSid, bool muted) {
|
||||
_sendRequest(SignalRequest(
|
||||
mute: MuteTrackRequest(
|
||||
sid: trackSid,
|
||||
muted: muted,
|
||||
),
|
||||
));
|
||||
}
|
||||
void sendMuteTrack(String trackSid, bool muted) => _sendRequest(lk_rtc.SignalRequest(
|
||||
mute: lk_rtc.MuteTrackRequest(
|
||||
sid: trackSid,
|
||||
muted: muted,
|
||||
),
|
||||
));
|
||||
|
||||
void sendAddTrack(
|
||||
{required String cid,
|
||||
required String name,
|
||||
required TrackType type,
|
||||
TrackDimension? dimension}) {
|
||||
final req = AddTrackRequest(
|
||||
void sendAddTrack({
|
||||
required String cid,
|
||||
required String name,
|
||||
required lk_models.TrackType type,
|
||||
TrackDimension? dimension,
|
||||
}) {
|
||||
final req = lk_rtc.AddTrackRequest(
|
||||
cid: cid,
|
||||
name: name,
|
||||
type: type,
|
||||
@@ -139,109 +196,109 @@ class SignalClient {
|
||||
req.width = dimension.width;
|
||||
req.height = dimension.height;
|
||||
}
|
||||
_sendRequest(SignalRequest(
|
||||
_sendRequest(lk_rtc.SignalRequest(
|
||||
addTrack: req,
|
||||
));
|
||||
}
|
||||
|
||||
void sendUpdateTrackSettings(UpdateTrackSettings settings) {
|
||||
_sendRequest(SignalRequest(
|
||||
trackSetting: settings,
|
||||
));
|
||||
}
|
||||
void sendUpdateTrackSettings(lk_rtc.UpdateTrackSettings settings) =>
|
||||
_sendRequest(lk_rtc.SignalRequest(
|
||||
trackSetting: settings,
|
||||
));
|
||||
|
||||
void sendUpdateSubscription(UpdateSubscription subscription) {
|
||||
_sendRequest(SignalRequest(
|
||||
subscription: subscription,
|
||||
));
|
||||
}
|
||||
void sendUpdateSubscription(lk_rtc.UpdateSubscription subscription) =>
|
||||
_sendRequest(lk_rtc.SignalRequest(
|
||||
subscription: subscription,
|
||||
));
|
||||
|
||||
void sendSetSimulcastLayers(String trackSid, List<VideoQuality> layers) {
|
||||
_sendRequest(SignalRequest(
|
||||
simulcast: SetSimulcastLayers(
|
||||
trackSid: trackSid,
|
||||
layers: layers,
|
||||
)));
|
||||
}
|
||||
void sendSetSimulcastLayers(String trackSid, List<lk_rtc.VideoQuality> layers) =>
|
||||
_sendRequest(lk_rtc.SignalRequest(
|
||||
simulcast: lk_rtc.SetSimulcastLayers(
|
||||
trackSid: trackSid,
|
||||
layers: layers,
|
||||
),
|
||||
));
|
||||
|
||||
void sendLeave() {
|
||||
_sendRequest(SignalRequest(
|
||||
leave: LeaveRequest(),
|
||||
));
|
||||
}
|
||||
void sendLeave() => _sendRequest(lk_rtc.SignalRequest(
|
||||
leave: lk_rtc.LeaveRequest(),
|
||||
));
|
||||
|
||||
void _sendRequest(SignalRequest req) {
|
||||
void _sendRequest(lk_rtc.SignalRequest req) {
|
||||
if (_ws == null) {
|
||||
log('could not send message, not connected');
|
||||
return;
|
||||
}
|
||||
|
||||
final buf = req.writeToBuffer();
|
||||
_ws?.sink.add(buf);
|
||||
_ws?.send(buf);
|
||||
}
|
||||
|
||||
void _handleMessage(dynamic message) {
|
||||
if (message is! List<int>) {
|
||||
return;
|
||||
}
|
||||
final 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.candidateInit), 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: ' + json.encode(msg));
|
||||
}
|
||||
Future<void> _onSocketData(dynamic message) async {
|
||||
if (message is! List<int>) return;
|
||||
final msg = lk_rtc.SignalResponse.fromBuffer(message);
|
||||
|
||||
// Ensure previous delegate method's future is completed
|
||||
// before calling another method
|
||||
await _lock.synchronized(() async {
|
||||
//
|
||||
switch (msg.whichMessage()) {
|
||||
case lk_rtc.SignalResponse_Message.join:
|
||||
if (!_connected) {
|
||||
_connected = true;
|
||||
await delegate?.onConnected(msg.join);
|
||||
}
|
||||
break;
|
||||
case lk_rtc.SignalResponse_Message.answer:
|
||||
await delegate?.onAnswer(toRTCSessionDescription(msg.answer));
|
||||
break;
|
||||
case lk_rtc.SignalResponse_Message.offer:
|
||||
await delegate?.onOffer(toRTCSessionDescription(msg.offer));
|
||||
break;
|
||||
case lk_rtc.SignalResponse_Message.trickle:
|
||||
await delegate?.onTrickle(
|
||||
toRTCIceCandidate(msg.trickle.candidateInit),
|
||||
msg.trickle.target,
|
||||
);
|
||||
break;
|
||||
case lk_rtc.SignalResponse_Message.update:
|
||||
await delegate?.onParticipantUpdate(msg.update.participants);
|
||||
break;
|
||||
case lk_rtc.SignalResponse_Message.trackPublished:
|
||||
await delegate?.onLocalTrackPublished(msg.trackPublished);
|
||||
break;
|
||||
case lk_rtc.SignalResponse_Message.speaker:
|
||||
await delegate?.onActiveSpeakersChanged(msg.speaker.speakers);
|
||||
break;
|
||||
case lk_rtc.SignalResponse_Message.leave:
|
||||
await delegate?.onLeave(msg.leave);
|
||||
break;
|
||||
case lk_rtc.SignalResponse_Message.mute:
|
||||
await delegate?.onMuteTrack(msg.mute);
|
||||
break;
|
||||
default:
|
||||
log('unsupported message: ' + json.encode(msg));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void _handleError(Object error) {
|
||||
void _handleError(dynamic error) {
|
||||
logger.warning('received websocket error $error');
|
||||
}
|
||||
|
||||
void _handleDone() {
|
||||
if (!_connected) {
|
||||
return;
|
||||
}
|
||||
void _onSocketDone() {
|
||||
if (!_connected) return;
|
||||
_ws = null;
|
||||
_connected = false;
|
||||
delegate?.onClose();
|
||||
}
|
||||
}
|
||||
|
||||
String _joinParams(String token) {
|
||||
return '?access_token=$token&protocol=$protocolVersion';
|
||||
}
|
||||
|
||||
RTCSessionDescription toRTCSessionDescription(SessionDescription sd) {
|
||||
RTCSessionDescription toRTCSessionDescription(lk_rtc.SessionDescription sd) {
|
||||
return RTCSessionDescription(sd.sdp, sd.type);
|
||||
}
|
||||
|
||||
SessionDescription fromRTCSessionDescription(RTCSessionDescription rsd) {
|
||||
return SessionDescription(type: rsd.type, sdp: rsd.sdp);
|
||||
lk_rtc.SessionDescription fromRTCSessionDescription(RTCSessionDescription rsd) {
|
||||
return lk_rtc.SessionDescription(type: rsd.type, sdp: rsd.sdp);
|
||||
}
|
||||
|
||||
RTCIceCandidate toRTCIceCandidate(String candidateInit) {
|
||||
|
||||
Reference in New Issue
Block a user