Data tracks: publishing

datatrack/datatrack.go

@@ -0,0 +1,108 @@
+// 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 is the public API for data tracks. Publish a track with
+// [github.com/livekit/server-sdk-go/v2.LocalParticipant.PublishDataTrack].
+package datatrack
+
+import (
+	backend "github.com/livekit/server-sdk-go/v2/internal/datatrack"
+)
+
+type (
+	// LocalTrack is a data track published by the local participant.
+	LocalTrack = backend.LocalTrack
+	// Info describes a published data track.
+	Info = backend.Info
+	// Frame is a frame published on a data track.
+	Frame = backend.Frame
+
+	// PublishOptions describe a data track to publish.
+	PublishOptions = backend.PublishOptions
+	// PublishOption customizes a publication.
+	PublishOption = backend.PublishOption
+
+	// SchemaID identifies a data track schema.
+	SchemaID = backend.SchemaID
+	// SchemaEncoding is the encoding of a schema definition: a WellKnownSchemaEncoding or a
+	// CustomSchemaEncoding.
+	SchemaEncoding = backend.SchemaEncoding
+	// WellKnownSchemaEncoding is a schema encoding defined by the protocol.
+	WellKnownSchemaEncoding = backend.WellKnownSchemaEncoding
+	// CustomSchemaEncoding is an application-defined schema encoding identified by name.
+	CustomSchemaEncoding = backend.CustomSchemaEncoding
+
+	// FrameEncoding is the encoding of frames pushed on a data track: a WellKnownFrameEncoding or
+	// a CustomFrameEncoding.
+	FrameEncoding = backend.FrameEncoding
+	// WellKnownFrameEncoding is a frame encoding defined by the protocol.
+	WellKnownFrameEncoding = backend.WellKnownFrameEncoding
+	// CustomFrameEncoding is an application-defined frame encoding identified by name.
+	CustomFrameEncoding = backend.CustomFrameEncoding
+)
+
+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
+
+	ErrSchemaMissingFrameEncoding = backend.ErrSchemaMissingFrameEncoding
+	ErrSchemaMissingID            = backend.ErrSchemaMissingID
+	ErrSchemaIncompatible         = backend.ErrSchemaIncompatible
+)
+
+const (
+	// SchemaEncodingOther is a well-known encoding not known to this SDK version.
+	SchemaEncodingOther      = backend.SchemaEncodingOther
+	SchemaEncodingProtobuf   = backend.SchemaEncodingProtobuf
+	SchemaEncodingFlatbuffer = backend.SchemaEncodingFlatbuffer
+	SchemaEncodingROS1Msg    = backend.SchemaEncodingROS1Msg
+	SchemaEncodingROS2Msg    = backend.SchemaEncodingROS2Msg
+	SchemaEncodingROS2IDL    = backend.SchemaEncodingROS2IDL
+	SchemaEncodingOMGIDL     = backend.SchemaEncodingOMGIDL
+	SchemaEncodingJSONSchema = backend.SchemaEncodingJSONSchema
+)
+
+const (
+	// FrameEncodingOther is a well-known encoding not known to this SDK version.
+	FrameEncodingOther      = backend.FrameEncodingOther
+	FrameEncodingROS1       = backend.FrameEncodingROS1
+	FrameEncodingCDR        = backend.FrameEncodingCDR
+	FrameEncodingProtobuf   = backend.FrameEncodingProtobuf
+	FrameEncodingFlatbuffer = backend.FrameEncodingFlatbuffer
+	FrameEncodingCBOR       = backend.FrameEncodingCBOR
+	FrameEncodingMsgPack    = backend.FrameEncodingMsgPack
+	FrameEncodingJSON       = backend.FrameEncodingJSON
+)
+
+// WithSchema associates a schema with the track's frames.
+func WithSchema(schema SchemaID) PublishOption {
+	return backend.WithSchema(schema)
+}
+
+// WithFrameEncoding declares the encoding of the track's frames.
+func WithFrameEncoding(encoding FrameEncoding) PublishOption {
+	return backend.WithFrameEncoding(encoding)
+}
+
+// UserTimestampNow returns the current Unix time in milliseconds, the interpretation
+// Frame.DurationSinceTimestamp expects.
+func UserTimestampNow() *uint64 {
+	return backend.UserTimestampNow()
+}

datatracksender.go

@@ -19,21 +19,20 @@ import (
 
 	protoLogger "github.com/livekit/protocol/logger"
 	"github.com/pion/webrtc/v4"
+
+	"github.com/livekit/server-sdk-go/v2/internal/datatrack"
 )
 
 const dataTrackBufferedAmountLowThreshold = 8 * 1024
 
-// dataTrackFramePackets is one frame serialized into data track packets.
-type dataTrackFramePackets [][]byte
-
 // dataTrackSender paces frames onto the data track channel, keeping only the freshest frame
 // while the channel's send buffer is above the low threshold.
 type dataTrackSender struct {
 	log protoLogger.Logger
 
 	lock   sync.Mutex
 	dc     *webrtc.DataChannel
-	frame  dataTrackFramePackets
+	frame  datatrack.FramePackets
 	notify chan struct{}
 	done   chan struct{}
 }
@@ -61,7 +60,7 @@ func (s *dataTrackSender) setLogger(log protoLogger.Logger) {
 	s.log = log
 }
 
-func (s *dataTrackSender) send(frame dataTrackFramePackets) {
+func (s *dataTrackSender) send(frame datatrack.FramePackets) {
 	if dropped := s.push(frame); dropped != nil {
 		s.log.Debugw("dropping data track frame", "numPackets", len(dropped))
 	}
@@ -78,7 +77,7 @@ func (s *dataTrackSender) stop() {
 	close(s.done)
 }
 
-func (s *dataTrackSender) push(frame dataTrackFramePackets) (dropped dataTrackFramePackets) {
+func (s *dataTrackSender) push(frame datatrack.FramePackets) (dropped datatrack.FramePackets) {
 	if len(frame) == 0 {
 		return nil
 	}
@@ -91,7 +90,7 @@ func (s *dataTrackSender) push(frame dataTrackFramePackets) (dropped dataTrackFr
 	return dropped
 }
 
-func (s *dataTrackSender) pop() dataTrackFramePackets {
+func (s *dataTrackSender) pop() datatrack.FramePackets {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -103,7 +102,7 @@ func (s *dataTrackSender) pop() dataTrackFramePackets {
 func (s *dataTrackSender) run() {
 	var (
 		dc       *webrtc.DataChannel
-		inFlight dataTrackFramePackets
+		inFlight datatrack.FramePackets
 	)
 	for {
 		select {

datatracksender_test.go

@@ -18,10 +18,12 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/server-sdk-go/v2/internal/datatrack"
 )
 
-func testFrame(marker byte, packets int) dataTrackFramePackets {
-	frame := make(dataTrackFramePackets, packets)
+func testFrame(marker byte, packets int) datatrack.FramePackets {
+	frame := make(datatrack.FramePackets, packets)
 	for i := range frame {
 		frame[i] = []byte{marker, byte(i)}
 	}

engine.go

@@ -34,6 +34,7 @@ import (
 	protosignalling "github.com/livekit/protocol/signalling"
 
 	"github.com/livekit/server-sdk-go/v2/e2ee"
+	"github.com/livekit/server-sdk-go/v2/internal/datatrack"
 	"github.com/livekit/server-sdk-go/v2/signalling"
 )
 
@@ -746,22 +747,24 @@ func (e *RTCEngine) GetDataChannelSub(kind livekit.DataPacket_Kind) *webrtc.Data
 }
 
 func (e *RTCEngine) waitUntilConnected(timeout time.Duration) error {
-	return waitUntilConnected(timeout, func() bool {
+	return waitUntilConnected(context.Background(), timeout, func() bool {
 		return e.IsConnected()
 	})
 }
 
-func (e *RTCEngine) ensurePublisherConnected(ensureDataReady bool) error {
+// ensurePublisherConnected waits for the publisher connection, giving up when ctx ends or the
+// connect timeout passes, whichever comes first.
+func (e *RTCEngine) ensurePublisherConnected(ctx context.Context, ensureDataReady bool) error {
 	e.pclock.Lock()
 	subscriberPrimary := e.subscriberPrimary
 	e.pclock.Unlock()
 	connectTimeout := e.connectionManager.getConnectTimeout()
 	if !subscriberPrimary {
-		return e.waitUntilConnected(connectTimeout)
+		return waitUntilConnected(ctx, connectTimeout, e.IsConnected)
 	}
 
 	var negotiated bool
-	return waitUntilConnected(connectTimeout, func() bool {
+	return waitUntilConnected(ctx, connectTimeout, func() bool {
 		if publisher, ok := e.Publisher(); ok {
 			if publisher.IsConnected() && (!ensureDataReady || e.dataPubChannelReady()) {
 				return true
@@ -1231,7 +1234,7 @@ func filterTURNServers(iceServers []*livekit.ICEServer) []*livekit.ICEServer {
 }
 
 func (e *RTCEngine) publishDataPacket(pck *livekit.DataPacket, kind livekit.DataPacket_Kind) error {
-	err := e.ensurePublisherConnected(true)
+	err := e.ensurePublisherConnected(context.Background(), true)
 	if err != nil {
 		e.log.Errorw("could not ensure publisher connected", err)
 		return err
@@ -1836,7 +1839,7 @@ func setConfiguration(pcTransport *PCTransport, configuration webrtc.Configurati
 	}
 }
 
-func waitUntilConnected(d time.Duration, test func() bool) error {
+func waitUntilConnected(ctx context.Context, d time.Duration, test func() bool) error {
 	if test() {
 		return nil
 	}
@@ -1849,6 +1852,9 @@ func waitUntilConnected(d time.Duration, test func() bool) error {
 
 	for {
 		select {
+		case <-ctx.Done():
+			return ctx.Err()
+
 		case <-timeout.C:
 			return ErrConnectionTimeout
 
@@ -1860,6 +1866,6 @@ func waitUntilConnected(d time.Duration, test func() bool) error {
 	}
 }
 
-func (e *RTCEngine) sendDataTrackFrame(frame dataTrackFramePackets) {
+func (e *RTCEngine) sendDataTrackFrame(frame datatrack.FramePackets) {
 	e.dataTrackSender.send(frame)
 }

examples/datatrack/publisher/main.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"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-publisher")
+	lksdk.SetLogger(logger.GetLogger())
+
+	cmd := &cli.Command{
+		Name:  "datatrack-publisher",
+		Usage: "publish sensor data on a data track",
+		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: "publisher"},
+		},
+		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 {
+	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"),
+	}, nil)
+	if err != nil {
+		return err
+	}
+	defer room.Disconnect()
+
+	track, err := room.LocalParticipant.PublishDataTrack(ctx, "my_sensor_data")
+	if err != nil {
+		return err
+	}
+	defer track.Unpublish()
+
+	pushFrames(ctx, track)
+	return nil
+}
+
+func readSensor() []byte {
+	// Dynamically read some sensor data...
+	return bytes.Repeat([]byte{0xfa}, 256)
+}
+
+func pushFrames(ctx context.Context, track *datatrack.LocalTrack) {
+	for {
+		logger.Infow("pushing frame")
+
+		frame := datatrack.Frame{
+			Payload:       readSensor(),
+			UserTimestamp: datatrack.UserTimestampNow(),
+		}
+		if err := track.TryPush(frame); err != nil {
+			logger.Warnw("failed to push frame", err)
+		}
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(500 * time.Millisecond):
+		}
+	}
+}

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/pion/webrtc/v4 v4.2.24
 	github.com/stretchr/testify v1.12.1
 	github.com/twitchtv/twirp v8.1.3+incompatible
+	github.com/urfave/cli/v3 v3.9.0
 	go.uber.org/atomic v1.12.0
 	golang.org/x/crypto v0.57.0
 	google.golang.org/protobuf v1.36.12

go.sum

@@ -282,6 +282,8 @@ github.com/tonistiigi/vt100 v0.0.0-20240514184818-90bafcd6abab h1:H6aJ0yKQ0gF49Q
 github.com/tonistiigi/vt100 v0.0.0-20240514184818-90bafcd6abab/go.mod h1:ulncasL3N9uLrVann0m+CDlJKWsIAP34MPcOJF6VRvc=
 github.com/twitchtv/twirp v8.1.3+incompatible h1:+F4TdErPgSUbMZMwp13Q/KgDVuI7HJXP61mNV3/7iuU=
 github.com/twitchtv/twirp v8.1.3+incompatible/go.mod h1:RRJoFSAmTEh2weEqWtpPE3vFK5YBhA6bqp2l1kfCC5A=
+github.com/urfave/cli/v3 v3.9.0 h1:AV9lIiPv3ukYnxunaCUsHnEozptYmDN2F0+yWqLMn/c=
+github.com/urfave/cli/v3 v3.9.0/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso=
 github.com/vbatts/tar-split v0.12.3 h1:Cd46rkGXI3Td4yrVNwU8ripbxFaQbmesqhjBUUYAJSw=
 github.com/vbatts/tar-split v0.12.3/go.mod h1:sQOc6OlqGCr7HkGx/IDBeKiTIvqhmj8KffNhEXG4Nq0=
 github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=

integration_test.go

@@ -833,7 +833,7 @@ func TestForceTLS(t *testing.T) {
 	pub.LocalParticipant.PublishDataPacket(UserData([]byte("test")), WithDataPublishReliable(true))
 
 	pub.Simulate(SimulateForceTLS)
-	require.Eventually(t, func() bool { return reconnected.Load() && pub.engine.ensurePublisherConnected(true) == nil }, 15*time.Second, 100*time.Millisecond)
+	require.Eventually(t, func() bool { return reconnected.Load() && pub.engine.ensurePublisherConnected(context.Background(), true) == nil }, 15*time.Second, 100*time.Millisecond)
 
 	pub.log.Infow("reconnected")