Data tracks core

Cargo.toml

@@ -55,7 +55,9 @@ bytes = "1.10"
 clap = "4.5"
 console-subscriber = "0.1"
 env_logger = "0.11"
+from_variants = "1.0.2"
 futures = "0.3"
+futures-core = "0.3"
 futures-util = "0.3"
 lazy_static = "1.4"
 log = "0.4"
@@ -68,6 +70,7 @@ serde = "1"
 serde_json = "1.0"
 thiserror = "1"
 tokio = "1"
+tokio-stream = "0.1"
 
 [profile.release]
 opt-level = "z"

livekit-datatrack/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "livekit-datatrack"
+version = "0.1.0"
+edition = "2021"
+license = "Apache-2.0"
+repository = "https://github.com/livekit/rust-sdks"
+readme = "README.md"
+
+[dependencies]
+livekit-protocol = { workspace = true }
+livekit-runtime = { workspace = true }
+log = { workspace = true }
+thiserror = "2.0.17"  # TODO: upgrade workspace version to 2.x.x
+tokio = { workspace = true, default-features = false, features = ["sync"] }
+futures-util = { workspace = true, default-features = false, features = ["sink"] }
+futures-core = { workspace = true }
+bytes = { workspace = true }
+from_variants = { workspace = true }
+tokio-stream = { workspace = true, features = ["sync"] }
+anyhow = { workspace = true } # For internal error handling only
+rand = { workspace = true }
+
+[dev-dependencies]
+test-case = "3.3"
+fake = { version = "4.4", features = ["derive"] }

livekit-datatrack/README.md

@@ -0,0 +1,6 @@
+# LiveKit Data Track
+
+**Important**:
+This is an internal crate that powers the data tracks feature in LiveKit client SDKs (including [Rust](https://crates.io/crates/livekit) and others) and is not usable directly.
+
+To use data tracks in your application, please use the public APIs provided by the client SDKs.

livekit-datatrack/src/e2ee.rs

@@ -0,0 +1,57 @@
+// Copyright 2025 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use bytes::Bytes;
+use core::fmt::Debug;
+use thiserror::Error;
+
+// TODO: If a core module for end-to-end encryption is created in the future
+// (livekit-e2ee), these traits should be moved to there.
+
+/// Encrypted payload and metadata required for decryption.
+pub struct EncryptedPayload {
+    pub payload: Bytes,
+    pub iv: [u8; 12],
+    pub key_index: u8,
+}
+
+/// An error indicating a payload could not be encrypted.
+#[derive(Debug, Error)]
+#[error("Encryption failed")]
+pub struct EncryptionError;
+
+/// An error indicating a payload could not be decrypted.
+#[derive(Debug, Error)]
+#[error("Decryption failed")]
+pub struct DecryptionError;
+
+/// Provider for encrypting payloads for E2EE.
+pub trait EncryptionProvider: Send + Sync + Debug {
+    /// Encrypts the given payload being sent by the local participant.
+    fn encrypt(&self, payload: Bytes) -> Result<EncryptedPayload, EncryptionError>;
+}
+
+/// Provider for decrypting payloads for E2EE.
+pub trait DecryptionProvider: Send + Sync + Debug {
+    /// Decrypts the given payload received from a remote participant.
+    ///
+    /// Sender identity is required in order for the proper key to be used
+    /// for decryption.
+    ///
+    fn decrypt(
+        &self,
+        payload: EncryptedPayload,
+        sender_identity: &str,
+    ) -> Result<Bytes, DecryptionError>;
+}

livekit-datatrack/src/error.rs

@@ -0,0 +1,24 @@
+// Copyright 2025 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use thiserror::Error;
+
+/// Internal data track error.
+///
+/// Occurrences of this error type are unexpected and likely indicate
+/// a bug. If encountered, please report on GitHub and include the full error description.
+///
+#[derive(Debug, Error)]
+#[error(transparent)]
+pub struct InternalError(#[from] anyhow::Error);

livekit-datatrack/src/frame.rs

@@ -0,0 +1,122 @@
+// Copyright 2025 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use bytes::Bytes;
+use core::fmt;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+/// A frame published on a data track, consisting of a payload and optional metadata.
+///
+/// # Examples
+///
+/// Create a frame from a [`Vec`] payload:
+///
+/// ```
+/// # use livekit_datatrack::api::DataTrackFrame;
+/// let some_payload = vec![0xFA; 256];
+/// let frame: DataTrackFrame = some_payload.into();
+///
+/// assert_eq!(frame.payload().len(), 256);
+/// ```
+///
+#[derive(Clone, Default)]
+pub struct DataTrackFrame {
+    pub(crate) payload: Bytes,
+    pub(crate) user_timestamp: Option<u64>,
+}
+
+impl DataTrackFrame {
+    /// Returns the frame's payload.
+    pub fn payload(&self) -> Bytes {
+        self.payload.clone() // Cheap clone
+    }
+
+    /// Returns the frame's user timestamp, if one is associated.
+    pub fn user_timestamp(&self) -> Option<u64> {
+        self.user_timestamp
+    }
+
+    /// If the frame has a user timestamp, calculate how long has passed
+    /// relative to the current system time.
+    ///
+    /// If a timestamp is present, it is assumed it is a UNIX timestamp in milliseconds
+    /// (as can be set with [`Self::with_user_timestamp_now`] on the publisher side).
+    /// If the timestamp is invalid or not present, the result is none.
+    ///
+    pub fn duration_since_timestamp(&self) -> Option<Duration> {
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH + Duration::from_millis(self.user_timestamp?))
+            .inspect_err(|err| log::error!("Failed to calculate duration: {err}"))
+            .ok()
+    }
+}
+
+impl DataTrackFrame {
+    /// Creates a frame from the given payload.
+    pub fn new(payload: impl Into<Bytes>) -> Self {
+        Self { payload: payload.into(), ..Default::default() }
+    }
+
+    /// Associates a user timestamp with the frame.
+    pub fn with_user_timestamp(mut self, value: u64) -> Self {
+        self.user_timestamp = Some(value);
+        self
+    }
+
+    /// Associates the current Unix timestamp (in milliseconds) with the frame.
+    pub fn with_user_timestamp_now(mut self) -> Self {
+        let timestamp = SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_millis() as u64)
+            .inspect_err(|err| log::error!("Failed to get system time: {err}"))
+            .ok();
+        self.user_timestamp = timestamp;
+        self
+    }
+}
+
+impl fmt::Debug for DataTrackFrame {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("DataTrackFrame")
+            .field("payload_len", &self.payload.len())
+            .field("user_timestamp", &self.user_timestamp)
+            .finish()
+    }
+}
+
+// MARK: - From implementations
+
+impl From<Bytes> for DataTrackFrame {
+    fn from(bytes: Bytes) -> Self {
+        Self { payload: bytes, ..Default::default() }
+    }
+}
+
+impl From<&'static [u8]> for DataTrackFrame {
+    fn from(slice: &'static [u8]) -> Self {
+        Self { payload: slice.into(), ..Default::default() }
+    }
+}
+
+impl From<Vec<u8>> for DataTrackFrame {
+    fn from(vec: Vec<u8>) -> Self {
+        Self { payload: vec.into(), ..Default::default() }
+    }
+}
+
+impl From<Box<[u8]>> for DataTrackFrame {
+    fn from(slice: Box<[u8]>) -> Self {
+        Self { payload: slice.into(), ..Default::default() }
+    }
+}