livekit_datatrack/local/
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, PublishError},
24    packet::Handle,
25};
26use anyhow::{anyhow, Context};
27use livekit_protocol as proto;
28use std::{borrow::Borrow, sync::RwLock};
29
30// MARK: - Output event -> protocol
31
32impl From<SfuPublishRequest> for proto::PublishDataTrackRequest {
33    fn from(event: SfuPublishRequest) -> Self {
34        use proto::encryption::Type;
35        let encryption = if event.uses_e2ee { Type::Gcm } else { Type::None }.into();
36        let schema = event.schema.map(Into::into);
37        let frame_encoding = event.frame_encoding.map(Into::into);
38        Self {
39            pub_handle: event.handle.into(),
40            name: event.name,
41            encryption,
42            schema,
43            frame_encoding,
44        }
45    }
46}
47
48impl From<SfuUnpublishRequest> for proto::UnpublishDataTrackRequest {
49    fn from(event: SfuUnpublishRequest) -> Self {
50        Self { pub_handle: event.handle.into() }
51    }
52}
53
54// MARK: - Protocol -> input event
55
56impl TryFrom<proto::PublishDataTrackResponse> for SfuPublishResponse {
57    type Error = InternalError;
58
59    fn try_from(msg: proto::PublishDataTrackResponse) -> Result<Self, Self::Error> {
60        let info: DataTrackInfo = msg.info.context("Missing info")?.try_into()?;
61        Ok(Self { handle: info.pub_handle, result: Ok(info) })
62    }
63}
64
65impl TryFrom<proto::UnpublishDataTrackResponse> for SfuUnpublishResponse {
66    type Error = InternalError;
67
68    fn try_from(msg: proto::UnpublishDataTrackResponse) -> Result<Self, Self::Error> {
69        let handle: Handle =
70            msg.info.context("Missing info")?.pub_handle.try_into().map_err(anyhow::Error::from)?;
71        Ok(Self { handle })
72    }
73}
74
75impl TryFrom<proto::DataTrackInfo> for DataTrackInfo {
76    type Error = InternalError;
77
78    fn try_from(msg: proto::DataTrackInfo) -> Result<Self, Self::Error> {
79        let handle: Handle = msg.pub_handle.try_into().map_err(anyhow::Error::from)?;
80        let uses_e2ee = match msg.encryption() {
81            proto::encryption::Type::None => false,
82            proto::encryption::Type::Gcm => true,
83            other => Err(anyhow!("Unsupported E2EE type: {:?}", other))?,
84        };
85        let frame_encoding = msg.frame_encoding.map(Into::into);
86        let sid: DataTrackSid = msg.sid.try_into().map_err(anyhow::Error::from)?;
87        let schema = msg.schema.map(|schema| schema.into());
88
89        Ok(Self {
90            pub_handle: handle,
91            sid: RwLock::new(sid).into(),
92            name: msg.name,
93            uses_e2ee,
94            schema,
95            frame_encoding,
96        })
97    }
98}
99
100pub fn publish_result_from_request_response(
101    msg: &proto::RequestResponse,
102) -> Option<SfuPublishResponse> {
103    use proto::request_response::{Reason, Request};
104    let Some(request) = &msg.request else { return None };
105    let Request::PublishDataTrack(request) = request else { return None };
106    let Ok(handle) = TryInto::<Handle>::try_into(request.pub_handle) else { return None };
107    let error = match msg.reason() {
108        // If new error reasons are introduced in the future, consider adding them
109        // to the public error enum if they are useful to the user.
110        Reason::NotAllowed => PublishError::NotAllowed,
111        Reason::DuplicateName => PublishError::DuplicateName,
112        Reason::InvalidName => PublishError::InvalidName,
113        _ => PublishError::Internal(anyhow!("SFU rejected: {}", msg.message).into()),
114    };
115    let event = SfuPublishResponse { handle, result: Err(error) };
116    Some(event)
117}
118
119// MARK: - Sync state support
120
121impl From<DataTrackInfo> for proto::DataTrackInfo {
122    fn from(info: DataTrackInfo) -> Self {
123        let encryption = if info.uses_e2ee() {
124            proto::encryption::Type::Gcm
125        } else {
126            proto::encryption::Type::None
127        } as i32;
128        let sid = info.sid().to_string();
129        let schema = info.schema.map(|schema| schema.into());
130        let frame_encoding = info.frame_encoding.map(Into::into);
131        Self {
132            pub_handle: info.pub_handle.into(),
133            sid,
134            name: info.name,
135            encryption,
136            schema,
137            frame_encoding,
138        }
139    }
140}
141
142/// Form publish responses for each publish data track to support sync state.
143pub fn publish_responses_for_sync_state(
144    published_tracks: impl IntoIterator<Item = impl Borrow<DataTrackInfo>>,
145) -> Vec<proto::PublishDataTrackResponse> {
146    published_tracks
147        .into_iter()
148        .map(|info| proto::PublishDataTrackResponse { info: Some(info.borrow().clone().into()) })
149        .collect()
150}
151
152#[cfg(test)]
153mod tests {
154    use crate::schema::{DataTrackFrameEncoding, DataTrackSchemaEncoding, DataTrackSchemaId};
155
156    use super::*;
157    use fake::{Fake, Faker};
158
159    #[test]
160    fn test_from_publish_request_event() {
161        let event = SfuPublishRequest {
162            handle: 1u32.try_into().unwrap(),
163            name: "track".into(),
164            uses_e2ee: true,
165            schema: None,
166            frame_encoding: None,
167        };
168        let request: proto::PublishDataTrackRequest = event.into();
169        assert_eq!(request.pub_handle, 1);
170        assert_eq!(request.name, "track");
171        assert_eq!(request.encryption(), proto::encryption::Type::Gcm);
172    }
173
174    #[test]
175    fn test_from_unpublish_request_event() {
176        let event = SfuUnpublishRequest { handle: 1u32.try_into().unwrap() };
177        let request: proto::UnpublishDataTrackRequest = event.into();
178        assert_eq!(request.pub_handle, 1);
179    }
180
181    #[test]
182    fn test_from_publish_response() {
183        let response = proto::PublishDataTrackResponse {
184            info: proto::DataTrackInfo {
185                pub_handle: 1,
186                sid: "DTR_1234".into(),
187                name: "track".into(),
188                encryption: proto::encryption::Type::Gcm.into(),
189                schema: proto::DataTrackSchemaId {
190                    name: "schema".into(),
191                    encoding: Some(DataTrackSchemaEncoding::JsonSchema.into()),
192                }
193                .into(),
194                frame_encoding: Some(DataTrackFrameEncoding::Json.into()),
195            }
196            .into(),
197        };
198        let event: SfuPublishResponse = response.try_into().unwrap();
199        assert_eq!(event.handle, 1u32.try_into().unwrap());
200
201        let info = event.result.expect("Expected ok result");
202        assert_eq!(info.pub_handle, 1u32.try_into().unwrap());
203        assert_eq!(*info.sid.read().unwrap(), "DTR_1234".to_string().try_into().unwrap());
204        assert_eq!(info.name, "track");
205        assert_eq!(
206            info.schema,
207            Some(DataTrackSchemaId::new("schema", DataTrackSchemaEncoding::JsonSchema))
208        );
209        assert_eq!(info.frame_encoding, Some(DataTrackFrameEncoding::Json));
210        assert!(info.uses_e2ee);
211    }
212
213    #[test]
214    fn test_frame_encoding_mapping() {
215        let base = proto::DataTrackInfo {
216            pub_handle: 1,
217            sid: "DTR_1234".into(),
218            name: "track".into(),
219            encryption: proto::encryption::Type::None.into(),
220            schema: None,
221            frame_encoding: None,
222        };
223
224        let info: DataTrackInfo = base.clone().try_into().unwrap();
225        assert_eq!(info.frame_encoding, None);
226
227        let unspecified = proto::DataTrackInfo {
228            frame_encoding: Some(DataTrackFrameEncoding::Other.into()),
229            ..base.clone()
230        };
231        let info: DataTrackInfo = unspecified.try_into().unwrap();
232        assert_eq!(info.frame_encoding, Some(DataTrackFrameEncoding::Other));
233
234        let custom = proto::DataTrackInfo {
235            frame_encoding: Some(DataTrackFrameEncoding::Custom("my_encoding".into()).into()),
236            ..base
237        };
238        let info: DataTrackInfo = custom.try_into().unwrap();
239        assert_eq!(info.frame_encoding, Some(DataTrackFrameEncoding::Custom("my_encoding".into())));
240    }
241
242    #[test]
243    fn test_from_request_response() {
244        use proto::request_response::{Reason, Request};
245        let response = proto::RequestResponse {
246            request: Request::PublishDataTrack(proto::PublishDataTrackRequest {
247                pub_handle: 1,
248                ..Default::default()
249            })
250            .into(),
251            reason: Reason::NotAllowed.into(),
252            ..Default::default()
253        };
254
255        let event = publish_result_from_request_response(&response).expect("Expected event");
256        assert_eq!(event.handle, 1u32.try_into().unwrap());
257        assert!(matches!(event.result, Err(PublishError::NotAllowed)));
258    }
259
260    #[test]
261    fn test_publish_responses_for_sync_state() {
262        let mut first: DataTrackInfo = Faker.fake();
263        first.uses_e2ee = true;
264
265        let mut second: DataTrackInfo = Faker.fake();
266        second.uses_e2ee = false;
267
268        let publish_responses = publish_responses_for_sync_state(vec![first, second]);
269        assert_eq!(
270            publish_responses[0].info.as_ref().unwrap().encryption(),
271            proto::encryption::Type::Gcm
272        );
273        assert_eq!(
274            publish_responses[1].info.as_ref().unwrap().encryption(),
275            proto::encryption::Type::None
276        );
277    }
278}
livekit_datatrack/local/proto.rs

livekit_datatrack/local/
proto.rs
