sdk compositing

pkg/config/base.go

@@ -64,6 +64,7 @@ type BaseConfig struct {
 	ChromeFlags                   map[string]interface{}              `yaml:"chrome_flags"`                       // additional flags to pass to Chrome
 	Latency                       LatencyConfig                       `yaml:"latency"`                            // gstreamer latencies, modifying these may break the service
 	LatencyOverrides              map[types.RequestType]LatencyConfig `yaml:"latency_overrides"`                  // latency overrides for different request types, experimental only, will be removed
+	EnableTemplateSDK             bool                                `yaml:"enable_template_sdk"`                // use GStreamer compositor instead of Chrome for default template layouts
 	EnableOneShotSenderReportSync bool                                `yaml:"enable_one_shot_sender_report_sync"` // temporary rollout flag enabling one-shot sender report correction for room composite / track requests that previously used audio PTS adjustment disabling
 	EnableSyncEngine              bool                                `yaml:"enable_sync_engine"`                 // use Chrome-inspired sync engine for improved cross-participant alignment and A/V sync
 	AudioTempoController          AudioTempoController                `yaml:"audio_tempo_controller"`             // audio tempo controller

pkg/config/pipeline.go

@@ -93,9 +93,10 @@ type SDKSourceParams struct {
 	TrackSource  string
 	TrackKind    string
 	ScreenShare  bool
+	Compositing  bool
 	VideoInCodec types.MimeType
 	AudioTracks  []*TrackSource
-	VideoTrack   *TrackSource
+	VideoTracks  []*TrackSource
 	AudioRoutes  []AudioRouteConfig
 }
 
@@ -111,16 +112,18 @@ type AudioRouteMatch struct {
 }
 
 type TrackSource struct {
-	TrackID            string
-	TrackKind          lksdk.TrackKind
-	ParticipantKind    lksdk.ParticipantKind
-	AudioChannel       *livekit.AudioChannel
-	AppSrc             *app.Source
-	MimeType           types.MimeType
-	PayloadType        webrtc.PayloadType
-	ClockRate          uint32
-	TempoController    *tempo.Controller
-	OnKeyframeRequired func()
+	TrackID             string
+	TrackKind           lksdk.TrackKind
+	ParticipantIdentity string
+	PublicationSource   livekit.TrackSource
+	ParticipantKind     lksdk.ParticipantKind
+	AudioChannel        *livekit.AudioChannel
+	AppSrc              *app.Source
+	MimeType            types.MimeType
+	PayloadType         webrtc.PayloadType
+	ClockRate           uint32
+	TempoController     *tempo.Controller
+	OnKeyframeRequired  func()
 }
 
 type AudioConfig struct {
@@ -227,7 +230,11 @@ func (p *PipelineConfig) Update(request *rpc.StartEgressRequest) error {
 		}
 		egress.RedactEncodedOutputs(clone)
 
-		if ShouldUseSDKSource(req.RoomComposite) {
+		if p.EnableTemplateSDK && req.RoomComposite.CustomBaseUrl == "" {
+			p.AudioMixing = req.RoomComposite.AudioMixing
+			p.SourceType = types.SourceTypeSDK
+			p.Compositing = !req.RoomComposite.AudioOnly
+		} else if ShouldUseSDKSource(req.RoomComposite) {
 			p.AudioMixing = req.RoomComposite.AudioMixing
 			p.SourceType = types.SourceTypeSDK
 		} else {
@@ -435,8 +442,10 @@ func (p *PipelineConfig) Update(request *rpc.StartEgressRequest) error {
 		case *livekit.ExportReplayRequest_Template:
 			tmpl := source.Template
 			p.RequestType = types.RequestTypeTemplate
-
-			if ShouldUseSDKSource(tmpl) {
+			if p.EnableTemplateSDK && tmpl.CustomBaseUrl == "" {
+				p.SourceType = types.SourceTypeSDK
+				p.Compositing = !tmpl.AudioOnly
+			} else if ShouldUseSDKSource(tmpl) {
 				p.SourceType = types.SourceTypeSDK
 			} else {
 				p.SourceType = types.SourceTypeWeb

pkg/gstreamer/callbacks.go

@@ -19,6 +19,7 @@ import (
 	"github.com/linkdata/deadlock"
 	"github.com/livekit/egress/pkg/config"
 	"github.com/livekit/egress/pkg/errors"
+	lksdk "github.com/livekit/server-sdk-go/v2"
 )
 
 type Callbacks struct {
@@ -36,8 +37,9 @@ type Callbacks struct {
 	onTrackMuted     []func(string)
 	onTrackUnmuted   []func(string)
 	onTrackRemoved   []func(string)
-	onSourceBinReset []func(*config.TrackSource) error
-	onEOSSent        func()
+	onSourceBinReset        []func(*config.TrackSource) error
+	onActiveSpeakersChanged []func([]lksdk.Participant)
+	onEOSSent               func()
 
 	pipelinePaused core.Fuse
 }
@@ -187,6 +189,22 @@ func (c *Callbacks) OnSourceBinReset(ts *config.TrackSource) error {
 	return nil
 }
 
+func (c *Callbacks) AddOnActiveSpeakersChanged(f func([]lksdk.Participant)) {
+	c.mu.Lock()
+	c.onActiveSpeakersChanged = append(c.onActiveSpeakersChanged, f)
+	c.mu.Unlock()
+}
+
+func (c *Callbacks) OnActiveSpeakersChanged(speakers []lksdk.Participant) {
+	c.mu.RLock()
+	handlers := c.onActiveSpeakersChanged
+	c.mu.RUnlock()
+
+	for _, f := range handlers {
+		f(speakers)
+	}
+}
+
 func (c *Callbacks) SetOnEOSSent(f func()) {
 	c.mu.Lock()
 	c.onEOSSent = f

pkg/pipeline/builder/image.go

@@ -122,6 +122,13 @@ func BuildImageBin(c *config.ImageConfig, pipeline *gstreamer.Pipeline, p *confi
 	if err != nil {
 		return nil, err
 	}
+	// async=true here would gate pipeline preroll on the first composited frame, stalling PAUSED→PLAYING.
+	if err = sink.SetProperty("sync", false); err != nil {
+		return nil, err
+	}
+	if err = sink.SetProperty("async", false); err != nil {
+		return nil, err
+	}
 
 	// File will be renamed if the TS prefix is configured
 	location := fmt.Sprintf("%s_%%05d%s", path.Join(c.LocalDir, c.ImagePrefix), types.FileExtensionForOutputType[c.OutputType])