Fix loop request, ex. camera1: ffmpeg:camera1

internal/ffmpeg/ffmpeg.go

@@ -179,6 +179,7 @@ func parseArgs(s string) *ffmpeg.Args {
 		Version: verAV,
 	}
 
+	var source = s
 	var query url.Values
 	if i := strings.IndexByte(s, '#'); i >= 0 {
 		query = streams.ParseQuery(s[i+1:])
@@ -221,6 +222,7 @@ func parseArgs(s string) *ffmpeg.Args {
 		default:
 			s += "?video&audio"
 		}
+		s += "&source=ffmpeg:" + url.QueryEscape(source)
 		args.Input = inputTemplate("rtsp", s, query)
 	} else if i = strings.Index(s, "?"); i > 0 {
 		switch s[:i] {

internal/rtsp/rtsp.go

@@ -188,6 +188,9 @@ func tcpHandler(conn *rtsp.Conn) {
 				conn.PacketSize = uint16(core.Atoi(s))
 			}
 
+			// will help to protect looping requests to same source
+			conn.Connection.Source = query.Get("source")
+
 			if err := stream.AddConsumer(conn); err != nil {
 				log.Warn().Err(err).Str("stream", name).Msg("[rtsp]")
 				return

internal/streams/add_consumer.go

@@ -22,6 +22,12 @@ func (s *Stream) AddConsumer(cons core.Consumer) (err error) {
 
 	producers:
 		for prodN, prod := range s.producers {
+			// check for loop request, ex. `camera1: ffmpeg:camera1`
+			if info, ok := cons.(core.Info); ok && prod.url == info.GetSource() {
+				log.Trace().Msgf("[streams] skip cons=%d prod=%d", consN, prodN)
+				continue
+			}
+
 			if prodErrors[prodN] != nil {
 				log.Trace().Msgf("[streams] skip cons=%d prod=%d", consN, prodN)
 				continue

pkg/core/connection.go

@@ -25,6 +25,7 @@ type Info interface {
 	SetSource(string)
 	SetURL(string)
 	WithRequest(*http.Request)
+	GetSource() string
 }
 
 // Connection just like webrtc.PeerConnection
@@ -123,6 +124,10 @@ func (c *Connection) WithRequest(r *http.Request) {
 	c.UserAgent = r.UserAgent()
 }
 
+func (c *Connection) GetSource() string {
+	return c.Source
+}
+
 // Create like os.Create, init Consumer with existing Transport
 func Create(w io.Writer) (*Connection, error) {
 	return &Connection{Transport: w}, nil