Data tracks: subscribing

callback.go

@@ -18,6 +18,7 @@ import (
 	"github.com/pion/webrtc/v4"
 
 	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/server-sdk-go/v2/datatrack"
 )
 
 // ParticipantAttributesChangedFunc is callback for Participant attribute change event.
@@ -44,6 +45,8 @@ type ParticipantCallback struct {
 	OnTrackSubscriptionFailed func(sid string, rp *RemoteParticipant)
 	OnTrackPublished          func(publication *RemoteTrackPublication, rp *RemoteParticipant)
 	OnTrackUnpublished        func(publication *RemoteTrackPublication, rp *RemoteParticipant)
+	OnDataTrackPublished      func(track *datatrack.RemoteTrack, rp *RemoteParticipant)
+	OnDataTrackUnpublished    func(track *datatrack.RemoteTrack, rp *RemoteParticipant)
 	OnDataReceived            func(data []byte, params DataReceiveParams) // Deprecated: Use OnDataPacket instead
 	OnDataPacket              func(data DataPacket, params DataReceiveParams)
 	OnTranscriptionReceived   func(transcriptionSegments []*TranscriptionSegment, p Participant, publication TrackPublication)
@@ -66,6 +69,8 @@ func NewParticipantCallback() *ParticipantCallback {
 		OnTrackSubscriptionFailed:  func(sid string, rp *RemoteParticipant) {},
 		OnTrackPublished:           func(publication *RemoteTrackPublication, rp *RemoteParticipant) {},
 		OnTrackUnpublished:         func(publication *RemoteTrackPublication, rp *RemoteParticipant) {},
+		OnDataTrackPublished:       func(track *datatrack.RemoteTrack, rp *RemoteParticipant) {},
+		OnDataTrackUnpublished:     func(track *datatrack.RemoteTrack, rp *RemoteParticipant) {},
 		OnDataReceived:             func(data []byte, params DataReceiveParams) {},
 		OnDataPacket:               func(data DataPacket, params DataReceiveParams) {},
 		OnTranscriptionReceived:    func(transcriptionSegments []*TranscriptionSegment, p Participant, publication TrackPublication) {},
@@ -113,6 +118,12 @@ func (cb *ParticipantCallback) Merge(other *ParticipantCallback) {
 	if other.OnTrackUnpublished != nil {
 		cb.OnTrackUnpublished = other.OnTrackUnpublished
 	}
+	if other.OnDataTrackPublished != nil {
+		cb.OnDataTrackPublished = other.OnDataTrackPublished
+	}
+	if other.OnDataTrackUnpublished != nil {
+		cb.OnDataTrackUnpublished = other.OnDataTrackUnpublished
+	}
 	if other.OnDataReceived != nil {
 		cb.OnDataReceived = other.OnDataReceived
 	}

datatrack/datatrack.go

@@ -12,8 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Package datatrack is the public API for data tracks. Publish a track with
-// [github.com/livekit/server-sdk-go/v2.LocalParticipant.PublishDataTrack].
+// Package datatrack is the public API for data tracks.
 package datatrack
 
 import (
@@ -23,6 +22,10 @@ import (
 type (
 	// LocalTrack is a data track published by the local participant.
 	LocalTrack = backend.LocalTrack
+	// RemoteTrack is a data track published by a remote participant.
+	RemoteTrack = backend.RemoteTrack
+	// Stream delivers the frames of a subscription.
+	Stream = backend.Stream
 	// Info describes a published data track.
 	Info = backend.Info
 	// Frame is a frame published on a data track.
@@ -33,6 +36,13 @@ type (
 	// PublishOption customizes a publication.
 	PublishOption = backend.PublishOption
 
+	// SubscribeOptions configure a subscription.
+	SubscribeOptions = backend.SubscribeOptions
+	// SubscribeOption customizes a subscription.
+	SubscribeOption = backend.SubscribeOption
+	// PipelineOptions configure how a remote track's packets are reassembled.
+	PipelineOptions = backend.PipelineOptions
+
 	// SchemaID identifies a data track schema.
 	SchemaID = backend.SchemaID
 	// SchemaEncoding is the encoding of a schema definition: a WellKnownSchemaEncoding or a
@@ -53,14 +63,15 @@ type (
 )
 
 var (
-	ErrNotAllowed     = backend.ErrNotAllowed
-	ErrDuplicateName  = backend.ErrDuplicateName
-	ErrInvalidName    = backend.ErrInvalidName
-	ErrLimitReached   = backend.ErrLimitReached
-	ErrPublishTimeout = backend.ErrPublishTimeout
-	ErrDisconnected   = backend.ErrDisconnected
-	ErrUnpublished    = backend.ErrUnpublished
-	ErrQueueFull      = backend.ErrQueueFull
+	ErrNotAllowed       = backend.ErrNotAllowed
+	ErrDuplicateName    = backend.ErrDuplicateName
+	ErrInvalidName      = backend.ErrInvalidName
+	ErrLimitReached     = backend.ErrLimitReached
+	ErrPublishTimeout   = backend.ErrPublishTimeout
+	ErrSubscribeTimeout = backend.ErrSubscribeTimeout
+	ErrDisconnected     = backend.ErrDisconnected
+	ErrUnpublished      = backend.ErrUnpublished
+	ErrQueueFull        = backend.ErrQueueFull
 
 	ErrSchemaMissingFrameEncoding = backend.ErrSchemaMissingFrameEncoding
 	ErrSchemaMissingID            = backend.ErrSchemaMissingID
@@ -101,6 +112,12 @@ func WithFrameEncoding(encoding FrameEncoding) PublishOption {
 	return backend.WithFrameEncoding(encoding)
 }
 
+// WithBufferSize sets the number of received frames buffered for the subscriber. It has no
+// effect when the track already has an active subscription.
+func WithBufferSize(frames int) SubscribeOption {
+	return backend.WithBufferSize(frames)
+}
+
 // UserTimestampNow returns the current Unix time in milliseconds, the interpretation
 // Frame.DurationSinceTimestamp expects.
 func UserTimestampNow() *uint64 {

examples/datatrack/subscriber/main.go

@@ -0,0 +1,90 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/livekit/protocol/logger"
+	lksdk "github.com/livekit/server-sdk-go/v2"
+	"github.com/livekit/server-sdk-go/v2/datatrack"
+	"github.com/urfave/cli/v3"
+)
+
+func main() {
+	logger.InitFromConfig(&logger.Config{Level: "info"}, "datatrack-subscriber")
+	lksdk.SetLogger(logger.GetLogger())
+
+	cmd := &cli.Command{
+		Name:  "datatrack-subscriber",
+		Usage: "subscribe to data tracks and log received frames",
+		Flags: []cli.Flag{
+			&cli.StringFlag{Name: "host", Usage: "livekit server host", Required: true, Sources: cli.EnvVars("LIVEKIT_URL")},
+			&cli.StringFlag{Name: "api-key", Usage: "livekit api key", Required: true, Sources: cli.EnvVars("LIVEKIT_API_KEY")},
+			&cli.StringFlag{Name: "api-secret", Usage: "livekit api secret", Required: true, Sources: cli.EnvVars("LIVEKIT_API_SECRET")},
+			&cli.StringFlag{Name: "room-name", Usage: "room name", Required: true},
+			&cli.StringFlag{Name: "identity", Usage: "participant identity", Value: "subscriber"},
+		},
+		Action: run,
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+	if err := cmd.Run(ctx, os.Args); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+}
+
+func run(ctx context.Context, cmd *cli.Command) error {
+	// Subscribe to any published data tracks
+	callback := &lksdk.RoomCallback{
+		ParticipantCallback: lksdk.ParticipantCallback{
+			OnDataTrackPublished: func(track *datatrack.RemoteTrack, rp *lksdk.RemoteParticipant) {
+				go subscribe(ctx, track)
+			},
+		},
+	}
+
+	room, err := lksdk.ConnectToRoom(cmd.String("host"), lksdk.ConnectInfo{
+		APIKey:              cmd.String("api-key"),
+		APISecret:           cmd.String("api-secret"),
+		RoomName:            cmd.String("room-name"),
+		ParticipantIdentity: cmd.String("identity"),
+	}, callback)
+	if err != nil {
+		return err
+	}
+	defer room.Disconnect()
+
+	<-ctx.Done()
+	return nil
+}
+
+// subscribe subscribes to the given data track and logs received frames.
+func subscribe(ctx context.Context, track *datatrack.RemoteTrack) {
+	lgr := logger.GetLogger().WithValues("trackSID", track.Info().SID, "trackName", track.Info().Name, "publisher", track.PublisherIdentity())
+	lgr.Infow("subscribing")
+
+	stream, err := track.Subscribe(ctx)
+	if err != nil {
+		lgr.Warnw("failed to subscribe", err)
+		return
+	}
+	defer stream.Close()
+
+	for {
+		frame, err := stream.Next(ctx)
+		if err != nil {
+			break
+		}
+		lgr.Infow("received frame", "bytes", len(frame.Payload))
+
+		if latency, ok := frame.DurationSinceTimestamp(); ok {
+			lgr.Infow("latency", "duration", latency)
+		}
+	}
+	lgr.Infow("unsubscribed")
+}

internal/datatrack/errors.go

@@ -17,12 +17,13 @@ package datatrack
 import "errors"
 
 var (
-	ErrNotAllowed     = errors.New("data track publishing unauthorized")
-	ErrDuplicateName  = errors.New("track name already taken")
-	ErrInvalidName    = errors.New("track name invalid")
-	ErrLimitReached   = errors.New("data track publication limit reached")
-	ErrPublishTimeout = errors.New("timed out publishing data track")
-	ErrDisconnected   = errors.New("room disconnected")
-	ErrUnpublished    = errors.New("track unpublished")
-	ErrQueueFull      = errors.New("queue full")
+	ErrNotAllowed       = errors.New("data track publishing unauthorized")
+	ErrDuplicateName    = errors.New("track name already taken")
+	ErrInvalidName      = errors.New("track name invalid")
+	ErrLimitReached     = errors.New("data track publication limit reached")
+	ErrPublishTimeout   = errors.New("timed out publishing data track")
+	ErrSubscribeTimeout = errors.New("timed out subscribing to data track")
+	ErrDisconnected     = errors.New("room disconnected")
+	ErrUnpublished      = errors.New("track unpublished")
+	ErrQueueFull        = errors.New("queue full")
 )

internal/datatrack/helpers_test.go

@@ -0,0 +1,64 @@
+// 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.
+
+package datatrack
+
+import (
+	"bytes"
+	"testing"
+	"time"
+
+	dtp "github.com/livekit/protocol/datatrack"
+)
+
+// testE2EEPrefix marks payloads "encrypted" by prefixingEncryptor.
+var testE2EEPrefix = []byte{0xde, 0xad, 0xbe, 0xef}
+
+// prefixingEncryptor marks payloads instead of encrypting them.
+type prefixingEncryptor struct{}
+
+func (prefixingEncryptor) Encrypt(payload []byte) ([]byte, dtp.ExtensionE2EE, error) {
+	return append(bytes.Clone(testE2EEPrefix), payload...), dtp.ExtensionE2EE{}, nil
+}
+
+// prefixStrippingDecryptor undoes prefixingEncryptor.
+type prefixStrippingDecryptor struct{}
+
+func (prefixStrippingDecryptor) Decrypt(payload []byte, _ dtp.ExtensionE2EE) ([]byte, error) {
+	return payload[len(testE2EEPrefix):], nil
+}
+
+// expectEvent receives from ch or fails fast. The 500 ms timeout is generous
+// enough to avoid flakes on slow CI while keeping failures quick.
+func expectEvent[T any](t *testing.T, ch <-chan T) T {
+	t.Helper()
+	select {
+	case event := <-ch:
+		return event
+	case <-time.After(500 * time.Millisecond):
+		var zero T
+		t.Fatalf("timed out waiting for %T", zero)
+		return zero
+	}
+}
+
+// expectNoEvent fails if ch delivers anything within the grace period.
+func expectNoEvent[T any](t *testing.T, ch <-chan T) {
+	t.Helper()
+	select {
+	case event := <-ch:
+		t.Fatalf("unexpected event %v", event)
+	case <-time.After(50 * time.Millisecond):
+	}
+}

internal/datatrack/local_test.go

@@ -21,7 +21,6 @@ import (
 	"testing"
 	"time"
 
-	dtp "github.com/livekit/protocol/datatrack"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/utils/pointer"
 	"github.com/stretchr/testify/require"
@@ -56,27 +55,6 @@ func (f *fakeLocalTransport) SendFrame(frame FramePackets) {
 	f.frames <- frame
 }
 
-// prefixingEncryptor marks payloads instead of encrypting them.
-type prefixingEncryptor struct{}
-
-func (prefixingEncryptor) Encrypt(payload []byte) ([]byte, dtp.ExtensionE2EE, error) {
-	return append([]byte{0xde, 0xad, 0xbe, 0xef}, payload...), dtp.ExtensionE2EE{}, nil
-}
-
-// expectEvent receives from ch or fails fast. The 500 ms timeout is generous
-// enough to avoid flakes on slow CI while keeping failures quick.
-func expectEvent[T any](t *testing.T, ch <-chan T) T {
-	t.Helper()
-	select {
-	case event := <-ch:
-		return event
-	case <-time.After(500 * time.Millisecond):
-		var zero T
-		t.Fatalf("timed out waiting for %T", zero)
-		return zero
-	}
-}
-
 // publishAsync calls Publish on another goroutine and delivers its outcome on the returned channel.
 func publishAsync(ctx context.Context, m *LocalManager, name string) <-chan publishResult {
 	result := make(chan publishResult, 1)
@@ -273,8 +251,8 @@ func TestLocalManager_PublishWithE2EE(t *testing.T) {
 	packets := expectEvent(t, transport.frames)
 	packet, err := parsePacket(packets[0])
 	require.NoError(t, err)
-	require.Equal(t, []byte{0xde, 0xad, 0xbe, 0xef}, packet.Payload[:4])
-	require.Equal(t, []byte{1, 2, 3, 4, 5}, packet.Payload[4:])
+	require.Equal(t, testE2EEPrefix, packet.Payload[:len(testE2EEPrefix)])
+	require.Equal(t, []byte{1, 2, 3, 4, 5}, packet.Payload[len(testE2EEPrefix):])
 	require.NotNil(t, extensionsOf(&packet.Header).E2EE)
 }
 

internal/datatrack/proto.go

@@ -17,6 +17,7 @@ import (
 	"fmt"
 
 	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
 )
 
 var (
@@ -273,3 +274,41 @@ func publishRejectionFromRequestResponse(msg *livekit.RequestResponse) (rejectio
 	}
 	return rejection, true
 }
+
+// subscriptionUpdate asks the SFU to start or stop delivering a track's packets.
+type subscriptionUpdate struct {
+	sid       livekit.TrackID
+	subscribe bool
+}
+
+func (u subscriptionUpdate) toProto() *livekit.UpdateDataSubscription {
+	return &livekit.UpdateDataSubscription{Updates: []*livekit.UpdateDataSubscription_Update{{
+		TrackSid:  string(u.sid),
+		Subscribe: u.subscribe,
+	}}}
+}
+
+// publicationUpdatesFromProto maps each remote participant to the data tracks it publishes. The
+// local participant is skipped and a disconnected participant contributes an empty list. Tracks
+// that fail to convert are dropped with a warning.
+func publicationUpdatesFromProto(participants []*livekit.ParticipantInfo, localIdentity string, log logger.Logger) map[string][]Info {
+	updates := make(map[string][]Info, len(participants))
+	for _, participant := range participants {
+		if participant.GetIdentity() == localIdentity {
+			continue
+		}
+		infos := make([]Info, 0, len(participant.GetDataTracks()))
+		if participant.GetState() != livekit.ParticipantInfo_DISCONNECTED {
+			for _, msg := range participant.GetDataTracks() {
+				info, err := infoFromProto(msg)
+				if err != nil {
+					log.Warnw("ignoring invalid data track info", err, "participant", participant.GetIdentity())
+					continue
+				}
+				infos = append(infos, info)
+			}
+		}
+		updates[participant.GetIdentity()] = infos
+	}
+	return updates
+}