rtsp: add rtspScale parameter to inject Scale header on PLAY (#5800)

Co-authored-by: aler9 <46489434+aler9@users.noreply.github.com>

api/openapi.yaml

@@ -782,6 +782,8 @@ components:
           $ref: "#/components/schemas/RTSPRangeType"
         rtspRangeStart:
           type: string
+        rtspScale:
+          type: string
         rtspUDPReadBufferSize:
           type: integer
           format: uint64

internal/conf/path.go

@@ -257,6 +257,7 @@ type Path struct {
 	SourceAnyPortEnable    *bool          `json:"sourceAnyPortEnable,omitempty" deprecated:"true"`
 	RTSPRangeType          RTSPRangeType  `json:"rtspRangeType"`
 	RTSPRangeStart         string         `json:"rtspRangeStart"`
+	RTSPScale              string         `json:"rtspScale"`
 	RTSPUDPReadBufferSize  *uint          `json:"rtspUDPReadBufferSize,omitempty" deprecated:"true"`
 	RTSPUDPSourcePortRange []uint         `json:"rtspUDPSourcePortRange"`
 

internal/staticsources/rtsp/source.go

@@ -139,6 +139,12 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
 			uint16(params.Conf.RTSPUDPSourcePortRange[1]),
 		},
 		OnRequest: func(req *base.Request) {
+			if params.Conf.RTSPScale != "" && req.Method == base.Play {
+				if req.Header == nil {
+					req.Header = base.Header{}
+				}
+				req.Header["Scale"] = base.HeaderValue{params.Conf.RTSPScale}
+			}
 			s.Log(logger.Debug, "[c->s] %v", req)
 		},
 		OnResponse: func(res *base.Response) {

internal/staticsources/rtsp/source_test.go

@@ -305,6 +305,96 @@ func TestNoPassword(t *testing.T) {
 	<-p.Unit
 }
 
+func TestScale(t *testing.T) {
+	var strm *gortsplib.ServerStream
+
+	media0 := test.UniqueMediaH264()
+
+	s := gortsplib.Server{
+		Handler: &testServer{
+			onDescribe: func(_ *gortsplib.ServerHandlerOnDescribeCtx) (*base.Response, *gortsplib.ServerStream, error) {
+				return &base.Response{
+					StatusCode: base.StatusOK,
+				}, strm, nil
+			},
+			onSetup: func(_ *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error) {
+				return &base.Response{
+					StatusCode: base.StatusOK,
+				}, strm, nil
+			},
+			onPlay: func(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) {
+				require.Equal(t, base.HeaderValue{"-1.0"}, ctx.Request.Header["Scale"])
+
+				go func() {
+					time.Sleep(100 * time.Millisecond)
+					err := strm.WritePacketRTP(media0, &rtp.Packet{
+						Header: rtp.Header{
+							Version:        0x02,
+							PayloadType:    96,
+							SequenceNumber: 57899,
+							Timestamp:      345234345,
+							SSRC:           978651231,
+							Marker:         true,
+						},
+						Payload: []byte{5, 1, 2, 3, 4},
+					})
+					require.NoError(t, err)
+				}()
+
+				return &base.Response{
+					StatusCode: base.StatusOK,
+				}, nil
+			},
+		},
+		RTSPAddress: "127.0.0.1:8555",
+	}
+
+	err := s.Start()
+	require.NoError(t, err)
+	defer s.Close()
+
+	strm = &gortsplib.ServerStream{
+		Server: &s,
+		Desc:   &description.Session{Medias: []*description.Media{media0}},
+	}
+	err = strm.Initialize()
+	require.NoError(t, err)
+	defer strm.Close()
+
+	cnf := &conf.Path{
+		RTSPUDPSourcePortRange: []uint{10000, 65535},
+		RTSPScale:              "-1.0",
+	}
+
+	p := &test.StaticSourceParent{}
+	p.Initialize()
+	defer p.Close()
+
+	so := &Source{
+		ReadTimeout:    conf.Duration(10 * time.Second),
+		WriteTimeout:   conf.Duration(10 * time.Second),
+		WriteQueueSize: 2048,
+		Parent:         p,
+	}
+
+	done := make(chan struct{})
+	defer func() { <-done }()
+
+	ctx, ctxCancel := context.WithCancel(context.Background())
+	defer ctxCancel()
+
+	go func() {
+		so.Run(defs.StaticSourceRunParams{ //nolint:errcheck
+			Context:        ctx,
+			ResolvedSource: "rtsp://127.0.0.1:8555/teststream",
+			Conf:           cnf,
+		})
+		close(done)
+	}()
+
+	<-p.Unit
+}
+
 func TestRange(t *testing.T) {
 	for _, ca := range []string{"clock", "npt", "smpte"} {
 		t.Run(ca, func(t *testing.T) {

mediamtx.yml

@@ -593,6 +593,10 @@ pathDefaults:
   # * npt: duration such as "300ms", "1.5m" or "2h45m", valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h"
   # * smpte: duration such as "300ms", "1.5m" or "2h45m", valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h"
   rtspRangeStart:
+  # Scale header value to send to the source, in order to play the stream at a different speed.
+  # Negative values play in reverse, values > 1 fast-forward,
+  # values 0 < x < 1 play slow motion.
+  rtspScale:
   # Range of ports used as source port in outgoing UDP packets.
   rtspUDPSourcePortRange: [10000, 65535]