Data tracks

Overview

Data tracks provide a publish/subscribe channel for streaming arbitrary binary data between participants, modeled on the same lifecycle as audio and video tracks. A participant publishes a named track, other participants subscribe to it, and the publisher pushes frames as data becomes available. Tracks are lightweight, so you can publish one per sensor or actuator and let each subscriber choose the streams it needs.

Delivery is in-order but lossy: frames aren't retransmitted, so under network pressure the channel drops older frames rather than delaying newer ones. This favors freshness over completeness, which suits continuous, high-frequency streams where the latest sensor reading or control command matters more than every intermediate value. Each frame carries a binary payload of any format, plus an optional 64-bit user timestamp to record capture time or measure end-to-end latency.

Use data tracks for continuous sensor streams (for example, IMU, LiDAR, RGBD), teleoperation control input, telemetry, and non-standard media such as MJPEG. When you need guaranteed delivery, use text and byte streams instead. For request-response interactions, use remote procedure calls.

Example: Continuous sensor data

This example publishes continuous sensor readings over a data track. A participant with access to an RGB color sensor publishes one reading per frame at a fixed rate. Because the sensor samples a single color value rather than an image, each frame contains one byte per color channel.

let track = room
    .local_participant()
    .publish_data_track("rgb_sensor").await?;

let sensor = MyRgbSensor::new(sensor_config)?;

// Maintain ~30 FPS publish rate. Using tokio:
let mut interval = time::interval(Duration::from_secs_f64(1.0 / 30.0));
interval.set_missed_tick_behavior(MissedTickBehavior::Skip);

while track.is_published() {
    interval.tick().await;
    let reading = sensor.latest_reading();

    let frame = DataTrackFrame::new(reading.value.into()) // [u8; 3]
        .with_user_timestamp(reading.timestamp);

    track.try_push(frame).ok();
}

track = await room.local_participant.publish_data_track(
    name="rgb_sensor"
)

sensor = MyRgbSensor(sensor_config)

# Maintain ~30 FPS publish rate.
while track.is_published():
    await asyncio.sleep(1 / 30)
    reading = sensor.latest_reading()

    frame = rtc.DataTrackFrame(
        payload=reading.value,  # 3 bytes
        user_timestamp=reading.timestamp,
    )
    track.try_push(frame)

std::shared_ptr<livekit::LocalDataTrack> track;
if (auto lp = room->localParticipant().lock()) {
  auto publish_result = lp->publishDataTrack("rgb_sensor");
  if (!publish_result) {
    std::cerr << "Failed to publish data track: "
              << publish_result.error().message << "\n";
    return;
  }
  track = publish_result.value();
} else {
  std::cerr << "Failed to get local participant\n";
  return;
}

MyRgbSensor sensor(sensor_config);

// Maintain ~30 FPS publish rate.
const auto period = std::chrono::microseconds(1'000'000 / 30);

while (track->isPublished()) {
  const auto next_push = std::chrono::steady_clock::now() + period;
  const auto reading = sensor.latestReading();

  livekit::DataTrackFrame frame;
  frame.payload = reading.value;  // 3 bytes
  frame.user_timestamp = reading.timestamp;

  track->tryPush(frame);
  std::this_thread::sleep_until(next_push);
}

const track = await room.localParticipant.publishDataTrack({
  name: 'rgb_sensor',
});

const sensor = new MyRgbSensor(sensorConfig);

// Maintain ~30 FPS publish rate.
while (track.isPublished()) {
  await new Promise((resolve) => setTimeout(resolve, 1000 / 30));
  const reading = sensor.latestReading();

  track.tryPush({
    payload: reading.value, // Uint8Array of 3 bytes
    userTimestamp: BigInt(reading.timestamp),
  });
}

IEnumerator PublishSensor(Room room)
{
    var publish = room.LocalParticipant.PublishDataTrack(
        "rgb_sensor");
    yield return publish;
    if (publish.IsError) yield break;

    var track = publish.Track;
    var sensor = new MyRgbSensor(sensorConfig);

    // Maintain ~30 FPS publish rate.
    while (track.IsPublished())
    {
        yield return new WaitForSeconds(1f / 30f);
        var reading = sensor.LatestReading();

        track.TryPush(new DataTrackFrame(
            reading.Value, // 3 bytes
            reading.Timestamp));
    }
}

Other participants discover the published track through a room event and subscribe to receive sensor readings as they arrive.

while let Some(event) = room_events.recv().await {
    let RoomEvent::DataTrackPublished(track) = event else {
        continue;
    };
    if track.info().name() != "rgb_sensor" {
        continue;
    }
    tokio::spawn(async move {
        if let Err(error) = handle_rgb_track(track).await {
            println!("Unable to handle track: {}", error);
        }
    });
}

async fn handle_rgb_track(track: RemoteDataTrack) -> Result<()> {
    let mut subscription = track.subscribe().await?;
    while let Some(frame) = subscription.next().await {
        let rgb: [u8; 3] = frame.payload().as_ref().try_into()
            .context("Unexpected frame format")?;
        let timestamp = frame.user_timestamp()
            .context("Expected timestamp")?;

        println!("Reading @ T{}: {:?}", timestamp, rgb);
        // Example output: "Reading @ T180000: [255, 36, 124]"
    }
    Ok(())
}

@room.on("data_track_published")
def on_data_track_published(track: rtc.RemoteDataTrack):
    if track.info.name != "rgb_sensor":
        return
    asyncio.create_task(handle_rgb_track(track))


async def handle_rgb_track(track: rtc.RemoteDataTrack):
    stream = track.subscribe()
    async for frame in stream:
        if len(frame.payload) != 3 or frame.user_timestamp is None:
            print("Unexpected frame format")
            continue
        rgb = tuple(frame.payload)

        print(f"Reading @ T{frame.user_timestamp}: {rgb}")
        # Example output: "Reading @ T180000: (255, 36, 124)"

Use addOnDataFrameCallback, which handles the publish event, subscription, and callback threading for you. Pass the publisher identity and track name:

const auto callback_id = room->addOnDataFrameCallback(
    "robot", "rgb_sensor",
    [](const std::vector<std::uint8_t>& payload,
       std::optional<std::uint64_t> user_timestamp) {
      if (payload.size() != 3 || !user_timestamp) {
        std::cerr << "Unexpected frame format\n";
        return;
      }

      std::cout << "Reading @ T" << *user_timestamp << ": ["
                << static_cast<int>(payload[0]) << ", "
                << static_cast<int>(payload[1]) << ", "
                << static_cast<int>(payload[2]) << "]\n";
      // Example output: "Reading @ T180000: [255, 36, 124]"
    });

// Later, when you no longer want frames from this data track:
room->removeOnDataFrameCallback(callback_id);

import { RoomEvent } from 'livekit-client';

room.on(RoomEvent.DataTrackPublished, async (track) => {
  if (track.info.name !== 'rgb_sensor') {
    return;
  }

  const stream = track.subscribe();
  for await (const frame of stream) {
    if (frame.payload.length !== 3 || !frame.userTimestamp) {
      console.error('Unexpected frame format');
      continue;
    }
    const rgb = Array.from(frame.payload);

    console.log(`Reading @ T${frame.userTimestamp}: [${rgb}]`);
    // Example output: "Reading @ T180000: [255,36,124]"
  }
});

room.DataTrackPublished += track =>
{
    if (track.Info.Name != "rgb_sensor") return;
    StartCoroutine(HandleRgbTrack(track));
};

IEnumerator HandleRgbTrack(RemoteDataTrack track)
{
    var stream = track.Subscribe();

    while (!stream.IsEos)
    {
        var read = stream.ReadFrame();
        yield return read;
        if (!read.IsCurrentReadDone) continue;

        var frame = read.Frame;
        if (frame.Payload.Length != 3 || frame.UserTimestamp == null)
        {
            Debug.LogError("Unexpected frame format");
            continue;
        }
        var rgb = frame.Payload;

        Debug.Log($"Reading @ T{frame.UserTimestamp}: " +
                  $"[{rgb[0]}, {rgb[1]}, {rgb[2]}]");
        // Example output: "Reading @ T180000: [255, 36, 124]"
    }

    stream.Close();
}

Adapt this example to your app. For more complex readings, serialize the payload in a format such as JSON or Protobuf. Subscribe only while the data is in use, such as when plotting values on screen, to reduce bandwidth. This is especially important for larger payloads such as images or point clouds.

Additional resources

Full data tracks API and examples in every client SDK.

Overview

Example: Continuous sensor data

Additional resources

Data tracks

Ask LiveKit