diff --git a/lib/src/extensions.dart b/lib/src/extensions.dart new file mode 100644 index 0000000..1ea5136 --- /dev/null +++ b/lib/src/extensions.dart @@ -0,0 +1,39 @@ +class RTCConfiguration { + int? iceCandidatePoolSize; + List? iceServers; + String? iceTransportPolicy; + + Map toMap() { + var iceServersMap = []; + iceServers?.forEach((element) { + iceServersMap.add(element.toMap()); + }); + return { + if (iceCandidatePoolSize != null) + "iceCandidatePoolSize": iceCandidatePoolSize, + if (iceServersMap.isNotEmpty) "iceServers": iceServersMap, + if (iceTransportPolicy != null) "iceTransportPolicy": iceTransportPolicy, + }; + } +} + +class RTCIceServer { + List urls; + String? username; + String? credential; + + RTCIceServer({required this.urls, this.username, this.credential}); + + Map toMap() { + return { + "urls": urls, + if (username != null) + "username": username, + if (credential != null) + "credential": credential, + }; + } +} + +const RTCIceTransportPolicyAll = 'all'; +const RTCIceTransportPolicyRelay = 'relay'; diff --git a/lib/src/logger.dart b/lib/src/logger.dart new file mode 100644 index 0000000..7fc435f --- /dev/null +++ b/lib/src/logger.dart @@ -0,0 +1,3 @@ +import 'package:logging/logging.dart'; + +final logger = Logger("livekit"); diff --git a/lib/src/participant/local_participant.dart b/lib/src/participant/local_participant.dart new file mode 100644 index 0000000..2d54945 --- /dev/null +++ b/lib/src/participant/local_participant.dart @@ -0,0 +1,5 @@ +import 'participant.dart'; + +class LocalParticipant extends Participant { + +} diff --git a/lib/src/participant/participant.dart b/lib/src/participant/participant.dart new file mode 100644 index 0000000..e8b1209 --- /dev/null +++ b/lib/src/participant/participant.dart @@ -0,0 +1,95 @@ +import '../proto/livekit_models.pb.dart'; +import '../track/track_publication.dart'; + +mixin ParticipantDelegate { + void onMetadataChanged(Participant participant); + void onSpeakingChanged(Participant participant, bool speaking); +} + +class Participant { + Map audioTracks = {}; + Map videoTracks = {}; + + /// map of track sid => published track + Map tracks = {}; + + /// audio level between 0-1, 1 being the loudest + double audioLevel = 0; + + /// server assigned unique id + String sid; + + /// user-assigned identity + String identity; + + /// client-assigned metadata, opaque to livekit + String? metadata; + + /// when the participant had last spoken + DateTime? lastSpokeAt; + + Participant(this.sid, this.identity); + + ParticipantDelegate? _roomDelegate; + ParticipantDelegate? delegate; + + ParticipantInfo? _participantInfo; + bool _isSpeaking = false; + + DateTime get joinedAt { + var pi = _participantInfo; + if (pi != null) { + return DateTime.fromMillisecondsSinceEpoch((pi.joinedAt as int) * 1000, + isUtc: true); + } + return DateTime.now(); + } + + /// if participant is currently speaking + bool get isSpeaking => _isSpeaking; + + set isSpeaking(bool speaking) { + if (_isSpeaking != speaking) { + return; + } + _isSpeaking = speaking; + if (speaking) { + lastSpokeAt = DateTime.now(); + } + delegate?.onSpeakingChanged(this, speaking); + _roomDelegate?.onSpeakingChanged(this, speaking); + } + + _setMetadata(String md) { + var changed = this._participantInfo?.metadata != md; + this.metadata = md; + if (changed) { + delegate?.onMetadataChanged(this); + _roomDelegate?.onMetadataChanged(this); + } + } + + _updateInfo(ParticipantInfo info) { + this.identity = info.identity; + this.sid = info.sid; + if (info.metadata.isNotEmpty) { + _setMetadata(info.metadata); + } + this._participantInfo = info; + } + + _addTrackPublication(TrackPublication pub) { + pub.track?.sid = pub.sid; + tracks[pub.sid] = pub; + switch (pub.kind) { + case TrackType.AUDIO: + audioTracks[pub.sid] = pub; + break; + case TrackType.VIDEO: + videoTracks[pub.sid] = pub; + break; + default: + // nothing + } + } +} diff --git a/lib/src/participant/remote_participant.dart b/lib/src/participant/remote_participant.dart new file mode 100644 index 0000000..eaa60dc --- /dev/null +++ b/lib/src/participant/remote_participant.dart @@ -0,0 +1,3 @@ +import 'participant.dart'; + +class RemoteParticipant extends Participant {} diff --git a/lib/src/room.dart b/lib/src/room.dart new file mode 100644 index 0000000..4482626 --- /dev/null +++ b/lib/src/room.dart @@ -0,0 +1,78 @@ +import 'dart:async'; +import 'package:flutter_webrtc/flutter_webrtc.dart'; + +import 'extensions.dart'; +import 'logger.dart'; +import 'participant/local_participant.dart'; +import 'participant/participant.dart'; +import 'participant/remote_participant.dart'; +import 'proto/livekit_models.pb.dart'; +import 'proto/livekit_rtc.pb.dart'; +import 'rtc_engine.dart'; +import 'signal_client.dart'; + +enum RoomState { + Disconnected, + Connected, + Reconnecting, +} + +class Room { + RoomState state = RoomState.Disconnected; + + /// map of SID to RemoteParticipant + Map participants = {}; + + /// the current participant + late LocalParticipant localParticipant; + + /// name of the room + late String name; + + /// sid of the room + late String sid; + + /// a list of participants that are actively speaking, including local participant. + List activeSpeakers = []; + + RTCEngine _engine; + + Completer? _connectCompleter; + + Room(SignalClient client, RTCConfiguration? rtcConfig) + : _engine = new RTCEngine(client, rtcConfig) { + _engine.onTrack = _onTrackAdded; + _engine.onDisconnected = _handleDisconnect; + _engine.onParticipantUpdateCallback = _handleParticipantUpdate; + _engine.onActiveSpeakerchangedCallback = _handleSpeakerUpdate; + _engine.onDataMessageCallback = _handleDataPacket; + + // TODO: handle reconnecting & reconnected events + } + + Future _connect(String url, String token, JoinOptions? opts) async { + var completer = new Completer(); + _connectCompleter = completer; + + var joinResponse = await _engine.join(url, token, opts); + logger.fine( + 'connected to LiveKit server, version: ${joinResponse.serverVersion}'); + + state = RoomState.Connected; + var pi = joinResponse.participant; + localParticipant = new LocalParticipant(pi.sid, pi.identity, _engine); + + return completer.future; + } + + _handleDisconnect() {} + + _handleParticipantUpdate(List participants) {} + + _handleSpeakerUpdate(List speakers) {} + + _handleDataPacket(UserPacket packet, DataPacket_Kind kind) {} + + _onTrackAdded( + MediaStreamTrack track, MediaStream? stream, RTCRtpReceiver? receiver) {} +} diff --git a/lib/src/rtc_engine.dart b/lib/src/rtc_engine.dart index 5a80a6b..9e3e456 100644 --- a/lib/src/rtc_engine.dart +++ b/lib/src/rtc_engine.dart @@ -1,11 +1,13 @@ 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'; + +import 'errors.dart'; +import 'extensions.dart'; +import 'proto/livekit_rtc.pb.dart'; +import 'proto/livekit_models.pb.dart'; +import 'signal_client.dart'; +import 'track/track.dart'; +import 'transport.dart'; const lossyDataChannel = '_lossy'; const reliableDataChannel = '_reliable'; @@ -18,13 +20,15 @@ typedef ParticipantUpdateCallback = void Function( List participants); typedef ActiveSpeakerChangedCallback = void Function( List speakers); +typedef DataPacketCallback = void Function( + UserPacket packet, DataPacket_Kind kind); class RTCEngine with SignalClientDelegate { PCTransport? publisher; PCTransport? subscriber; SignalClient client; // config for RTCPeerConnection - Map rtcConfig = {}; + RTCConfiguration rtcConfig = new RTCConfiguration(); // data channels for packets RTCDataChannel? reliableDC; RTCDataChannel? lossyDC; @@ -42,9 +46,10 @@ class RTCEngine with SignalClientDelegate { TrackCallback? onTrack; ParticipantUpdateCallback? onParticipantUpdateCallback; ActiveSpeakerChangedCallback? onActiveSpeakerchangedCallback; + DataPacketCallback? onDataMessageCallback; GenericCallback? onDisconnected; - RTCEngine(this.client, Map? rtcConfig) { + RTCEngine(this.client, RTCConfiguration? rtcConfig) { if (rtcConfig != null) { this.rtcConfig = rtcConfig; } @@ -133,9 +138,9 @@ class RTCEngine with SignalClientDelegate { return; } - var pubPC = await createPeerConnection(rtcConfig); + var pubPC = await createPeerConnection(rtcConfig.toMap()); publisher = new PCTransport(pubPC); - var subPC = await createPeerConnection(rtcConfig); + var subPC = await createPeerConnection(rtcConfig.toMap()); subscriber = new PCTransport(subPC); pubPC.onIceCandidate = (RTCIceCandidate candidate) { @@ -194,7 +199,24 @@ class RTCEngine with SignalClientDelegate { reliableDC?.onMessage = _handleDataMessage; } - _handleDataMessage(RTCDataChannelMessage message) {} + _handleDataMessage(RTCDataChannelMessage message) { + // always expect binary + if (!message.isBinary) { + return; + } + + var dp = DataPacket.fromBuffer(message.binary); + switch (dp.whichValue()) { + case DataPacket_Value.speaker: + onActiveSpeakerchangedCallback?.call(dp.speaker.speakers); + break; + case DataPacket_Value.user: + onDataMessageCallback?.call(dp.user, dp.kind); + break; + default: + // do nothing + } + } _handleDisconnect(String reason) { // TODO: implement method @@ -206,21 +228,19 @@ class RTCEngine with SignalClientDelegate { // create peer connections this.isClosed = false; - if (rtcConfig['iceServers'] == null && response.iceServers.length > 0) { - var iceServers = []; + if (rtcConfig.iceServers == null && response.iceServers.length > 0) { + List iceServers = []; response.iceServers.forEach((item) { - Map iceServer = { - 'urls': item.urls, - }; + var iceServer = new RTCIceServer(urls: item.urls); if (item.username.isNotEmpty) { - iceServer['username'] = item.username; + iceServer.username = item.username; } if (item.credential.isNotEmpty) { - iceServer['credential'] = item.credential; + iceServer.credential = item.credential; } iceServers.add(iceServer); }); - rtcConfig['iceServers'] = iceServers; + rtcConfig.iceServers = iceServers; } _configurePeerConnections(); diff --git a/lib/src/signal_client.dart b/lib/src/signal_client.dart index c5e1bed..ab3594e 100644 --- a/lib/src/signal_client.dart +++ b/lib/src/signal_client.dart @@ -174,7 +174,7 @@ class SignalClient { delegate.onOffer(toRTCSessionDescription(msg.offer)); break; case SignalResponse_Message.trickle: - delegate.onTrickle(toRTCIceCandidate(msg.trickle), msg.trickle.target); + delegate.onTrickle(toRTCIceCandidate(msg.trickle.candidateInit), msg.trickle.target); break; case SignalResponse_Message.update: delegate.onParticipantUpdate(msg.update.participants); diff --git a/lib/src/track/track.dart b/lib/src/track/track.dart index f460acb..7357de4 100644 --- a/lib/src/track/track.dart +++ b/lib/src/track/track.dart @@ -4,3 +4,7 @@ class TrackDimension { TrackDimension(this.width, this.height); } + +class Track { + String? sid; +} diff --git a/lib/src/track/track_publication.dart b/lib/src/track/track_publication.dart new file mode 100644 index 0000000..a498cb0 --- /dev/null +++ b/lib/src/track/track_publication.dart @@ -0,0 +1,31 @@ +import '../proto/livekit_models.pb.dart'; +import 'track.dart'; + +class TrackPublication { + Track? track; + String name; + String sid; + TrackType kind; + bool muted = false; + bool simulcasted = false; + TrackDimension? dimension; + + bool get isSubscribed => track != null; + + TrackPublication({required this.sid, required this.name, required this.kind}); + + TrackPublication.fromInfo(TrackInfo info) + : sid = info.sid, + name = info.name, + kind = info.type { + _updateFromInfo(info); + } + + _updateFromInfo(TrackInfo info) { + muted = info.muted; + simulcasted = info.simulcast; + if (info.type == TrackType.VIDEO) { + dimension = new TrackDimension(info.width, info.height); + } + } +} diff --git a/pubspec.lock b/pubspec.lock index bf386e0..481fbd5 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -95,6 +95,13 @@ packages: url: "https://pub.dartlang.org" source: hosted version: "0.6.5" + logging: + dependency: "direct main" + description: + name: logging + url: "https://pub.dartlang.org" + source: hosted + version: "1.0.1" matcher: dependency: transitive description: diff --git a/pubspec.yaml b/pubspec.yaml index 02a56a9..ef69574 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -11,6 +11,7 @@ dependencies: flutter: sdk: flutter flutter_webrtc: ^0.6.4 + logging: ^1.0.1 protobuf: ^2.0.0 web_socket_channel: ^2.1.0