Add support for SRT in URL Pull ingress (#206)

livekit · Jan 26, 2024 · 1f3ff62 · 1f3ff62
1 parent 8e4bb98
commit 1f3ff62
Show file tree

Hide file tree

Showing 3 changed files with 48 additions and 24 deletions.
diff --git 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
@@ -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
@@ -16,6 +16,7 @@ package urlpull
 
 import (
 	"context"
+	"strings"
 
 	"github.com/go-gst/go-gst/gst"
 
@@ -46,24 +47,42 @@ 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.ErrUnsupportedURLFormat
 	}
 
 	queue, err := gst.NewElement("queue2")
 	if err != nil {
 		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)