Update version to 1.9.1

feat(build): add multi-platform build shell script

go.mod

@@ -7,20 +7,20 @@ require (
 	github.com/expr-lang/expr v1.16.5
 	github.com/gorilla/websocket v1.5.1
 	github.com/miekg/dns v1.1.59
-	github.com/pion/ice/v2 v2.3.19
+	github.com/pion/ice/v2 v2.3.24
 	github.com/pion/interceptor v0.1.29
 	github.com/pion/rtcp v1.2.14
 	github.com/pion/rtp v1.8.6
 	github.com/pion/sdp/v3 v3.0.9
 	github.com/pion/srtp/v2 v2.0.18
 	github.com/pion/stun v0.6.1
-	github.com/pion/webrtc/v3 v3.2.39
+	github.com/pion/webrtc/v3 v3.2.40
 	github.com/rs/zerolog v1.32.0
 	github.com/sigurn/crc16 v0.0.0-20240131213347-83fcde1e29d1
 	github.com/sigurn/crc8 v0.0.0-20220107193325-2243fe600f9f
 	github.com/stretchr/testify v1.9.0
 	github.com/tadglines/go-pkgs v0.0.0-20210623144937-b983b20f54f9
-	golang.org/x/crypto v0.22.0
+	golang.org/x/crypto v0.23.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -32,7 +32,7 @@ require (
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/pion/datachannel v1.5.6 // indirect
-	github.com/pion/dtls/v2 v2.2.10 // indirect
+	github.com/pion/dtls/v2 v2.2.11 // indirect
 	github.com/pion/logging v0.2.2 // indirect
 	github.com/pion/mdns v0.0.12 // indirect
 	github.com/pion/randutil v0.1.0 // indirect
@@ -41,8 +41,8 @@ require (
 	github.com/pion/turn/v2 v2.1.6 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	golang.org/x/mod v0.17.0 // indirect
-	golang.org/x/net v0.24.0 // indirect
+	golang.org/x/net v0.25.0 // indirect
 	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/sys v0.19.0 // indirect
+	golang.org/x/sys v0.20.0 // indirect
 	golang.org/x/tools v0.20.0 // indirect
 )

go.sum

@@ -33,8 +33,12 @@ github.com/pion/datachannel v1.5.6/go.mod h1:1eKT6Q85pRnr2mHiWHxJwO50SfZRtWHTsNI
 github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s=
 github.com/pion/dtls/v2 v2.2.10 h1:u2Axk+FyIR1VFTPurktB+1zoEPGIW3bmyj3LEFrXjAA=
 github.com/pion/dtls/v2 v2.2.10/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE=
+github.com/pion/dtls/v2 v2.2.11 h1:9U/dpCYl1ySttROPWJgqWKEylUdT0fXp/xst6JwY5Ks=
+github.com/pion/dtls/v2 v2.2.11/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE=
 github.com/pion/ice/v2 v2.3.19 h1:1GoMRTMnB6bCP4aGy2MjxK3w4laDkk+m7svJb/eqybc=
 github.com/pion/ice/v2 v2.3.19/go.mod h1:KXJJcZK7E8WzrBEYnV4UtqEZsGeWfHxsNqhVcVvgjxw=
+github.com/pion/ice/v2 v2.3.24 h1:RYgzhH/u5lH0XO+ABatVKCtRd+4U1GEaCXSMjNr13tI=
+github.com/pion/ice/v2 v2.3.24/go.mod h1:KXJJcZK7E8WzrBEYnV4UtqEZsGeWfHxsNqhVcVvgjxw=
 github.com/pion/interceptor v0.1.29 h1:39fsnlP1U8gw2JzOFWdfCU82vHvhW9o0rZnZF56wF+M=
 github.com/pion/interceptor v0.1.29/go.mod h1:ri+LGNjRUc5xUNtDEPzfdkmSqISixVTBF/z/Zms/6T4=
 github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
@@ -72,6 +76,8 @@ github.com/pion/turn/v2 v2.1.6 h1:Xr2niVsiPTB0FPtt+yAWKFUkU1eotQbGgpTIld4x1Gc=
 github.com/pion/turn/v2 v2.1.6/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
 github.com/pion/webrtc/v3 v3.2.39 h1:Lf2SIMGdE3M9VNm48KpoX5pR8SJ6TsMnktzOkc/oB0o=
 github.com/pion/webrtc/v3 v3.2.39/go.mod h1:AQ8p56OLbm3MjhYovYdgPuyX6oc+JcKx/HFoCGFcYzA=
+github.com/pion/webrtc/v3 v3.2.40 h1:Wtfi6AZMQg+624cvCXUuSmrKWepSB7zfgYDOYqsSOVU=
+github.com/pion/webrtc/v3 v3.2.40/go.mod h1:M1RAe3TNTD1tzyvqHrbVODfwdPGSXOUo/OgpoGGJqFY=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/profile v1.4.0/go.mod h1:NWz/XGvpEW1FyYQ7fCx4dqYBLlfTcE+A9FLAkNKqjFE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -107,6 +113,8 @@ golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDf
 golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs=
 golang.org/x/crypto v0.22.0 h1:g1v0xeRhjcugydODzvb3mEM9SQ0HGp9s/nh3COQ/C30=
 golang.org/x/crypto v0.22.0/go.mod h1:vr6Su+7cTlO45qkww3VDJlzDn0ctJvRgYbC2NvXHt+M=
+golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI=
+golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.17.0 h1:zY54UmvipHiNd+pm+m0x9KhZ9hl1/7QNMyxXbc6ICqA=
@@ -124,6 +132,8 @@ golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/net v0.22.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
 golang.org/x/net v0.24.0 h1:1PcaxkF854Fu3+lvBIx5SYn9wRlBzzcnHZSiaFFAb0w=
 golang.org/x/net v0.24.0/go.mod h1:2Q7sJY5mzlzWjKtYUEXSlBWCdyaioyXzRB2RtU8KVE8=
+golang.org/x/net v0.25.0 h1:d/OCCoBEUq33pjydKrGQhw7IlUPI2Oylr+8qLx49kac=
+golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -148,6 +158,8 @@ golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
 golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y=
+golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=

internal/api/ws/ws.go

@@ -11,7 +11,7 @@ import (
 	"github.com/AlexxIT/go2rtc/internal/api"
 	"github.com/AlexxIT/go2rtc/internal/app"
 	"github.com/gorilla/websocket"
-	"github.com/rs/zerolog/log"
+	"github.com/rs/zerolog"
 )
 
 func Init() {
@@ -23,11 +23,15 @@ func Init() {
 
 	app.LoadConfig(&cfg)
 
+	log = app.GetLogger("api")
+
 	initWS(cfg.Mod.Origin)
 
 	api.HandleFunc("api/ws", apiWS)
 }
 
+var log zerolog.Logger
+
 // Message - struct for data exchange in Web API
 type Message struct {
 	Type  string `json:"type"`

internal/app/app.go

@@ -15,7 +15,7 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-var Version = "1.9.0"
+var Version = "1.9.1"
 var UserAgent = "go2rtc/" + Version
 
 var ConfigPath string

internal/exec/exec.go

@@ -5,6 +5,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"io"
 	"net/url"
 	"os"
 	"os/exec"
@@ -49,34 +50,34 @@ func Init() {
 
 func execHandle(rawURL string) (core.Producer, error) {
 	var path string
+	var query url.Values
 
-	rawURL, rawQuery, _ := strings.Cut(rawURL, "#")
-
-	args := shell.QuoteSplit(rawURL[5:]) // remove `exec:`
-	for i, arg := range args {
-		if arg == "{output}" {
-			if rtsp.Port == "" {
-				return nil, errors.New("rtsp module disabled")
-			}
-
-			sum := md5.Sum([]byte(rawURL))
-			path = "/" + hex.EncodeToString(sum[:])
-			args[i] = "rtsp://127.0.0.1:" + rtsp.Port + path
-			break
+	// RTSP flow should have `{output}` inside URL
+	// pipe flow may have `#{params}` inside URL
+	if i := strings.Index(rawURL, "{output}"); i > 0 {
+		if rtsp.Port == "" {
+			return nil, errors.New("exec: rtsp module disabled")
 		}
+
+		sum := md5.Sum([]byte(rawURL))
+		path = "/" + hex.EncodeToString(sum[:])
+		rawURL = rawURL[:i] + "rtsp://127.0.0.1:" + rtsp.Port + path + rawURL[i+8:]
+	} else if i = strings.IndexByte(rawURL, '#'); i > 0 {
+		query = streams.ParseQuery(rawURL[i+1:])
+		rawURL = rawURL[:i]
 	}
 
+	args := shell.QuoteSplit(rawURL[5:]) // remove `exec:`
 	cmd := exec.Command(args[0], args[1:]...)
 	if log.Debug().Enabled() {
 		cmd.Stderr = os.Stderr
 	}
 
 	if path == "" {
-		query := streams.ParseQuery(rawQuery)
 		return handlePipe(rawURL, cmd, query)
 	}
 
-	return handleRTSP(rawURL, path, cmd)
+	return handleRTSP(rawURL, cmd, path)
 }
 
 func handlePipe(_ string, cmd *exec.Cmd, query url.Values) (core.Producer, error) {
@@ -101,15 +102,23 @@ func handlePipe(_ string, cmd *exec.Cmd, query url.Values) (core.Producer, error
 	return prod, err
 }
 
-func handleRTSP(url, path string, cmd *exec.Cmd) (core.Producer, error) {
+func handleRTSP(url string, cmd *exec.Cmd, path string) (core.Producer, error) {
+	stderr := limitBuffer{buf: make([]byte, 512)}
+
+	if cmd.Stderr != nil {
+		cmd.Stderr = io.MultiWriter(cmd.Stderr, &stderr)
+	} else {
+		cmd.Stderr = &stderr
+	}
+
 	if log.Trace().Enabled() {
 		cmd.Stdout = os.Stdout
 	}
 
-	ch := make(chan core.Producer)
+	waiter := make(chan core.Producer)
 
 	waitersMu.Lock()
-	waiters[path] = ch
+	waiters[path] = waiter
 	waitersMu.Unlock()
 
 	defer func() {
@@ -127,16 +136,9 @@ func handleRTSP(url, path string, cmd *exec.Cmd) (core.Producer, error) {
 		return nil, err
 	}
 
-	chErr := make(chan error)
-
+	done := make(chan error, 1)
 	go func() {
-		err := cmd.Wait()
-		// unblocking write to channel
-		select {
-		case chErr <- err:
-		default:
-			log.Trace().Str("url", url).Msg("[exec] close")
-		}
+		done <- cmd.Wait()
 	}()
 
 	select {
@@ -144,9 +146,10 @@ func handleRTSP(url, path string, cmd *exec.Cmd) (core.Producer, error) {
 		_ = cmd.Process.Kill()
 		log.Error().Str("url", url).Msg("[exec] timeout")
 		return nil, errors.New("timeout")
-	case err := <-chErr:
-		return nil, fmt.Errorf("exec: %s", err)
-	case prod := <-ch:
+	case <-done:
+		// limit message size
+		return nil, errors.New("exec: " + stderr.String())
+	case prod := <-waiter:
 		log.Debug().Stringer("launch", time.Since(ts)).Msg("[exec] run")
 		return prod, nil
 	}
@@ -159,3 +162,22 @@ var (
 	waiters   = map[string]chan core.Producer{}
 	waitersMu sync.Mutex
 )
+
+type limitBuffer struct {
+	buf []byte
+	n   int
+}
+
+func (l *limitBuffer) String() string {
+	if l.n == len(l.buf) {
+		return string(l.buf) + "..."
+	}
+	return string(l.buf[:l.n])
+}
+
+func (l *limitBuffer) Write(p []byte) (int, error) {
+	if l.n < cap(l.buf) {
+		l.n += copy(l.buf[l.n:], p)
+	}
+	return len(p), nil
+}

internal/ffmpeg/ffmpeg.go

@@ -7,6 +7,7 @@ import (
 	"github.com/AlexxIT/go2rtc/internal/app"
 	"github.com/AlexxIT/go2rtc/internal/ffmpeg/device"
 	"github.com/AlexxIT/go2rtc/internal/ffmpeg/hardware"
+	"github.com/AlexxIT/go2rtc/internal/ffmpeg/virtual"
 	"github.com/AlexxIT/go2rtc/internal/rtsp"
 	"github.com/AlexxIT/go2rtc/internal/streams"
 	"github.com/AlexxIT/go2rtc/pkg/ffmpeg"
@@ -145,7 +146,7 @@ func parseArgs(s string) *ffmpeg.Args {
 	}
 
 	var query url.Values
-	if i := strings.IndexByte(s, '#'); i > 0 {
+	if i := strings.IndexByte(s, '#'); i >= 0 {
 		query = streams.ParseQuery(s[i+1:])
 		args.Video = len(query["video"])
 		args.Audio = len(query["audio"])
@@ -193,6 +194,11 @@ func parseArgs(s string) *ffmpeg.Args {
 		if err != nil {
 			return nil
 		}
+	} else if strings.HasPrefix(s, "virtual?") {
+		var err error
+		if args.Input, err = virtual.GetInput(s[8:]); err != nil {
+			return nil
+		}
 	} else {
 		args.Input = inputTemplate("file", s, query)
 	}

internal/ffmpeg/virtual/virtual.go

@@ -0,0 +1,59 @@
+package virtual
+
+import (
+	"net/url"
+)
+
+func GetInput(src string) (string, error) {
+	query, err := url.ParseQuery(src)
+	if err != nil {
+		return "", err
+	}
+
+	// set defaults (using Add instead of Set)
+	query.Add("source", "testsrc")
+	query.Add("size", "1920x1080")
+	query.Add("decimals", "2")
+
+	// https://ffmpeg.org/ffmpeg-filters.html
+	source := query.Get("source")
+	input := "-re -f lavfi -i " + source
+
+	sep := "=" // first separator
+	for key, values := range query {
+		value := values[0]
+
+		// https://ffmpeg.org/ffmpeg-utils.html#video-size-syntax
+		switch key {
+		case "color", "rate", "duration", "sar":
+		case "size":
+			switch value {
+			case "720":
+				value = "1280x720"
+			case "1080":
+				value = "1920x1080"
+			case "2K":
+				value = "2560x1440"
+			case "4K":
+				value = "3840x2160"
+			case "8K":
+				value = "7680x4230" // https://reolink.com/blog/8k-resolution/
+			}
+		case "decimals":
+			if source != "testsrc" {
+				continue
+			}
+		default:
+			continue
+		}
+
+		input += sep + key + "=" + value
+		sep = ":" // next separator
+	}
+
+	if s := query.Get("format"); s != "" {
+		input += ",format=" + s
+	}
+
+	return input, nil
+}

internal/mjpeg/init.go

@@ -57,6 +57,8 @@ func handlerKeyframe(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 		log.Debug().Msgf("[mjpeg] transcoding time=%s", time.Since(ts))
+	case core.CodecJPEG:
+		b = mjpeg.FixJPEG(b)
 	}
 
 	h := w.Header()
@@ -163,7 +165,7 @@ func handlerWS(tr *ws.Transport, _ *ws.Message) error {
 	cons.UserAgent = tr.Request.UserAgent()
 
 	if err := stream.AddConsumer(cons); err != nil {
-		log.Error().Err(err).Caller().Send()
+		log.Debug().Err(err).Msg("[mjpeg] add consumer")
 		return err
 	}
 

internal/rtsp/rtsp.go

@@ -239,7 +239,7 @@ func tcpHandler(conn *rtsp.Conn) {
 
 	if closer != nil {
 		if err := conn.Handle(); err != nil {
-			log.Debug().Msgf("[rtsp] handle=%s", err)
+			log.Debug().Err(err).Msg("[rtsp] handle")
 		}
 
 		closer()

internal/streams/add_consumer.go

@@ -3,18 +3,17 @@ package streams
 import (
 	"errors"
 	"strings"
-	"sync/atomic"
 
 	"github.com/AlexxIT/go2rtc/pkg/core"
 )
 
 func (s *Stream) AddConsumer(cons core.Consumer) (err error) {
-	// support for multiple simultaneous requests from different consumers
-	consN := atomic.AddInt32(&s.requests, 1) - 1
+	// support for multiple simultaneous pending from different consumers
+	consN := s.pending.Add(1) - 1
 
-	var prodErrors []error
+	var prodErrors = make([]error, len(s.producers))
 	var prodMedias []*core.Media
-	var prods []*Producer // matched producers for consumer
+	var prodStarts []*Producer
 
 	// Step 1. Get consumer medias
 	consMedias := cons.GetMedias()
@@ -23,15 +22,20 @@ func (s *Stream) AddConsumer(cons core.Consumer) (err error) {
 
 	producers:
 		for prodN, prod := range s.producers {
+			if prodErrors[prodN] != nil {
+				log.Trace().Msgf("[streams] skip cons=%d prod=%d", consN, prodN)
+				continue
+			}
+
 			if err = prod.Dial(); err != nil {
-				log.Trace().Err(err).Msgf("[streams] skip prod=%s", prod.url)
-				prodErrors = append(prodErrors, err)
+				log.Trace().Err(err).Msgf("[streams] dial cons=%d prod=%d", consN, prodN)
+				prodErrors[prodN] = err
 				continue
 			}
 
 			// Step 2. Get producer medias (not tracks yet)
 			for _, prodMedia := range prod.GetMedias() {
-				log.Trace().Msgf("[streams] check prod=%d media=%s", prodN, prodMedia)
+				log.Trace().Msgf("[streams] check cons=%d prod=%d media=%s", consN, prodN, prodMedia)
 				prodMedias = append(prodMedias, prodMedia)
 
 				// Step 3. Match consumer/producer codecs list
@@ -44,11 +48,12 @@ func (s *Stream) AddConsumer(cons core.Consumer) (err error) {
 
 				switch prodMedia.Direction {
 				case core.DirectionRecvonly:
-					log.Trace().Msgf("[streams] match prod=%d => cons=%d", prodN, consN)
+					log.Trace().Msgf("[streams] match cons=%d <= prod=%d", consN, prodN)
 
 					// Step 4. Get recvonly track from producer
 					if track, err = prod.GetTrack(prodMedia, prodCodec); err != nil {
 						log.Info().Err(err).Msg("[streams] can't get track")
+						prodErrors[prodN] = err
 						continue
 					}
 					// Step 5. Add track to consumer
@@ -68,11 +73,12 @@ func (s *Stream) AddConsumer(cons core.Consumer) (err error) {
 					// Step 5. Add track to producer
 					if err = prod.AddTrack(prodMedia, prodCodec, track); err != nil {
 						log.Info().Err(err).Msg("[streams] can't add track")
+						prodErrors[prodN] = err
 						continue
 					}
 				}
 
-				prods = append(prods, prod)
+				prodStarts = append(prodStarts, prod)
 
 				if !consMedia.MatchAll() {
 					break producers
@@ -82,11 +88,11 @@ func (s *Stream) AddConsumer(cons core.Consumer) (err error) {
 	}
 
 	// stop producers if they don't have readers
-	if atomic.AddInt32(&s.requests, -1) == 0 {
+	if s.pending.Add(-1) == 0 {
 		s.stopProducers()
 	}
 
-	if len(prods) == 0 {
+	if len(prodStarts) == 0 {
 		return formatError(consMedias, prodMedias, prodErrors)
 	}
 
@@ -95,7 +101,7 @@ func (s *Stream) AddConsumer(cons core.Consumer) (err error) {
 	s.mu.Unlock()
 
 	// there may be duplicates, but that's not a problem
-	for _, prod := range prods {
+	for _, prod := range prodStarts {
 		prod.start()
 	}
 
@@ -103,6 +109,20 @@ func (s *Stream) AddConsumer(cons core.Consumer) (err error) {
 }
 
 func formatError(consMedias, prodMedias []*core.Media, prodErrors []error) error {
+	// 1. Return errors if any not nil
+	var text string
+
+	for _, err := range prodErrors {
+		if err != nil {
+			text = appendString(text, err.Error())
+		}
+	}
+
+	if len(text) != 0 {
+		return errors.New("streams: " + text)
+	}
+
+	// 2. Return "codecs not matched"
 	if prodMedias != nil {
 		var prod, cons string
 
@@ -125,16 +145,7 @@ func formatError(consMedias, prodMedias []*core.Media, prodErrors []error) error
 		return errors.New("streams: codecs not matched: " + prod + " => " + cons)
 	}
 
-	if prodErrors != nil {
-		var text string
-
-		for _, err := range prodErrors {
-			text = appendString(text, err.Error())
-		}
-
-		return errors.New("streams: " + text)
-	}
-
+	// 3. Return unknown error
 	return errors.New("streams: unknown error")
 }
 

internal/streams/producer.go

@@ -245,10 +245,10 @@ func (p *Producer) stop() {
 
 	switch p.state {
 	case stateExternal:
-		log.Debug().Msgf("[streams] can't stop external producer")
+		log.Trace().Msgf("[streams] skip stop external producer")
 		return
 	case stateNone:
-		log.Debug().Msgf("[streams] can't stop none producer")
+		log.Trace().Msgf("[streams] skip stop none producer")
 		return
 	case stateStart:
 		p.workerID++

internal/streams/stream.go

@@ -3,6 +3,7 @@ package streams
 import (
 	"encoding/json"
 	"sync"
+	"sync/atomic"
 
 	"github.com/AlexxIT/go2rtc/pkg/core"
 )
@@ -11,7 +12,7 @@ type Stream struct {
 	producers []*Producer
 	consumers []core.Consumer
 	mu        sync.Mutex
-	requests  int32
+	pending   atomic.Int32
 }
 
 func NewStream(source any) *Stream {

pkg/h264/payloader.go

@@ -67,11 +67,15 @@ func EmitNalus(nals []byte, isAVC bool, emit func([]byte)) {
 		}
 	} else {
 		for {
-			end := 4 + binary.BigEndian.Uint32(nals)
-			emit(nals[4:end])
-			if int(end) >= len(nals) {
+			n := uint32(len(nals))
+			if n < 4 {
 				break
 			}
+			end := 4 + binary.BigEndian.Uint32(nals)
+			if n < end {
+				break
+			}
+			emit(nals[4:end])
 			nals = nals[end:]
 		}
 	}

pkg/hls/reader.go

@@ -88,7 +88,7 @@ func (r *reader) RoundTrip(_ *http.Request) (*http.Response, error) {
 }
 
 func (r *reader) getSegment() ([]byte, error) {
-	for i := 0; i < 5; i++ {
+	for i := 0; i < 10; i++ {
 		if r.playlist == nil {
 			if wait := time.Second - time.Since(r.lastTime); wait > 0 {
 				time.Sleep(wait)

pkg/magic/producer.go

@@ -34,12 +34,12 @@ func Open(r io.Reader) (core.Producer, error) {
 	case bytes.HasPrefix(b, []byte(flv.Signature)):
 		return flv.Open(rd)
 
-	case bytes.HasPrefix(b, []byte{0xFF, 0xF1}):
-		return aac.Open(rd)
-
 	case bytes.HasPrefix(b, []byte("--")):
 		return multipart.Open(rd)
 
+	case b[0] == 0xFF && b[1]&0xF7 == 0xF1:
+		return aac.Open(rd)
+
 	case b[0] == mpegts.SyncByte:
 		return mpegts.Open(rd)
 	}

pkg/mjpeg/helpers.go

@@ -0,0 +1,35 @@
+package mjpeg
+
+import (
+	"bytes"
+	"image/jpeg"
+)
+
+// FixJPEG - reencode JPEG if it has wrong header
+//
+// for example, this app produce "bad" images:
+// https://github.com/jacksonliam/mjpg-streamer
+//
+// and they can't be uploaded to the Telegram servers:
+// {"ok":false,"error_code":400,"description":"Bad Request: IMAGE_PROCESS_FAILED"}
+func FixJPEG(b []byte) []byte {
+	// skip non-JPEG
+	if len(b) < 10 || b[0] != 0xFF || b[1] != 0xD8 {
+		return b
+	}
+	// skip if header OK for imghdr library
+	// https://docs.python.org/3/library/imghdr.html
+	if string(b[2:4]) == "\xFF\xDB" || string(b[6:10]) == "JFIF" || string(b[6:10]) == "Exif" {
+		return b
+	}
+
+	img, err := jpeg.Decode(bytes.NewReader(b))
+	if err != nil {
+		return b
+	}
+	buf := bytes.NewBuffer(nil)
+	if err = jpeg.Encode(buf, img, nil); err != nil {
+		return b
+	}
+	return buf.Bytes()
+}

pkg/rtsp/server.go

@@ -146,7 +146,19 @@ func (c *Conn) Accept() error {
 			if strings.HasPrefix(tr, transport) {
 				c.session = core.RandString(8, 10)
 				c.state = StateSetup
-				res.Header.Set("Transport", tr[:len(transport)+3])
+
+				if c.mode == core.ModePassiveConsumer {
+					if i := reqTrackID(req); i >= 0 && i < len(c.senders) {
+						// mark sender as SETUP
+						c.senders[i].Media.ID = MethodSetup
+						tr = fmt.Sprintf("RTP/AVP/TCP;unicast;interleaved=%d-%d", i*2, i*2+1)
+						res.Header.Set("Transport", tr)
+					} else {
+						res.Status = "400 Bad Request"
+					}
+				} else {
+					res.Header.Set("Transport", tr[:len(transport)+3])
+				}
 			} else {
 				res.Status = "461 Unsupported transport"
 			}
@@ -156,6 +168,15 @@ func (c *Conn) Accept() error {
 			}
 
 		case MethodRecord, MethodPlay:
+			if c.mode == core.ModePassiveConsumer {
+				// stop unconfigured senders
+				for _, track := range c.senders {
+					if track.Media.ID != MethodSetup {
+						track.Close()
+					}
+				}
+			}
+
 			res := &tcp.Response{Request: req}
 			err = c.WriteResponse(res)
 			c.playOK = true
@@ -172,3 +193,18 @@ func (c *Conn) Accept() error {
 		}
 	}
 }
+
+func reqTrackID(req *tcp.Request) int {
+	var s string
+	if req.URL.RawQuery != "" {
+		s = req.URL.RawQuery
+	} else {
+		s = req.URL.Path
+	}
+	if i := strings.LastIndexByte(s, '='); i > 0 {
+		if i, err := strconv.Atoi(s[i+1:]); err == nil {
+			return i
+		}
+	}
+	return -1
+}

pkg/stdin/client.go

@@ -1,15 +1,13 @@
 package stdin
 
 import (
-	"io"
 	"os/exec"
 
 	"github.com/AlexxIT/go2rtc/pkg/core"
 )
 
 type Client struct {
-	cmd  *exec.Cmd
-	pipe io.WriteCloser
+	cmd *exec.Cmd
 
 	medias []*core.Media
 	sender *core.Sender
@@ -17,14 +15,8 @@ type Client struct {
 }
 
 func NewClient(cmd *exec.Cmd) (*Client, error) {
-	pipe, err := PipeCloser(cmd)
-	if err != nil {
-		return nil, err
-	}
-
 	c := &Client{
-		pipe: pipe,
-		cmd:  cmd,
+		cmd: cmd,
 		medias: []*core.Media{
 			{
 				Kind:      core.KindAudio,

pkg/stdin/consumer.go

@@ -2,6 +2,7 @@ package stdin
 
 import (
 	"encoding/json"
+	"errors"
 
 	"github.com/AlexxIT/go2rtc/pkg/core"
 	"github.com/pion/rtp"
@@ -17,9 +18,14 @@ func (c *Client) GetTrack(media *core.Media, codec *core.Codec) (*core.Receiver,
 
 func (c *Client) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver) error {
 	if c.sender == nil {
+		stdin, err := c.cmd.StdinPipe()
+		if err != nil {
+			return err
+		}
+
 		c.sender = core.NewSender(media, track.Codec)
 		c.sender.Handler = func(packet *rtp.Packet) {
-			_, _ = c.pipe.Write(packet.Payload)
+			_, _ = stdin.Write(packet.Payload)
 			c.send += len(packet.Payload)
 		}
 	}
@@ -36,7 +42,10 @@ func (c *Client) Stop() (err error) {
 	if c.sender != nil {
 		c.sender.Close()
 	}
-	return c.pipe.Close()
+	if c.cmd.Process == nil {
+		return nil
+	}
+	return errors.Join(c.cmd.Process.Kill(), c.cmd.Wait())
 }
 
 func (c *Client) MarshalJSON() ([]byte, error) {

pkg/stdin/pipe.go

@@ -1,26 +0,0 @@
-package stdin
-
-import (
-	"errors"
-	"io"
-	"os/exec"
-)
-
-type pipeCloser struct {
-	io.Writer
-	io.Closer
-	cmd *exec.Cmd
-}
-
-func PipeCloser(cmd *exec.Cmd) (io.WriteCloser, error) {
-	stdin, err := cmd.StdinPipe()
-	if err != nil {
-		return nil, err
-	}
-
-	return pipeCloser{stdin, stdin, cmd}, nil
-}
-
-func (p pipeCloser) Close() (err error) {
-	return errors.Join(p.Closer.Close(), p.cmd.Process.Kill(), p.cmd.Wait())
-}

pkg/webrtc/helpers.go

@@ -47,6 +47,9 @@ func UnmarshalMedias(descriptions []*sdp.MediaDescription) (medias []*core.Media
 				continue
 			}
 
+			// skip non-media codecs to avoid confusing users in info and logs
+			media.Codecs = SkipNonMediaCodecs(media.Codecs)
+
 			medias = append(medias, media)
 		}
 	}
@@ -54,6 +57,21 @@ func UnmarshalMedias(descriptions []*sdp.MediaDescription) (medias []*core.Media
 	return
 }
 
+func SkipNonMediaCodecs(input []*core.Codec) (output []*core.Codec) {
+	for _, codec := range input {
+		switch codec.Name {
+		case "RTX", "RED", "ULPFEC", "FLEXFEC-03":
+			continue
+		case "CN", "TELEPHONE-EVENT":
+			continue // https://datatracker.ietf.org/doc/html/rfc7874
+		}
+		// VP8, VP9, H264, H265, AV1
+		// OPUS, G722, PCMU, PCMA
+		output = append(output, codec)
+	}
+	return
+}
+
 // WithResampling - will add for consumer: PCMA/0, PCMU/0, PCM/0, PCML/0
 // so it can add resampling for PCMA/PCMU and repack for PCM/PCML
 func WithResampling(medias []*core.Media) []*core.Media {

scripts/build.sh

@@ -0,0 +1,82 @@
+#!/bin/sh
+
+check_command() {
+    if ! command -v $1 &> /dev/null
+    then
+        echo "Error: $1 could not be found. Please install it."
+        exit 1
+    fi
+}
+
+# Check for required commands
+check_command go
+check_command 7z
+check_command upx
+
+# Windows amd64
+export GOOS=windows
+export GOARCH=amd64
+FILENAME="go2rtc_win64.zip"
+go build -ldflags "-s -w" -trimpath && 7z a -mx9 -bso0 -sdel $FILENAME go2rtc.exe
+
+# Windows 386
+export GOOS=windows
+export GOARCH=386
+FILENAME="go2rtc_win32.zip"
+go build -ldflags "-s -w" -trimpath && 7z a -mx9 -bso0 -sdel $FILENAME go2rtc.exe
+
+# Windows arm64
+export GOOS=windows
+export GOARCH=arm64
+FILENAME="go2rtc_win_arm64.zip"
+go build -ldflags "-s -w" -trimpath && 7z a -mx9 -bso0 -sdel $FILENAME go2rtc.exe
+
+# Linux amd64
+export GOOS=linux
+export GOARCH=amd64
+FILENAME="go2rtc_linux_amd64"
+go build -ldflags "-s -w" -trimpath -o $FILENAME && upx --lzma --force-overwrite -q --no-progress $FILENAME
+
+# Linux 386
+export GOOS=linux
+export GOARCH=386
+FILENAME="go2rtc_linux_i386"
+go build -ldflags "-s -w" -trimpath -o $FILENAME && upx --lzma --force-overwrite -q --no-progress $FILENAME
+
+# Linux arm64
+export GOOS=linux
+export GOARCH=arm64
+FILENAME="go2rtc_linux_arm64"
+go build -ldflags "-s -w" -trimpath -o $FILENAME && upx --lzma --force-overwrite -q --no-progress $FILENAME
+
+# Linux arm v7
+export GOOS=linux
+export GOARCH=arm
+export GOARM=7
+FILENAME="go2rtc_linux_arm"
+go build -ldflags "-s -w" -trimpath -o $FILENAME && upx --lzma --force-overwrite -q --no-progress $FILENAME
+
+# Linux arm v6
+export GOOS=linux
+export GOARCH=arm
+export GOARM=6
+FILENAME="go2rtc_linux_armv6"
+go build -ldflags "-s -w" -trimpath -o $FILENAME && upx --lzma --force-overwrite -q --no-progress $FILENAME
+
+# Linux mipsle
+export GOOS=linux
+export GOARCH=mipsle
+FILENAME="go2rtc_linux_mipsel"
+go build -ldflags "-s -w" -trimpath -o $FILENAME && upx --lzma --force-overwrite -q --no-progress $FILENAME
+
+# Darwin amd64
+export GOOS=darwin
+export GOARCH=amd64
+FILENAME="go2rtc_mac_amd64.zip"
+go build -ldflags "-s -w" -trimpath && 7z a -mx9 -bso0 -sdel $FILENAME go2rtc
+
+# Darwin arm64
+export GOOS=darwin
+export GOARCH=arm64
+FILENAME="go2rtc_mac_arm64.zip"
+go build -ldflags "-s -w" -trimpath && 7z a -mx9 -bso0 -sdel $FILENAME go2rtc