From 764bd136c1cfaedbd731d1e5c684d2462cc3e964 Mon Sep 17 00:00:00 2001
From: David Colburn <xero73@gmail.com>
Date: Mon, 15 Jan 2024 23:20:00 -0800
Subject: [PATCH] throttle PLIs (#585)

---
 go.mod                               |  4 ++--
 go.sum                               |  8 ++++----
 pkg/pipeline/source/sdk/appwriter.go | 17 ++++++++++-------
 3 files changed, 16 insertions(+), 13 deletions(-)

diff --git a/go.mod b/go.mod
index 514ba640..5c98fd6b 100644
--- a/go.mod
+++ b/go.mod
@@ -21,9 +21,9 @@ require (
 	github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
 	github.com/livekit/protocol v1.9.4-0.20231207062609-297e4990e7d3
 	github.com/livekit/psrpc v0.5.2
-	github.com/livekit/server-sdk-go v1.1.5-0.20231214205747-54bbfd63ef52
+	github.com/livekit/server-sdk-go v1.1.5-0.20240116053441-167f58a9a787
 	github.com/pion/rtp v1.8.3
-	github.com/pion/webrtc/v3 v3.2.23
+	github.com/pion/webrtc/v3 v3.2.24
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.17.0
 	github.com/prometheus/client_model v0.4.1-0.20230718164431-9a2bf3000d16
diff --git a/go.sum b/go.sum
index ac6259c7..fb931532 100644
--- a/go.sum
+++ b/go.sum
@@ -184,8 +184,8 @@ github.com/livekit/protocol v1.9.4-0.20231207062609-297e4990e7d3 h1:1Au/odnRyEe7
 github.com/livekit/protocol v1.9.4-0.20231207062609-297e4990e7d3/go.mod h1:EKF+U1kf2Fh7wG2xLET2dNDVxi7Q+6nHN6uYs5NsIVY=
 github.com/livekit/psrpc v0.5.2 h1:+MvG8Otm/J6MTg2MP/uuMbrkxOWsrj2hDhu/I1VIU1U=
 github.com/livekit/psrpc v0.5.2/go.mod h1:cQjxg1oCxYHhxxv6KJH1gSvdtCHQoRZCHgPdm5N8v2g=
-github.com/livekit/server-sdk-go v1.1.5-0.20231214205747-54bbfd63ef52 h1:x4wfnG3P/2QgC5PY7HnLyxVzKHRSN5xwuu19DX9L0Y0=
-github.com/livekit/server-sdk-go v1.1.5-0.20231214205747-54bbfd63ef52/go.mod h1:xfNiF6LjyG4nLT58cU/K9ZPuBB6Bgtyiy6Zn+VNZY70=
+github.com/livekit/server-sdk-go v1.1.5-0.20240116053441-167f58a9a787 h1:nYF73ZmDd29QlsKUaxxCAAAQHAMZ3duY9DFuqtcrdt8=
+github.com/livekit/server-sdk-go v1.1.5-0.20240116053441-167f58a9a787/go.mod h1:L3Oc1kc4pHcvyfz2S/McEguPwaC5ZAATbhoWz3gQmeQ=
 github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs=
 github.com/mackerelio/go-osstat v0.2.4/go.mod h1:Zy+qzGdZs3A9cuIqmgbJvwbmLQH9dJvtio5ZjJTbdlQ=
 github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg=
@@ -259,8 +259,8 @@ github.com/pion/transport/v3 v3.0.1/go.mod h1:UY7kiITrlMv7/IKgd5eTUcaahZx5oUN3l9
 github.com/pion/turn/v2 v2.1.3/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
 github.com/pion/turn/v2 v2.1.4 h1:2xn8rduI5W6sCZQkEnIUDAkrBQNl2eYIBCHMZ3QMmP8=
 github.com/pion/turn/v2 v2.1.4/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
-github.com/pion/webrtc/v3 v3.2.23 h1:GbqEuxBbVLFhXk0GwxKAoaIJYiEa9TyoZPEZC+2HZxM=
-github.com/pion/webrtc/v3 v3.2.23/go.mod h1:1CaT2fcZzZ6VZA+O1i9yK2DU4EOcXVvSbWG9pr5jefs=
+github.com/pion/webrtc/v3 v3.2.24 h1:MiFL5DMo2bDaaIFWr0DDpwiV/L4EGbLZb+xoRvfEo1Y=
+github.com/pion/webrtc/v3 v3.2.24/go.mod h1:1CaT2fcZzZ6VZA+O1i9yK2DU4EOcXVvSbWG9pr5jefs=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
diff --git a/pkg/pipeline/source/sdk/appwriter.go b/pkg/pipeline/source/sdk/appwriter.go
index a3fb96d4..1b2612ec 100644
--- a/pkg/pipeline/source/sdk/appwriter.go
+++ b/pkg/pipeline/source/sdk/appwriter.go
@@ -63,10 +63,11 @@ type AppWriter struct {
 	src       *app.Source
 	startTime time.Time
 
-	buffer     *jitter.Buffer
-	translator Translator
-	callbacks  *gstreamer.Callbacks
-	sendPLI    func()
+	buffer      *jitter.Buffer
+	translator  Translator
+	callbacks   *gstreamer.Callbacks
+	sendPLI     func()
+	pliThrottle core.Throttle
 
 	// a/v sync
 	sync *synchronizer.Synchronizer
@@ -126,22 +127,24 @@ func NewAppWriter(
 	case types.MimeTypeH264:
 		depacketizer = &codecs.H264Packet{}
 		w.translator = NewNullTranslator()
-		w.sendPLI = func() { rp.WritePLI(track.SSRC()) }
 
 	case types.MimeTypeVP8:
 		depacketizer = &codecs.VP8Packet{}
 		w.translator = NewVP8Translator(w.logger)
-		w.sendPLI = func() { rp.WritePLI(track.SSRC()) }
 
 	case types.MimeTypeVP9:
 		depacketizer = &codecs.VP9Packet{}
 		w.translator = NewNullTranslator()
-		w.sendPLI = func() { rp.WritePLI(track.SSRC()) }
 
 	default:
 		return nil, errors.ErrNotSupported(string(ts.MimeType))
 	}
 
+	if track.Kind() == webrtc.RTPCodecTypeVideo {
+		w.pliThrottle = core.NewThrottle(time.Second)
+		w.sendPLI = func() { w.pliThrottle(func() { rp.WritePLI(track.SSRC()) }) }
+	}
+
 	w.buffer = jitter.NewBuffer(
 		depacketizer,
 		ts.ClockRate,