BIG rewrite stream info

Author: AlexxIT

internal/exec/exec.go

@@ -8,6 +8,7 @@ import (
 	"net/url"
 	"os"
 	"os/exec"
+	"slices"
 	"strings"
 	"sync"
 	"time"
@@ -80,7 +81,7 @@ func execHandle(rawURL string) (core.Producer, error) {
 	return handleRTSP(rawURL, cmd, path)
 }
 
-func handlePipe(_ string, cmd *exec.Cmd, query url.Values) (core.Producer, error) {
+func handlePipe(source string, cmd *exec.Cmd, query url.Values) (core.Producer, error) {
 	if query.Get("backchannel") == "1" {
 		return stdin.NewClient(cmd)
 	}
@@ -104,12 +105,17 @@ func handlePipe(_ string, cmd *exec.Cmd, query url.Values) (core.Producer, error
 		return nil, fmt.Errorf("exec/pipe: %w\n%s", err, cmd.Stderr)
 	}
 
+	if info, ok := prod.(core.Info); ok {
+		info.SetProtocol("pipe")
+		setRemoteInfo(info, source, cmd.Args)
+	}
+
 	log.Debug().Stringer("launch", time.Since(ts)).Msg("[exec] run pipe")
 
 	return prod, nil
 }
 
-func handleRTSP(url string, cmd *exec.Cmd, path string) (core.Producer, error) {
+func handleRTSP(source string, cmd *exec.Cmd, path string) (core.Producer, error) {
 	if log.Trace().Enabled() {
 		cmd.Stdout = os.Stdout
 	}
@@ -131,7 +137,7 @@ func handleRTSP(url string, cmd *exec.Cmd, path string) (core.Producer, error) {
 	ts := time.Now()
 
 	if err := cmd.Start(); err != nil {
-		log.Error().Err(err).Str("url", url).Msg("[exec]")
+		log.Error().Err(err).Str("source", source).Msg("[exec]")
 		return nil, err
 	}
 
@@ -143,13 +149,14 @@ func handleRTSP(url string, cmd *exec.Cmd, path string) (core.Producer, error) {
 	select {
 	case <-time.After(time.Second * 60):
 		_ = cmd.Process.Kill()
-		log.Error().Str("url", url).Msg("[exec] timeout")
+		log.Error().Str("source", source).Msg("[exec] timeout")
 		return nil, errors.New("exec: timeout")
 	case <-done:
 		// limit message size
 		return nil, fmt.Errorf("exec/rtsp\n%s", cmd.Stderr)
 	case prod := <-waiter:
 		log.Debug().Stringer("launch", time.Since(ts)).Msg("[exec] run rtsp")
+		setRemoteInfo(prod, source, cmd.Args)
 		prod.OnClose = func() error {
 			log.Debug().Msgf("[exec] kill rtsp")
 			return errors.Join(cmd.Process.Kill(), cmd.Wait())
@@ -210,3 +217,15 @@ func trimSpace(b []byte) []byte {
 	}
 	return b[start:stop]
 }
+
+func setRemoteInfo(info core.Info, source string, args []string) {
+	info.SetSource(source)
+
+	if i := slices.Index(args, "-i"); i > 0 && i < len(args)-1 {
+		rawURL := args[i+1]
+		if u, err := url.Parse(rawURL); err == nil && u.Host != "" {
+			info.SetRemoteAddr(u.Host)
+			info.SetURL(rawURL)
+		}
+	}
+}

internal/ffmpeg/producer.go

@@ -13,7 +13,7 @@ import (
 )
 
 type Producer struct {
-	core.SuperProducer
+	core.Connection
 	url    string
 	query  url.Values
 	ffmpeg core.Producer
@@ -31,7 +31,8 @@ func NewProducer(url string) (core.Producer, error) {
 		return nil, errors.New("ffmpeg: unsupported params: " + url[i:])
 	}
 
-	p.Type = "FFmpeg producer"
+	p.ID = core.NewID()
+	p.FormatName = "ffmpeg"
 	p.Medias = []*core.Media{
 		{
 			// we can support only audio, because don't know FmtpLine for H264 and PayloadType for MJPEG
@@ -81,7 +82,7 @@ func (p *Producer) Stop() error {
 
 func (p *Producer) MarshalJSON() ([]byte, error) {
 	if p.ffmpeg == nil {
-		return json.Marshal(p.SuperProducer)
+		return json.Marshal(p.Connection)
 	}
 	return json.Marshal(p.ffmpeg)
 }