From 8099000fab49b0a666432700388b83df030a541d Mon Sep 17 00:00:00 2001
From: Benjamin Pracht <benjamin@livekit.io>
Date: Fri, 26 Jan 2024 13:29:02 -0800
Subject: [PATCH 1/2] Add support for SRT in URL Pull ingress

---
 pkg/errors/errors.go        |  1 +
 pkg/media/input.go          | 32 +++++++++++++++++-------------
 pkg/media/urlpull/source.go | 39 +++++++++++++++++++++++++++----------
 3 files changed, 48 insertions(+), 24 deletions(-)

diff --git a/pkg/errors/errors.go b/pkg/errors/errors.go
index 668610f..7e10bcf 100644
--- a/pkg/errors/errors.go
+++ b/pkg/errors/errors.go
@@ -32,6 +32,7 @@ var (
 	ErrSourceNotReady          = psrpc.NewErrorf(psrpc.FailedPrecondition, "source encoder not ready")
 	ErrUnsupportedDecodeFormat = psrpc.NewErrorf(psrpc.NotAcceptable, "unsupported format for the source media")
 	ErrUnsupportedEncodeFormat = psrpc.NewErrorf(psrpc.InvalidArgument, "unsupported mime type for encoder")
+	ErrUnsupportedURLFormat    = psrpc.NewErrorf(psrpc.InvalidArgument, "unsupported URL type")
 	ErrDuplicateTrack          = psrpc.NewErrorf(psrpc.NotAcceptable, "more than 1 track with given media kind")
 	ErrUnableToAddPad          = psrpc.NewErrorf(psrpc.Internal, "could not add pads to bin")
 	ErrMissingResourceId       = psrpc.NewErrorf(psrpc.InvalidArgument, "missing resource ID")
diff --git a/pkg/media/input.go b/pkg/media/input.go
index a09e2f3..66c9062 100644
--- a/pkg/media/input.go
+++ b/pkg/media/input.go
@@ -232,25 +232,29 @@ func (i *Input) addBitrateProbe(kind types.StreamKind) {
 
 	for _, pad := range pads {
 		caps := pad.GetCurrentCaps()
-		gstStruct := caps.GetStructureAt(0)
-		padKind := getKindFromGstMimeType(gstStruct)
+		if caps != nil && caps.GetSize() > 0 {
+			gstStruct := caps.GetStructureAt(0)
+			padKind := getKindFromGstMimeType(gstStruct)
 
-		if padKind == kind {
-			g := i.trackStatsGatherer[kind]
+			if padKind == kind {
+				g := i.trackStatsGatherer[kind]
 
-			pad.AddProbe(gst.PadProbeTypeBuffer, func(pad *gst.Pad, info *gst.PadProbeInfo) gst.PadProbeReturn {
-				buffer := info.GetBuffer()
-				if buffer == nil {
-					return gst.PadProbeOK
-				}
+				pad.AddProbe(gst.PadProbeTypeBuffer, func(pad *gst.Pad, info *gst.PadProbeInfo) gst.PadProbeReturn {
+					buffer := info.GetBuffer()
+					if buffer == nil {
+						return gst.PadProbeOK
+					}
 
-				size := buffer.GetSize()
-				g.MediaReceived(size)
+					size := buffer.GetSize()
+					g.MediaReceived(size)
 
-				return gst.PadProbeOK
-			})
+					return gst.PadProbeOK
+				})
 
-			return
+				return
+			}
+		} else {
+			logger.Debugw("could not retrieve multiqueue pad caps", "error", err)
 		}
 	}
 
diff --git a/pkg/media/urlpull/source.go b/pkg/media/urlpull/source.go
index c4fe90d..f2a0b18 100644
--- a/pkg/media/urlpull/source.go
+++ b/pkg/media/urlpull/source.go
@@ -16,6 +16,7 @@ package urlpull
 
 import (
 	"context"
+	"strings"
 
 	"github.com/go-gst/go-gst/gst"
 
@@ -46,14 +47,30 @@ type URLSource struct {
 func NewURLSource(ctx context.Context, p *params.Params) (*URLSource, error) {
 	bin := gst.NewBin("input")
 
-	elem, err := gst.NewElement("souphttpsrc")
-	if err != nil {
-		return nil, err
-	}
+	var elem *gst.Element
+	var err error
+	if strings.HasPrefix(p.Url, "http://") || strings.HasPrefix(p.Url, "https://") {
+		elem, err = gst.NewElement("souphttpsrc")
+		if err != nil {
+			return nil, err
+		}
 
-	err = elem.SetProperty("location", p.Url)
-	if err != nil {
-		return nil, err
+		err = elem.SetProperty("location", p.Url)
+		if err != nil {
+			return nil, err
+		}
+
+	} else if strings.HasPrefix(p.Url, "srt://") {
+		elem, err = gst.NewElement("srtclientsrc")
+		if err != nil {
+			return nil, err
+		}
+		err = elem.SetProperty("uri", p.Url)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		return nil, errors.ErrUnsupportedEncodeFormat
 	}
 
 	queue, err := gst.NewElement("queue2")
@@ -61,9 +78,11 @@ func NewURLSource(ctx context.Context, p *params.Params) (*URLSource, error) {
 		return nil, err
 	}
 
-	err = queue.SetProperty("use-buffering", true)
-	if err != nil {
-		return nil, err
+	if strings.HasPrefix(p.Url, "http://") || strings.HasPrefix(p.Url, "https://") {
+		err = queue.SetProperty("use-buffering", true)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	err = bin.AddMany(elem, queue)

From 9791c4adb02b32812a92e60f65d517f2e8bb019b Mon Sep 17 00:00:00 2001
From: Benjamin Pracht <benjamin@livekit.io>
Date: Fri, 26 Jan 2024 14:05:20 -0800
Subject: [PATCH 2/2] Correct error

---
 pkg/media/urlpull/source.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pkg/media/urlpull/source.go b/pkg/media/urlpull/source.go
index f2a0b18..9999d99 100644
--- a/pkg/media/urlpull/source.go
+++ b/pkg/media/urlpull/source.go
@@ -70,7 +70,7 @@ func NewURLSource(ctx context.Context, p *params.Params) (*URLSource, error) {
 			return nil, err
 		}
 	} else {
-		return nil, errors.ErrUnsupportedEncodeFormat
+		return nil, errors.ErrUnsupportedURLFormat
 	}
 
 	queue, err := gst.NewElement("queue2")