livekit_datatrack/remote/
proto.rs

1// Copyright 2025 LiveKit, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Conversions between [`super::events`] and [`livekit_protocol`] wire types.
16//!
17//! Where there is a one-to-one mapping between proto message and event, a `From`
18//! or `TryFrom` implementation is defined. Otherwise, a helper function extracts
19//! the event from a larger composite proto message.
20
21use super::events::*;
22use crate::{
23    api::{DataTrackInfo, DataTrackSid, InternalError},
24    packet::Handle,
25};
26use livekit_protocol as proto;
27use std::{collections::HashMap, mem};
28
29// MARK: - Protocol -> input event
30
31impl TryFrom<proto::DataTrackSubscriberHandles> for SfuSubscriberHandles {
32    type Error = InternalError;
33
34    fn try_from(msg: proto::DataTrackSubscriberHandles) -> Result<Self, Self::Error> {
35        let mapping = msg
36            .sub_handles
37            .into_iter()
38            .map(|(handle, info)| -> Result<_, InternalError> {
39                let handle: Handle = handle.try_into().map_err(anyhow::Error::from)?;
40                let sid: DataTrackSid = info.track_sid.try_into().map_err(anyhow::Error::from)?;
41                Ok((handle, sid))
42            })
43            .collect::<Result<HashMap<Handle, DataTrackSid>, _>>()?;
44        Ok(SfuSubscriberHandles { mapping })
45    }
46}
47
48/// Extracts an [`SfuPublicationUpdates`] event from a join response.
49///
50/// This takes ownership of the `data_tracks` vector for each participant
51/// (except for the local participant), leaving an empty vector in its place.
52///
53pub fn event_from_join(
54    msg: &mut proto::JoinResponse,
55) -> Result<SfuPublicationUpdates, InternalError> {
56    event_from_participant_info(&mut msg.other_participants, None)
57}
58
59/// Extracts an [`SfuPublicationUpdates`] event from a participant update.
60///
61/// This takes ownership of the `data_tracks` vector for each participant in
62/// the update, leaving an empty vector in its place.
63///
64pub fn event_from_participant_update(
65    msg: &mut proto::ParticipantUpdate,
66    local_participant_identity: &str,
67) -> Result<SfuPublicationUpdates, InternalError> {
68    // TODO: is there a better way to exclude the local participant?
69    event_from_participant_info(&mut msg.participants, local_participant_identity.into())
70}
71
72fn event_from_participant_info(
73    msg: &mut [proto::ParticipantInfo],
74    local_participant_identity: Option<&str>,
75) -> Result<SfuPublicationUpdates, InternalError> {
76    let updates = msg
77        .iter_mut()
78        .filter(|participant| {
79            local_participant_identity.is_none_or(|identity| participant.identity != identity)
80        })
81        .map(|participant| -> Result<_, InternalError> {
82            Ok((participant.identity.clone(), extract_track_info(participant)?))
83        })
84        .collect::<Result<HashMap<String, Vec<DataTrackInfo>>, _>>()?;
85    Ok(SfuPublicationUpdates { updates })
86}
87
88fn extract_track_info(
89    msg: &mut proto::ParticipantInfo,
90) -> Result<Vec<DataTrackInfo>, InternalError> {
91    mem::take(&mut msg.data_tracks)
92        .into_iter()
93        .map(TryInto::<DataTrackInfo>::try_into)
94        .collect::<Result<Vec<_>, InternalError>>()
95}
96
97// MARK: - Output event -> protocol
98
99impl From<SfuUpdateSubscription> for proto::UpdateDataSubscription {
100    fn from(event: SfuUpdateSubscription) -> Self {
101        let update = proto::update_data_subscription::Update {
102            track_sid: event.sid.into(),
103            subscribe: event.subscribe,
104            options: Default::default(),
105        };
106        Self { updates: vec![update] }
107    }
108}
109
110#[cfg(test)]
111mod tests {
112    use super::*;
113
114    #[test]
115    fn test_from_subscriber_handles() {
116        let sub_handles = [
117            (
118                1,
119                proto::data_track_subscriber_handles::PublishedDataTrack {
120                    track_sid: "DTR_1234".into(),
121                    ..Default::default()
122                },
123            ),
124            (
125                2,
126                proto::data_track_subscriber_handles::PublishedDataTrack {
127                    track_sid: "DTR_4567".into(),
128                    ..Default::default()
129                },
130            ),
131        ];
132        let subscriber_handles =
133            proto::DataTrackSubscriberHandles { sub_handles: HashMap::from(sub_handles) };
134
135        let event: SfuSubscriberHandles = subscriber_handles.try_into().unwrap();
136        assert_eq!(
137            event.mapping.get(&1u32.try_into().unwrap()).unwrap(),
138            &"DTR_1234".to_string().try_into().unwrap()
139        );
140        assert_eq!(
141            event.mapping.get(&2u32.try_into().unwrap()).unwrap(),
142            &"DTR_4567".to_string().try_into().unwrap()
143        );
144    }
145
146    #[test]
147    fn test_extract_track_info() {
148        let data_tracks = vec![proto::DataTrackInfo {
149            pub_handle: 1,
150            sid: "DTR_1234".into(),
151            name: "track1".into(),
152            encryption: proto::encryption::Type::Gcm.into(),
153            schema: None,
154            frame_encoding: None,
155        }];
156        let mut participant_info = proto::ParticipantInfo { data_tracks, ..Default::default() };
157
158        let track_info = extract_track_info(&mut participant_info).unwrap();
159        assert!(participant_info.data_tracks.is_empty(), "Expected original vec taken");
160        assert_eq!(track_info.len(), 1);
161
162        let first = track_info.first().unwrap();
163        assert_eq!(first.pub_handle, 1u32.try_into().unwrap());
164        assert_eq!(first.name, "track1");
165        assert_eq!(*first.sid.read().unwrap(), "DTR_1234".to_string().try_into().unwrap());
166    }
167}
livekit_datatrack/remote/proto.rs

livekit_datatrack/remote/
proto.rs
