Allow SRT output on egress server

pkg/config/output.go

@@ -15,6 +15,8 @@
 package config
 
 import (
+	"strings"
+
 	"github.com/livekit/egress/pkg/errors"
 	"github.com/livekit/egress/pkg/types"
 	"github.com/livekit/protocol/egress"
@@ -84,7 +86,12 @@ 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)
+		outputType := types.OutputTypeRTMP
+		if len(stream.Urls) > 0 && strings.HasPrefix(stream.Urls[0], "srt://") {
+			outputType = types.OutputTypeSRT
+		}
+
+		conf, err := p.getStreamConfig(outputType, stream.Urls)
 		if err != nil {
 			return err
 		}

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
 	}

pkg/config/pipeline.go

@@ -615,6 +615,14 @@ func (p *PipelineConfig) ValidateUrl(rawUrl string, outputType types.OutputType)
 		}
 		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")

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

pkg/pipeline/controller.go

@@ -16,6 +16,7 @@ package pipeline
 
 import (
 	"context"
+	"strings"
 	"sync"
 	"time"
 
@@ -263,8 +264,13 @@ func (c *Controller) UpdateStream(ctx context.Context, req *livekit.UpdateStream
 
 	// add stream outputs first
 	for _, rawUrl := range req.AddOutputUrls {
+		outputType := types.OutputTypeRTMP
+		if strings.HasPrefix(rawUrl, "srt://") {
+			outputType = types.OutputTypeSRT
+		}
 		// validate and redact url
-		url, redacted, err := c.ValidateUrl(rawUrl, types.OutputTypeRTMP)
+		//url, redacted, err := c.ValidateUrl(rawUrl, types.OutputTypeRTMP)
+		url, redacted, err := c.ValidateUrl(rawUrl, outputType)
 		if err != nil {
 			errs.AppendErr(err)
 			continue
@@ -297,7 +303,12 @@ func (c *Controller) UpdateStream(ctx context.Context, req *livekit.UpdateStream
 
 	// remove stream outputs
 	for _, rawUrl := range req.RemoveOutputUrls {
-		url, _, err := c.ValidateUrl(rawUrl, types.OutputTypeRTMP)
+		outputType := types.OutputTypeRTMP
+		if strings.HasPrefix(rawUrl, "srt://") {
+			outputType = types.OutputTypeSRT
+		}
+		url, _, err := c.ValidateUrl(rawUrl, outputType)
+		//url, _, err := c.ValidateUrl(rawUrl, types.OutputTypeRTMP)
 		if err != nil {
 			errs.AppendErr(err)
 			continue

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,