Data tracks

Cargo.toml

@@ -8,6 +8,7 @@ members = [
     "livekit-uniffi",
     "livekit-ffi-node-bindings",
     "livekit-runtime",
+    "livekit-datatrack",
     "libwebrtc",
     "soxr-sys",
     "yuv-sys",
@@ -19,6 +20,7 @@ members = [
     "examples/api",
     "examples/basic_room",
     "examples/basic_text_stream",
+    "examples/basic_data_track",
     "examples/encrypted_text_stream",
     "examples/local_audio",
     "examples/local_video",
@@ -40,6 +42,7 @@ livekit-api = { version = "0.4.12", path = "livekit-api" }
 livekit-ffi = { version = "0.12.44", path = "livekit-ffi" }
 livekit-protocol = { version = "0.6.0", path = "livekit-protocol" }
 livekit-runtime = { version = "0.4.0", path = "livekit-runtime" }
+livekit-datatrack = { version = "0.1.0", path = "livekit-datatrack" }
 soxr-sys = { version = "0.1.1", path = "soxr-sys" }
 webrtc-sys = { version = "0.3.21", path = "webrtc-sys" }
 webrtc-sys-build = { version = "0.3.12", path = "webrtc-sys/build" }

examples/basic_data_track/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "basic_data_track"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[dependencies]
+tokio = { version = "1", features = ["full"] }
+livekit = { workspace = true, features = ["rustls-tls-native-roots"] }
+futures-util = { version = "0.3", default-features = false, features = ["sink"] }
+anyhow = "1.0.100"
+log = "0.4.26"
+env_logger = "0.11.7"
+
+[[bin]]
+name = "publisher"
+path = "src/publisher.rs"
+
+[[bin]]
+name = "subscriber"
+path = "src/subscriber.rs"

examples/basic_data_track/README.md

@@ -0,0 +1,21 @@
+# Basic Data Track
+
+Simple example of publishing and subscribing to a data track.
+
+## Usage
+
+1. Run the publisher:
+
+```sh
+export LIVEKIT_URL="..."
+export LIVEKIT_TOKEN="<first participant token>"
+cargo run --bin publisher
+```
+
+2. In a second terminal, run the subscriber:
+
+```sh
+export LIVEKIT_URL="..."
+export LIVEKIT_TOKEN="<second participant token>"
+cargo run --bin subscriber
+```

examples/basic_data_track/src/publisher.rs

@@ -0,0 +1,39 @@
+use anyhow::Result;
+use livekit::prelude::*;
+use std::{env, time::Duration};
+use tokio::{signal, time};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+
+    let url = env::var("LIVEKIT_URL").expect("LIVEKIT_URL is not set");
+    let token = env::var("LIVEKIT_TOKEN").expect("LIVEKIT_TOKEN is not set");
+
+    let (room, _) = Room::connect(&url, &token, RoomOptions::default()).await?;
+
+    let track = room.local_participant().publish_data_track("my_sensor_data").await?;
+
+    tokio::select! {
+        _ = push_frames(track) => {}
+        _ = signal::ctrl_c() => {}
+    }
+    Ok(())
+}
+
+async fn read_sensor() -> Vec<u8> {
+    // Dynamically read some sensor data...
+    vec![0xFA; 256]
+}
+
+async fn push_frames(track: LocalDataTrack) {
+    loop {
+        log::info!("Pushing frame");
+
+        let frame = DataTrackFrame::new(read_sensor().await)
+            .with_user_timestamp_now();
+
+        track.try_push(frame).inspect_err(|err| println!("Failed to push frame: {}", err)).ok();
+        time::sleep(Duration::from_millis(500)).await
+    }
+}

examples/basic_data_track/src/subscriber.rs

@@ -0,0 +1,52 @@
+use anyhow::Result;
+use futures_util::StreamExt;
+use livekit::prelude::*;
+use std::env;
+use tokio::{signal, sync::mpsc::UnboundedReceiver};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+
+    let url = env::var("LIVEKIT_URL").expect("LIVEKIT_URL is not set");
+    let token = env::var("LIVEKIT_TOKEN").expect("LIVEKIT_TOKEN is not set");
+
+    let (_, rx) = Room::connect(&url, &token, RoomOptions::default()).await?;
+
+    tokio::select! {
+        _ = handle_first_publication(rx) => {}
+        _ = signal::ctrl_c() => {}
+    }
+    Ok(())
+}
+
+/// Subscribe to the first data track published.
+async fn handle_first_publication(mut rx: UnboundedReceiver<RoomEvent>) -> Result<()> {
+    log::info!("Waiting for publication…");
+    while let Some(event) = rx.recv().await {
+        let RoomEvent::RemoteDataTrackPublished(track) = event else {
+            continue;
+        };
+        subscribe(track).await?
+    }
+    Ok(())
+}
+
+/// Subscribes to the given data track and logs received frames.
+async fn subscribe(track: RemoteDataTrack) -> Result<()> {
+    log::info!(
+        "Subscribing to '{}' published by '{}'",
+        track.info().name(),
+        track.publisher_identity()
+    );
+    let mut stream = track.subscribe().await?;
+    while let Some(frame) = stream.next().await {
+        log::info!("Received frame ({} bytes)", frame.payload().len());
+
+        if let Some(duration) = frame.duration_since_timestamp() {
+            log::info!("Latency: {:?}", duration);
+        }
+    }
+    log::info!("Unsubscribed");
+    Ok(())
+}

livekit-api/src/services/connector.rs

@@ -164,6 +164,7 @@ impl ConnectorClient {
                 proto::DisconnectWhatsAppCallRequest {
                     whatsapp_call_id: call_id.into(),
                     whatsapp_api_key: api_key.into(),
+                    ..Default::default()
                 },
                 self.base
                     .auth_header(VideoGrants { room_create: true, ..Default::default() }, None)?,

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 = { version = "0.4.28" }
+thiserror = "2.0.17"
+tokio = { version = "1.48.0", default-features = false, features = ["sync"] }
+futures-util = { version = "0.3", default-features = false, features = ["sink"] }
+futures-core = "0.3.31"
+bytes = "1.10.1"
+from_variants = "1.0.2"
+tokio-stream = { version = "0.1.17", features = ["sync"] }
+anyhow = "1.0.100" # For internal error handling only
+rand = "0.9.2"
+
+[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);