Allow SRT output on egress server (#688)

* initial commit, preliminary srt support * use switch/case instead of condition logic --------- Co-authored-by: David Colburn <[email protected]>
livekit · Jul 22, 2024 · 6b16609 · 6b16609
1 parent c644ec9
commit 6b16609
Show file tree

Hide file tree

Showing 6 changed files with 114 additions and 3 deletions.
diff --git a/pkg/config/output.go b/pkg/config/output.go
@@ -15,6 +15,8 @@
 package config
 
 import (
+	"net/url"
+
 	"github.com/livekit/egress/pkg/errors"
 	"github.com/livekit/egress/pkg/types"
 	"github.com/livekit/protocol/egress"
@@ -85,7 +87,22 @@ func (p *PipelineConfig) updateEncodedOutputs(req egress.EncodedOutput) error {
 		return errors.ErrInvalidInput("multiple stream outputs")
 	}
 	if stream != nil {
-		conf, err := p.getStreamConfig(types.OutputTypeRTMP, stream.Urls)
+		u, err := url.Parse(stream.Urls[0])
+		if err != nil {
+			return errors.ErrInvalidInput("malformed url")
+		}
+
+		var outputType types.OutputType
+		switch u.Scheme {
+		case "srt":
+			outputType = types.OutputTypeSRT
+		case "rtmp":
+			outputType = types.OutputTypeRTMP
+		default:
+			return errors.ErrInvalidInput("invalid stream type")
+		}
+
+		conf, err := p.getStreamConfig(outputType, stream.Urls)
 		if err != nil {
 			return err
 		}

diff --git a/pkg/config/output_stream.go b/pkg/config/output_stream.go
@@ -67,6 +67,10 @@ func (p *PipelineConfig) getStreamConfig(outputType types.OutputType, urls []str
 		p.AudioOutCodec = types.MimeTypeAAC
 		p.VideoOutCodec = types.MimeTypeH264
 
+	case types.OutputTypeSRT:
+		p.AudioOutCodec = types.MimeTypeAAC
+		p.VideoOutCodec = types.MimeTypeH264
+
 	case types.OutputTypeRaw:
 		p.AudioOutCodec = types.MimeTypeRawAudio
 	}

diff --git a/pkg/config/pipeline.go b/pkg/config/pipeline.go
@@ -596,6 +596,43 @@ func (p *PipelineConfig) UpdateInfoFromSDK(identifier string, replacements map[s
 	return nil
 }
 
+// func (p *PipelineConfig) ValidateUrl(rawUrl string, outputType types.OutputType) (string, string, error) {
+// 	parsed, err := url.Parse(rawUrl)
+// 	if err != nil {
+// 		return "", "", errors.ErrInvalidUrl(rawUrl, err.Error())
+// 	}
+
+// 	switch outputType {
+// 	case types.OutputTypeRTMP:
+// 		if parsed.Scheme == "mux" {
+// 			rawUrl = fmt.Sprintf("rtmps://global-live.mux.com:443/app/%s", parsed.Host)
+// 		}
+
+// 		redacted, ok := utils.RedactStreamKey(rawUrl)
+// 		if !ok {
+// 			return "", "", errors.ErrInvalidUrl(rawUrl, "rtmp urls must be of format rtmp(s)://{host}(/{path})/{app}/{stream_key}( live=1)")
+// 		}
+// 		return rawUrl, redacted, nil
+
+// 	case types.OutputTypeSRT:
+// 		if parsed.Scheme != "srt" {
+// 			return "", "", errors.ErrInvalidUrl(rawUrl, "invalid scheme")
+// 		}
+// 		// Todo: Optionally, you can redact the SRT stream key or other sensitive parts of the URL
+// 		redacted := rawUrl
+// 		return rawUrl, redacted, nil
+
+// 	case types.OutputTypeRaw:
+// 		if parsed.Scheme != "ws" && parsed.Scheme != "wss" {
+// 			return "", "", errors.ErrInvalidUrl(rawUrl, "invalid scheme")
+// 		}
+// 		return rawUrl, rawUrl, nil
+
+// 	default:
+// 		return "", "", errors.ErrInvalidInput("stream output type")
+// 	}
+// }
+
 func (p *PipelineConfig) GetEncodedOutputs() []OutputConfig {
 	ret := make([]OutputConfig, 0)
 

diff --git a/pkg/pipeline/builder/stream.go b/pkg/pipeline/builder/stream.go
@@ -68,6 +68,13 @@ func BuildStreamBin(pipeline *gstreamer.Pipeline, p *config.PipelineConfig) (*St
 			return nil, nil, errors.ErrGstPipelineError(err)
 		}
 
+	case types.OutputTypeSRT:
+		// add SRT steam
+		mux, err = gst.NewElement("mpegtsmux")
+		if err != nil {
+			return nil, nil, errors.ErrGstPipelineError(err)
+		}
+
 	default:
 		err = errors.ErrInvalidInput("output type")
 	}
@@ -145,6 +152,14 @@ func (sb *StreamBin) AddStream(url string) error {
 		if err = sink.Set("location", url); err != nil {
 			return errors.ErrGstPipelineError(err)
 		}
+	case types.OutputTypeSRT:
+		sink, err = gst.NewElementWithName("srtsink", fmt.Sprintf("srtsink_%s", name))
+		if err != nil {
+			return errors.ErrGstPipelineError(err)
+		}
+		if err = sink.SetProperty("uri", url); err != nil {
+			return errors.ErrGstPipelineError(err)
+		}
 
 	default:
 		return errors.ErrInvalidInput("output type")
@@ -161,7 +176,7 @@ func (sb *StreamBin) AddStream(url string) error {
 		// It is later released in RemoveSink
 		proxy.Ref()
 
-		// Intercept flows from rtmp2sink. Anything besides EOS will be ignored
+		// Intercept flows from the sink. Anything besides EOS will be ignored
 		proxy.SetChainFunction(func(self *gst.Pad, _ *gst.Object, buffer *gst.Buffer) gst.FlowReturn {
 			// Buffer gets automatically unreferenced by go-gst.
 			// Without referencing it here, it will sometimes be garbage collected before being written

diff --git a/pkg/pipeline/controller.go b/pkg/pipeline/controller.go
@@ -16,6 +16,7 @@ package pipeline
 
 import (
 	"context"
+	"net/url"
 	"sync"
 	"time"
 
@@ -263,8 +264,23 @@ func (c *Controller) UpdateStream(ctx context.Context, req *livekit.UpdateStream
 
 	// add stream outputs first
 	for _, rawUrl := range req.AddOutputUrls {
+		u, err := url.Parse(rawUrl)
+		if err != nil {
+			return errors.ErrInvalidInput("malformed url")
+		}
+
+		var outputType types.OutputType
+		switch u.Scheme {
+		case "srt":
+			outputType = types.OutputTypeSRT
+		case "rtmp":
+			outputType = types.OutputTypeRTMP
+		default:
+			return errors.ErrInvalidInput("invalid stream type")
+		}
+
 		// validate and redact url
-		url, redacted, err := config.ValidateUrl(rawUrl, types.OutputTypeRTMP)
+		url, redacted, err := config.ValidateUrl(rawUrl, outputType)
 		if err != nil {
 			errs.AppendErr(err)
 			continue
@@ -298,6 +314,21 @@ func (c *Controller) UpdateStream(ctx context.Context, req *livekit.UpdateStream
 
 	// remove stream outputs
 	for _, rawUrl := range req.RemoveOutputUrls {
+		u, err := url.Parse(rawUrl)
+		if err != nil {
+			return errors.ErrInvalidInput("malformed url")
+		}
+
+		var outputType types.OutputType
+		switch u.Scheme {
+		case "srt":
+			outputType = types.OutputTypeSRT
+		case "rtmp":
+			outputType = types.OutputTypeRTMP
+		default:
+			return errors.ErrInvalidInput("invalid stream type")
+		}
+
 		url, err := o.GetStreamUrl(rawUrl)
 		if err != nil {
 			errs.AppendErr(err)

diff --git a/pkg/types/types.go b/pkg/types/types.go
@@ -66,6 +66,7 @@ const (
 	OutputTypeWebM        OutputType = "video/webm"
 	OutputTypeJPEG        OutputType = "image/jpeg"
 	OutputTypeRTMP        OutputType = "rtmp"
+	OutputTypeSRT         OutputType = "srt"
 	OutputTypeHLS         OutputType = "application/x-mpegurl"
 	OutputTypeJSON        OutputType = "application/json"
 	OutputTypeBlob        OutputType = "application/octet-stream"
@@ -89,6 +90,7 @@ var (
 		OutputTypeTS:   MimeTypeAAC,
 		OutputTypeWebM: MimeTypeOpus,
 		OutputTypeRTMP: MimeTypeAAC,
+		OutputTypeSRT:  MimeTypeAAC,
 		OutputTypeHLS:  MimeTypeAAC,
 	}
 
@@ -98,6 +100,7 @@ var (
 		OutputTypeTS:   MimeTypeH264,
 		OutputTypeWebM: MimeTypeVP8,
 		OutputTypeRTMP: MimeTypeH264,
+		OutputTypeSRT:  MimeTypeH264,
 		OutputTypeHLS:  MimeTypeH264,
 	}
 
@@ -153,6 +156,10 @@ var (
 			MimeTypeAAC:  true,
 			MimeTypeH264: true,
 		},
+		OutputTypeSRT: {
+			MimeTypeAAC:  true,
+			MimeTypeH264: true,
+		},
 		OutputTypeHLS: {
 			MimeTypeAAC:  true,
 			MimeTypeH264: true,