Data tracks integration

.changeset/initial_support_for_data_tracks.md

@@ -0,0 +1,5 @@
+---
+livekit: minor
+---
+
+# Initial support for data tracks

.changeset/rename_to_data_track_stream.md

@@ -0,0 +1,5 @@
+---
+livekit-datatrack: patch
+---
+
+# Rename type to DataTrackStream

Cargo.toml

@@ -19,6 +19,7 @@ members = [
 
     "examples/agent_dispatch",
     "examples/api",
+    "examples/basic_data_track",
     "examples/basic_room",
     "examples/basic_text_stream",
     "examples/encrypted_text_stream",

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 = { workspace = true, features = ["full"] }
+livekit = { workspace = true, features = ["rustls-tls-native-roots"] }
+futures-util = { workspace = true, default-features = false, features = ["sink"] }
+anyhow = { workspace = true }
+log = { workspace = true }
+env_logger = { workspace = true }
+
+[[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 reading = read_sensor().await;
+        let frame = DataTrackFrame::new(reading).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,51 @@
+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_publications(rx) => {}
+        _ = signal::ctrl_c() => {}
+    }
+    Ok(())
+}
+
+/// Subscribes to any published data tracks.
+async fn handle_publications(mut rx: UnboundedReceiver<RoomEvent>) -> Result<()> {
+    while let Some(event) = rx.recv().await {
+        let RoomEvent::DataTrackPublished(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 subscription = track.subscribe().await?;
+    while let Some(frame) = subscription.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(())
+}