Data tracks UniFFI

.changeset/data_track_public_fields.md

@@ -0,0 +1,6 @@
+---
+livekit-datatrack: patch
+livekit: patch
+---
+
+# Make some fields public for data track types

.changeset/make_data_track_e2ee_errors_enums.md

@@ -0,0 +1,6 @@
+---
+livekit: patch
+livekit-datatrack: patch
+---
+
+# Make data track E2EE errors enums

.changeset/uniffi_data_tracks.md

@@ -0,0 +1,5 @@
+---
+livekit-uniffi: minor
+---
+
+# Expose data tracks core functionality

.changeset/use_concrete_type_for_data_track_manager_output_events.md

@@ -0,0 +1,7 @@
+---
+livekit-datatrack: patch
+livekit: patch
+livekit-ffi: patch
+---
+
+# Use concrete type for data track manager output events

livekit-datatrack/src/e2ee.rs

@@ -19,22 +19,29 @@ 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.
 
+/// Twelve byte AES initialization vector (IV).
+pub type InitializationVector = [u8; 12];
+
 /// Encrypted payload and metadata required for decryption.
 pub struct EncryptedPayload {
     pub payload: Bytes,
-    pub iv: [u8; 12],
+    pub iv: InitializationVector,
     pub key_index: u8,
 }
 
 /// An error indicating a payload could not be encrypted.
 #[derive(Debug, Error)]
-#[error("Encryption failed")]
-pub struct EncryptionError;
+pub enum EncryptionError {
+    #[error("Encryption failed")]
+    Failed,
+}
 
 /// An error indicating a payload could not be decrypted.
 #[derive(Debug, Error)]
-#[error("Decryption failed")]
-pub struct DecryptionError;
+pub enum DecryptionError {
+    #[error("Decryption failed")]
+    Failed,
+}
 
 /// Provider for encrypting payloads for E2EE.
 pub trait EncryptionProvider: Send + Sync + Debug {

livekit-datatrack/src/frame.rs

@@ -32,8 +32,8 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 ///
 #[derive(Clone, Default)]
 pub struct DataTrackFrame {
-    pub(crate) payload: Bytes,
-    pub(crate) user_timestamp: Option<u64>,
+    pub payload: Bytes,
+    pub user_timestamp: Option<u64>,
 }
 
 impl DataTrackFrame {

livekit-datatrack/src/local/manager.rs

@@ -25,7 +25,13 @@ use crate::{
 };
 use anyhow::{anyhow, Context};
 use futures_core::Stream;
-use std::{collections::HashMap, sync::Arc, time::Duration};
+use std::{
+    collections::HashMap,
+    pin::Pin,
+    sync::Arc,
+    task::{Context as TaskContext, Poll},
+    time::Duration,
+};
 use tokio::sync::{mpsc, oneshot, watch};
 use tokio_stream::wrappers::ReceiverStream;
 
@@ -58,7 +64,7 @@ impl Manager {
     /// - Channel for sending [`InputEvent`]s to be processed by the manager.
     /// - Stream for receiving [`OutputEvent`]s produced by the manager.
     ///
-    pub fn new(options: ManagerOptions) -> (Self, ManagerInput, impl Stream<Item = OutputEvent>) {
+    pub fn new(options: ManagerOptions) -> (Self, ManagerInput, ManagerOutput) {
         let (event_in_tx, event_in_rx) = mpsc::channel(Self::EVENT_BUFFER_COUNT);
         let (event_out_tx, event_out_rx) = mpsc::channel(Self::EVENT_BUFFER_COUNT);
 
@@ -72,7 +78,7 @@ impl Manager {
             descriptors: HashMap::new(),
         };
 
-        let event_out = ReceiverStream::new(event_out_rx);
+        let event_out = ManagerOutput(ReceiverStream::new(event_out_rx));
         (manager, event_in, event_out)
     }
 
@@ -400,6 +406,18 @@ pub struct ManagerInput {
     _drop_guard: Arc<DropGuard>,
 }
 
+/// Stream of [`OutputEvent`]s produced by [`Manager`].
+#[derive(Debug)]
+pub struct ManagerOutput(ReceiverStream<OutputEvent>);
+
+impl Stream for ManagerOutput {
+    type Item = OutputEvent;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<Option<Self::Item>> {
+        Pin::new(&mut self.0).poll_next(cx)
+    }
+}
+
 /// Guard that sends shutdown event when the last reference is dropped.
 #[derive(Debug)]
 struct DropGuard {

livekit-datatrack/src/local/mod.rs

@@ -152,7 +152,7 @@ impl Drop for LocalTrackInner {
 ///
 #[derive(Clone, Debug)]
 pub struct DataTrackOptions {
-    pub(crate) name: String,
+    pub name: String,
 }
 
 impl DataTrackOptions {

livekit-datatrack/src/remote/manager.rs

@@ -27,7 +27,9 @@ use bytes::Bytes;
 use std::{
     collections::{HashMap, HashSet},
     mem,
+    pin::Pin,
     sync::Arc,
+    task::{Context as TaskContext, Poll},
 };
 use tokio::sync::{broadcast, mpsc, oneshot, watch};
 use tokio_stream::{wrappers::ReceiverStream, Stream};
@@ -70,7 +72,7 @@ impl Manager {
     /// - Channel for sending [`InputEvent`]s to be processed by the manager.
     /// - Stream for receiving [`OutputEvent`]s produced by the manager.
     ///
-    pub fn new(options: ManagerOptions) -> (Self, ManagerInput, impl Stream<Item = OutputEvent>) {
+    pub fn new(options: ManagerOptions) -> (Self, ManagerInput, ManagerOutput) {
         let (event_in_tx, event_in_rx) = mpsc::channel(Self::EVENT_BUFFER_COUNT);
         let (event_out_tx, event_out_rx) = mpsc::channel(Self::EVENT_BUFFER_COUNT);
 
@@ -84,7 +86,7 @@ impl Manager {
             sub_handles: HashMap::default(),
         };
 
-        let event_out = ReceiverStream::new(event_out_rx);
+        let event_out = ManagerOutput(ReceiverStream::new(event_out_rx));
         (manager, event_in, event_out)
     }
 
@@ -443,6 +445,18 @@ pub struct ManagerInput {
     _drop_guard: Arc<DropGuard>,
 }
 
+/// Stream of [`OutputEvent`]s produced by [`Manager`].
+#[derive(Debug)]
+pub struct ManagerOutput(ReceiverStream<OutputEvent>);
+
+impl Stream for ManagerOutput {
+    type Item = OutputEvent;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<Option<Self::Item>> {
+        Pin::new(&mut self.0).poll_next(cx)
+    }
+}
+
 /// Guard that sends shutdown event when the last reference is dropped.
 #[derive(Debug)]
 struct DropGuard {

livekit-datatrack/src/track.rs

@@ -42,7 +42,7 @@ impl<L> DataTrack<L> {
         &self.info
     }
 
-    /// Whether or not the track is still published.
+    /// Whether or not the track is currently published.
     pub fn is_published(&self) -> bool {
         match self.inner.as_ref() {
             DataTrackInner::Local(inner) => inner.is_published(),

livekit-uniffi/Cargo.toml

@@ -14,10 +14,16 @@ publish = false
 [dependencies]
 livekit-protocol = { workspace = true }
 livekit-api = { workspace = true }
+livekit-datatrack = { workspace = true }
 uniffi = { version = "0.30.0", features = ["cli", "scaffolding-ffi-buffer-fns"] }
 log = { workspace = true }
-tokio = { workspace = true, features = ["sync"] }
+tokio = { workspace = true, features = ["sync", "rt-multi-thread"] }
+tokio-util =  "0.7.18"
+prost = "0.12"
+futures-util = { workspace = true, default-features = false, features = ["sink"] }
+bytes = { workspace = true }
 once_cell = "1.21.3"
+thiserror = { workspace = true }
 
 [build-dependencies]
 uniffi = { version = "0.30.0", features = ["build", "scaffolding-ffi-buffer-fns"] }

livekit-uniffi/src/common.rs

@@ -0,0 +1,17 @@
+// Copyright 2026 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;
+
+uniffi::custom_type!(Bytes, Vec<u8>, { remote });

livekit-uniffi/src/data_track/common.rs

@@ -0,0 +1,67 @@
+// Copyright 2026 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 livekit_datatrack::api::{DataTrackFrame, DataTrackSid};
+use livekit_protocol as proto;
+use prost::Message;
+
+uniffi::custom_type!(DataTrackSid, String, {
+    remote,
+    lower: |s| String::from(s),
+    try_lift: |s| DataTrackSid::try_from(s).map_err(|e| uniffi::deps::anyhow::anyhow!("{e}")),
+});
+
+#[uniffi::remote(Record)]
+pub struct DataTrackFrame {
+    pub payload: Bytes,
+    pub user_timestamp: Option<u64>,
+}
+
+/// Information about a published data track.
+#[derive(uniffi::Record)]
+pub struct DataTrackInfo {
+    pub sid: DataTrackSid,
+    pub name: String,
+    pub uses_e2ee: bool,
+}
+
+impl From<&livekit_datatrack::api::DataTrackInfo> for DataTrackInfo {
+    fn from(info: &livekit_datatrack::api::DataTrackInfo) -> Self {
+        Self { sid: info.sid(), name: info.name().to_string(), uses_e2ee: info.uses_e2ee() }
+    }
+}
+
+/// Signal response crossing the FFI boundary could not be processed.
+#[derive(uniffi::Error, thiserror::Error, Debug)]
+#[uniffi(flat_error)]
+pub enum HandleSignalResponseError {
+    #[error("Response decoding failed: {0}")]
+    Decode(prost::DecodeError),
+    #[error("Response container has no message")]
+    EmptyMessage,
+    #[error("Unsupported response type in this context")]
+    UnsupportedType,
+    #[error(transparent)]
+    Internal(livekit_datatrack::api::InternalError),
+}
+
+/// Deserializes a signal response crossing the FFI boundary, returning the message variant.
+pub(crate) fn deserialize_signal_response(
+    res: &[u8],
+) -> Result<proto::signal_response::Message, HandleSignalResponseError> {
+    let res =
+        proto::SignalResponse::decode(res).map_err(|err| HandleSignalResponseError::Decode(err))?;
+    res.message.ok_or(HandleSignalResponseError::EmptyMessage)
+}