Disable max buffer linit on URL input queue. Log SRT stats every minute (#335)

Author: biglittlebigben

pkg/media/urlpull/source.go

@@ -17,11 +17,14 @@ package urlpull
 import (
 	"context"
 	"strings"
+	"time"
 
+	"github.com/frostbyte73/core"
 	"github.com/go-gst/go-gst/gst"
 
 	"github.com/livekit/ingress/pkg/errors"
 	"github.com/livekit/ingress/pkg/params"
+	"github.com/livekit/protocol/logger"
 )
 
 var (
@@ -39,12 +42,16 @@ var (
 )
 
 type URLSource struct {
-	params *params.Params
-	src    *gst.Element
-	pad    *gst.Pad
+	params     *params.Params
+	src        *gst.Element
+	pad        *gst.Pad
+	printStats func()
+
+	done core.Fuse
 }
 
 func NewURLSource(ctx context.Context, p *params.Params) (*URLSource, error) {
+	var printStats func()
 	bin := gst.NewBin("input")
 
 	var elem *gst.Element
@@ -69,6 +76,15 @@ func NewURLSource(ctx context.Context, p *params.Params) (*URLSource, error) {
 		if err != nil {
 			return nil, err
 		}
+
+		printStats = func() {
+			str, _ := elem.GetProperty("stats")
+			if str != nil {
+				if v, ok := str.(*gst.Structure); ok {
+					logger.Infow("SRT input stats", "stats", v.String())
+				}
+			}
+		}
 	} else {
 		return nil, errors.ErrUnsupportedURLFormat
 	}
@@ -78,6 +94,11 @@ func NewURLSource(ctx context.Context, p *params.Params) (*URLSource, error) {
 		return nil, err
 	}
 
+	// Disable buffer count limit and rely on bytes and time limits
+	if err := queue.SetProperty("max-size-buffers", uint(0)); 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 {
@@ -106,9 +127,10 @@ func NewURLSource(ctx context.Context, p *params.Params) (*URLSource, error) {
 	}
 
 	return &URLSource{
-		params: p,
-		src:    bin.Element,
-		pad:    pad,
+		params:     p,
+		src:        bin.Element,
+		pad:        pad,
+		printStats: printStats,
 	}, nil
 }
 
@@ -138,11 +160,30 @@ func (s *URLSource) ValidateCaps(caps *gst.Caps) error {
 }
 
 func (u *URLSource) Start(ctx context.Context) error {
+	if u.printStats == nil {
+		return nil
+	}
+
+	go func() {
+		ticker := time.NewTicker(time.Minute)
+		for {
+			select {
+			case <-u.done.Watch():
+				ticker.Stop()
+				return
+			case <-ticker.C:
+				u.printStats()
+			}
+		}
+	}()
+
 	return nil
 }
 
 func (u *URLSource) Close() error {
 	// TODO find a way to send a EOS event without hanging
 
+	u.done.Break()
+
 	return nil
 }