From 21dcd9efac90e5c37abf44d4451a4e863abc127a Mon Sep 17 00:00:00 2001
From: David Colburn <xero73@gmail.com>
Date: Tue, 26 Mar 2024 16:26:46 -0400
Subject: [PATCH] don't link audio to image bins

---
 pkg/gstreamer/bin.go          | 12 ++++++++++++
 pkg/pipeline/builder/image.go |  3 +++
 test/multi.go                 |  5 ++++-
 test/participant.go           |  2 +-
 test/room_composite.go        |  9 ++++++---
 test/track_composite.go       |  2 +-
 test/web.go                   |  2 +-
 7 files changed, 28 insertions(+), 7 deletions(-)

diff --git a/pkg/gstreamer/bin.go b/pkg/gstreamer/bin.go
index 25f25375..526dea1d 100644
--- a/pkg/gstreamer/bin.go
+++ b/pkg/gstreamer/bin.go
@@ -37,6 +37,7 @@ type Bin struct {
 	latency  uint64
 
 	linkFunc   func() error
+	shouldLink func(string) bool
 	eosFunc    func() bool
 	getSrcPad  func(string) *gst.Pad
 	getSinkPad func(string) *gst.Pad
@@ -317,6 +318,13 @@ func (b *Bin) SetLinkFunc(f func() error) {
 	b.linkFunc = f
 }
 
+func (b *Bin) SetShouldLink(f func(string) bool) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	b.shouldLink = f
+}
+
 // Set a custom linking function which returns a pad for the named src bin
 func (b *Bin) SetGetSrcPad(f func(srcName string) *gst.Pad) {
 	b.mu.Lock()
@@ -484,6 +492,10 @@ func (b *Bin) queueLinkPeersLocked(src, sink *Bin) error {
 	srcName := src.bin.GetName()
 	sinkName := sink.bin.GetName()
 
+	if (src.shouldLink != nil && !src.shouldLink(sinkName)) || (sink.shouldLink != nil && !sink.shouldLink(srcName)) {
+		return nil
+	}
+
 	queueName := fmt.Sprintf("%s_%s_queue", srcName, sinkName)
 	queue, err := BuildQueue(queueName, b.latency, true)
 	if err != nil {
diff --git a/pkg/pipeline/builder/image.go b/pkg/pipeline/builder/image.go
index f666e789..9eb4c678 100644
--- a/pkg/pipeline/builder/image.go
+++ b/pkg/pipeline/builder/image.go
@@ -74,6 +74,9 @@ func BuildImageBin(c *config.ImageConfig, pipeline *gstreamer.Pipeline, p *confi
 			return queue.GetStaticPad("sink")
 		}
 	})
+	b.SetShouldLink(func(srcBin string) bool {
+		return srcBin != "audio"
+	})
 
 	videoRate, err := gst.NewElement("videorate")
 	if err != nil {
diff --git a/test/multi.go b/test/multi.go
index ade90e27..05380e36 100644
--- a/test/multi.go
+++ b/test/multi.go
@@ -31,7 +31,7 @@ import (
 func (r *Runner) runMultipleTest(
 	t *testing.T,
 	req *rpc.StartEgressRequest,
-	file, stream, segments bool,
+	file, stream, segments, images bool,
 	filenameSuffix livekit.SegmentedFileSuffix,
 ) {
 	egressID := r.startEgress(t, req)
@@ -65,4 +65,7 @@ func (r *Runner) runMultipleTest(
 	if segments {
 		r.verifySegments(t, p, filenameSuffix, res, false)
 	}
+	if images {
+		r.verifyImages(t, p, 0, res)
+	}
 }
diff --git a/test/participant.go b/test/participant.go
index cea9be54..1677a9c0 100644
--- a/test/participant.go
+++ b/test/participant.go
@@ -265,7 +265,7 @@ func (r *Runner) testParticipantMulti(t *testing.T) {
 				},
 			}
 
-			r.runMultipleTest(t, req, true, true, false, livekit.SegmentedFileSuffix_INDEX)
+			r.runMultipleTest(t, req, true, true, false, false, livekit.SegmentedFileSuffix_INDEX)
 		},
 	)
 
diff --git a/test/room_composite.go b/test/room_composite.go
index ba483975..07c5f77a 100644
--- a/test/room_composite.go
+++ b/test/room_composite.go
@@ -319,13 +319,16 @@ func (r *Runner) testRoomCompositeMulti(t *testing.T) {
 						FileType: livekit.EncodedFileType_MP4,
 						Filepath: r.getFilePath("rc_multiple_{time}"),
 					}},
-					StreamOutputs: []*livekit.StreamOutput{{
-						Protocol: livekit.StreamProtocol_RTMP,
+					ImageOutputs: []*livekit.ImageOutput{{
+						CaptureInterval: 10,
+						Width:           1280,
+						Height:          720,
+						FilenamePrefix:  r.getFilePath("rc_image"),
 					}},
 				},
 			},
 		}
 
-		r.runMultipleTest(t, req, true, true, false, livekit.SegmentedFileSuffix_TIMESTAMP)
+		r.runMultipleTest(t, req, true, false, false, true, livekit.SegmentedFileSuffix_TIMESTAMP)
 	})
 }
diff --git a/test/track_composite.go b/test/track_composite.go
index 9a275bd2..6bb8aa64 100644
--- a/test/track_composite.go
+++ b/test/track_composite.go
@@ -315,7 +315,7 @@ func (r *Runner) testTrackCompositeMulti(t *testing.T) {
 				},
 			}
 
-			r.runMultipleTest(t, req, false, true, true, livekit.SegmentedFileSuffix_INDEX)
+			r.runMultipleTest(t, req, false, true, true, false, livekit.SegmentedFileSuffix_INDEX)
 		},
 	)
 }
diff --git a/test/web.go b/test/web.go
index 73f564ad..d5ba8440 100644
--- a/test/web.go
+++ b/test/web.go
@@ -158,6 +158,6 @@ func (r *Runner) testWebMulti(t *testing.T) {
 			},
 		}
 
-		r.runMultipleTest(t, req, true, false, true, livekit.SegmentedFileSuffix_INDEX)
+		r.runMultipleTest(t, req, true, false, true, false, livekit.SegmentedFileSuffix_INDEX)
 	})
 }