Control RTSP stream speed (#11)

Control RTSP speed

internal/streams/custom.go

@@ -0,0 +1,58 @@
+package streams
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/AlexxIT/go2rtc/pkg/rtsp"
+)
+
+// only support RTSP sources
+func apiStreamsSpeed(w http.ResponseWriter, r *http.Request) {
+	query := r.URL.Query()
+	streamName := query.Get("name")
+
+	if streamName == "" {
+		http.Error(w, "name required", http.StatusBadRequest)
+		return
+	}
+
+	switch r.Method {
+	case "PUT":
+		streamsMu.RLock()
+		stream := Get(streamName)
+		defer streamsMu.RUnlock()
+
+		if stream == nil {
+			http.Error(w, "", http.StatusNotFound)
+			return
+		}
+
+		speedStr := query.Get("speed")
+		if speedStr == "" {
+			http.Error(w, "speed required", http.StatusBadRequest)
+			return
+		}
+
+		_, err := strconv.ParseFloat(speedStr, 64)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+
+		for _, producer := range stream.producers {
+			if conn, ok := producer.conn.(*rtsp.Conn); ok {
+				conn.Connection.Speed = speedStr
+				err := conn.Play()
+				if err != nil {
+					log.Error().Msgf("[stream] conn.Play(): %+v\n", err)
+					http.Error(w, err.Error(), http.StatusInternalServerError)
+				}
+			}
+		}
+
+		return
+	}
+
+	http.Error(w, "", http.StatusNotFound)
+}

internal/streams/streams.go

@@ -29,6 +29,9 @@ func Init() {
 	api.HandleFunc("api/streams", apiStreams)
 	api.HandleFunc("api/streams.dot", apiStreamsDOT)
 
+	// custom
+	api.HandleFunc("api/custom/streams/speed", apiStreamsSpeed)
+
 	if cfg.Publish == nil {
 		return
 	}

pkg/core/connection.go

@@ -50,8 +50,10 @@ type Connection struct {
 
 	Transport any `json:"-"`
 
+	// custom
 	stopBitrateWorker chan struct{} `json:"-"`
 	Bitrate           int           `json:"bitrate,omitempty"` // bytes per second
+	Speed             string        `json:"speed,omitempty"`   // empty string indicates normal speed
 }
 
 func (c *Connection) GetMedias() []*Media {

pkg/rtsp/client.go

@@ -313,6 +313,11 @@ func (c *Conn) SetupMedia(media *core.Media) (byte, error) {
 
 func (c *Conn) Play() (err error) {
 	req := &tcp.Request{Method: MethodPlay, URL: c.URL}
+	if c.Speed != "" {
+		req.Header = map[string][]string{
+			"Scale": {c.Speed},
+		}
+	}
 	return c.WriteRequest(req)
 }