livekit_datatrack/local/
mod.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
15use crate::{
16    api::{DataTrack, DataTrackFrame, DataTrackInfo, DataTrackSchemaError, InternalError},
17    schema::{DataTrackFrameEncoding, DataTrackSchemaId},
18    track::DataTrackInner,
19};
20use std::{fmt, marker::PhantomData, sync::Arc};
21use thiserror::Error;
22use tokio::sync::{mpsc, watch};
23
24pub(crate) mod events;
25pub(crate) mod manager;
26pub(crate) mod proto;
27
28mod packetizer;
29mod pipeline;
30
31/// Data track published by the local participant.
32pub type LocalDataTrack = DataTrack<Local>;
33
34/// Marker type indicating a [`DataTrack`] belongs to the local participant.
35///
36/// See also: [`LocalDataTrack`]
37///
38#[derive(Debug, Clone)]
39pub struct Local;
40
41impl DataTrack<Local> {
42    pub(crate) fn new(info: Arc<DataTrackInfo>, inner: LocalTrackInner) -> Self {
43        Self { info, inner: Arc::new(inner.into()), _location: PhantomData }
44    }
45
46    fn inner(&self) -> &LocalTrackInner {
47        match &*self.inner {
48            DataTrackInner::Local(track) => track,
49            DataTrackInner::Remote(_) => unreachable!(), // Safe (type state)
50        }
51    }
52}
53
54impl DataTrack<Local> {
55    /// Try pushing a frame to subscribers of the track.
56    ///
57    /// # Example
58    ///
59    /// ```
60    /// # use livekit_datatrack::api::{LocalDataTrack, DataTrackFrame, PushFrameError};
61    /// # fn example(track: LocalDataTrack) -> Result<(), PushFrameError> {
62    /// fn read_sensor() -> Vec<u8> {
63    ///     // Read some sensor data...
64    ///     vec![0xFA; 16]
65    /// }
66    ///
67    /// let frame = read_sensor().into(); // Convert to frame
68    /// track.try_push(frame)?;
69    ///
70    /// # Ok(())
71    /// # }
72    /// ```
73    ///
74    /// See [`DataTrackFrame`] for more ways to construct a frame and how to attach metadata.
75    ///
76    /// # Errors
77    ///
78    /// Pushing a frame can fail for several reasons:
79    ///
80    /// - The track has been unpublished by the local participant or SFU
81    /// - The room is no longer connected
82    /// - Frames are being pushed too fast
83    ///
84    pub fn try_push(&self, frame: DataTrackFrame) -> Result<(), PushFrameError> {
85        match self.inner().publish_state() {
86            manager::PublishState::Republishing => {
87                return Err(PushFrameError::new(frame, PushFrameErrorReason::QueueFull))?
88            }
89            manager::PublishState::Unpublished => {
90                return Err(PushFrameError::new(frame, PushFrameErrorReason::TrackUnpublished))?;
91            }
92            manager::PublishState::Published => {}
93        }
94        self.inner()
95            .frame_tx
96            .try_send(frame)
97            .map_err(|err| PushFrameError::new(err.into_inner(), PushFrameErrorReason::QueueFull))
98    }
99
100    /// Unpublishes the track.
101    pub fn unpublish(&self) {
102        self.inner().local_unpublish();
103    }
104}
105
106#[derive(Debug, Clone)]
107pub(crate) struct LocalTrackInner {
108    pub frame_tx: mpsc::Sender<DataTrackFrame>,
109    pub state_tx: watch::Sender<manager::PublishState>,
110}
111
112impl LocalTrackInner {
113    fn publish_state(&self) -> manager::PublishState {
114        *self.state_tx.borrow()
115    }
116
117    pub(crate) fn is_published(&self) -> bool {
118        // Note: a track which is internally in the "resubscribing" state
119        // is still considered published from the public API perspective.
120        self.publish_state() != manager::PublishState::Unpublished
121    }
122
123    pub(crate) async fn wait_for_unpublish(&self) {
124        _ = self
125            .state_tx
126            .subscribe()
127            .wait_for(|state| *state == manager::PublishState::Unpublished)
128            .await
129    }
130
131    fn local_unpublish(&self) {
132        _ = self.state_tx.send(manager::PublishState::Unpublished);
133    }
134}
135
136impl Drop for LocalTrackInner {
137    fn drop(&mut self) {
138        // Implicit unpublish when handle dropped.
139        self.local_unpublish();
140    }
141}
142
143/// Options for publishing a data track.
144///
145/// # Examples
146///
147/// Create options for publishing a track named "my_track":
148///
149/// ```
150/// # use livekit_datatrack::api::DataTrackOptions;
151/// let options = DataTrackOptions::new("my_track");
152/// ```
153///
154#[derive(Clone, Debug)]
155pub struct DataTrackOptions {
156    pub(crate) name: String,
157    pub(crate) schema: Option<DataTrackSchemaId>,
158    pub(crate) frame_encoding: Option<DataTrackFrameEncoding>,
159}
160
161impl DataTrackOptions {
162    /// Creates options with the given track name.
163    ///
164    /// The track name is used to identify the track to other participants.
165    ///
166    /// # Requirements
167    /// - Must not be empty
168    /// - Must be unique per publisher
169    ///
170    pub fn new(name: impl Into<String>) -> Self {
171        Self { name: name.into(), schema: None, frame_encoding: None }
172    }
173
174    /// Sets the schema associated with frames sent on the track.
175    pub fn with_schema(self, schema: DataTrackSchemaId) -> Self {
176        Self { schema: Some(schema), ..self }
177    }
178
179    /// Sets the encoding of frames sent on the track.
180    pub fn with_frame_encoding(self, encoding: DataTrackFrameEncoding) -> Self {
181        Self { frame_encoding: Some(encoding), ..self }
182    }
183}
184
185impl From<String> for DataTrackOptions {
186    fn from(name: String) -> Self {
187        Self::new(name)
188    }
189}
190
191impl From<&str> for DataTrackOptions {
192    fn from(name: &str) -> Self {
193        Self::new(name.to_string())
194    }
195}
196
197/// An error that can occur when publishing a data track.
198#[derive(Debug, Error)]
199#[cfg_attr(feature = "uniffi", derive(uniffi::Error))]
200#[cfg_attr(feature = "uniffi", uniffi(flat_error))]
201pub enum PublishError {
202    /// Local participant does not have permission to publish data tracks.
203    ///
204    /// Ensure the participant's token contains the `canPublishData` grant.
205    ///
206    #[error("Data track publishing unauthorized")]
207    NotAllowed,
208
209    /// A track with the same name is already published by the local participant.
210    #[error("Track name already taken")]
211    DuplicateName,
212
213    /// The track name is invalid.
214    ///
215    /// This occurs when the name is empty or exceeds the allowed maximum length.
216    ///
217    #[error("Track name invalid")]
218    InvalidName,
219
220    /// Request to publish the track took long to complete.
221    #[error("Publish data track timed-out")]
222    Timeout,
223
224    /// No additional data tracks can be published by the local participant.
225    #[error("Data track publication limit reached")]
226    LimitReached,
227
228    /// Cannot publish data track when the room is disconnected.
229    #[error("Room disconnected")]
230    Disconnected,
231
232    /// Schema metadata is invalid.
233    #[error(transparent)]
234    InvalidSchema(DataTrackSchemaError),
235
236    /// Internal error, please report on GitHub.
237    #[error(transparent)]
238    Internal(#[from] InternalError),
239}
240
241/// Frame could not be pushed to a data track.
242#[derive(Debug, Error)]
243#[error("Failed to publish frame: {reason}")]
244pub struct PushFrameError {
245    frame: DataTrackFrame,
246    reason: PushFrameErrorReason,
247}
248
249impl PushFrameError {
250    pub(crate) fn new(frame: DataTrackFrame, reason: PushFrameErrorReason) -> Self {
251        Self { frame, reason }
252    }
253
254    /// Returns the reason the frame could not be pushed.
255    pub fn reason(&self) -> PushFrameErrorReason {
256        self.reason
257    }
258
259    /// Consumes the error and returns the frame that couldn't be pushed.
260    ///
261    /// This may be useful for implementing application-specific retry logic.
262    ///
263    pub fn into_frame(self) -> DataTrackFrame {
264        self.frame
265    }
266}
267
268/// Reason why a data track frame could not be pushed.
269#[derive(Debug, Clone, Copy)]
270#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
271pub enum PushFrameErrorReason {
272    /// Track is no longer published.
273    TrackUnpublished,
274    /// Frame was dropped due to the pipeline queue being full.
275    QueueFull,
276}
277
278impl fmt::Display for PushFrameErrorReason {
279    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
280        match self {
281            Self::TrackUnpublished => write!(f, "track unpublished"),
282            Self::QueueFull => write!(f, "queue full"),
283        }
284    }
285}
livekit_datatrack/local/mod.rs

livekit_datatrack/local/
mod.rs
