Update stream log message

Add ability for API to set name of stream

README.md

@@ -172,6 +172,8 @@ streams:
     - rtsp://admin:password@192.168.1.123/cam/realmonitor?channel=1&subtype=1
 ```
 
+**PS.** For disable bachannel just add `#backchannel=0` to end of RTSP link.
+
 #### Source: RTMP
 
 You can get stream from RTMP server, for example [Frigate](https://docs.frigate.video/configuration/rtmp). Support ONLY `H264` video codec without audio.
@@ -385,13 +387,13 @@ ngrok:
   command: ...
 ```
 
-**Own TCP-tunnel**
+**Hard tech way 1. Own TCP-tunnel**
 
-If you have personal VPS, you can create TCP-tunnel and setup in the same way as "Static public IP". But use your VPS IP-address in YAML config.
+If you have personal [VPS](https://en.wikipedia.org/wiki/Virtual_private_server), you can create TCP-tunnel and setup in the same way as "Static public IP". But use your VPS IP-address in YAML config.
 
-**Using TURN-server**
+**Hard tech way 2. Using TURN-server**
 
-TODO...
+If you have personal [VPS](https://en.wikipedia.org/wiki/Virtual_private_server), you can install TURN server (e.g. [coturn](https://github.com/coturn/coturn), config [example](https://github.com/AlexxIT/WebRTC/wiki/Coturn-Example)).
 
 ```yaml
 webrtc:
@@ -547,6 +549,30 @@ If you need Web interface protection without Home Assistant Add-on - you need to
 
 PS. Additionally WebRTC opens a lot of random UDP ports for transmit encrypted media. They work without problems on the local network. And sometimes work for external access, even if you haven't opened ports on your router. But for stable external WebRTC access, you need to configure the TCP port.
 
+## Codecs madness
+
+`AVC/H.264` codec can be played almost anywhere. But `HEVC/H.265` has a lot of limitations in supporting with different devices and browsers. It's all about patents and money, you can't do anything about it.
+
+Device | WebRTC | MSE | MP4
+-------|--------|-----|----
+*latency*       | best        | medium      | bad
+Desktop Chrome  | H264        | H264, H265* | H264, H265*
+Desktop Safari  | H264, H265* | H264        | no
+Desktop Edge    | H264        | H264, H265* | H264, H265*
+Desktop Firefox | H264        | H264        | H264
+Desktop Opera   | no          | H264        | H264
+iPhone Safari   | H264, H265* | no          | no
+iPad Safari     | H264, H265* | H264        | no
+Android Chrome  | H264        | H264        | H264
+masOS Hass App  | no          | no          | no
+
+- WebRTC audio codecs: `PCMU/8000`, `PCMA/8000`, `OPUS/48000/2`
+- MSE/MP4 audio codecs: not supported yet (should be: `AAC`)
+- Chrome H265: [read this](https://github.com/StaZhu/enable-chromium-hevc-hardware-decoding)
+- Edge H265: [read this](https://www.reddit.com/r/MicrosoftEdge/comments/v9iw8k/enable_hevc_support_in_edge/)
+- Desktop Safari H265: Menu > Develop > Experimental > WebRTC H265
+- iOS Safari H265: Settings > Safari > Advanced > Experimental > WebRTC H265
+
 ## FAQ
 
 **Q. What's the difference between go2rtc, WebRTC Camera and RTSPtoWebRTC?**

cmd/api/api.go

@@ -85,10 +85,15 @@ var wsHandlers = make(map[string]WSHandler)
 
 func streamsHandler(w http.ResponseWriter, r *http.Request) {
 	src := r.URL.Query().Get("src")
+	name := r.URL.Query().Get("name")
+	
+	if name == "" {
+		name = src
+	}
 
 	switch r.Method {
 	case "PUT":
-		streams.New(src, src)
+		streams.New(name, src)
 		return
 	case "DELETE":
 		streams.Delete(src)

cmd/app/app.go

@@ -3,6 +3,7 @@ package app
 import (
 	"flag"
 	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
 	"gopkg.in/yaml.v3"
 	"io"
 	"os"
@@ -30,10 +31,18 @@ func Init() {
 		}
 	}
 
+	log.Logger = NewLogger(cfg.Mod["format"], cfg.Mod["level"])
+
+	modules = cfg.Mod
+
+	path, _ := os.Getwd()
+	log.Debug().Str("os", runtime.GOOS).Str("arch", runtime.GOARCH).
+		Str("cwd", path).Int("conf_size", len(data)).Msgf("[app]")
+}
+
+func NewLogger(format string, level string) zerolog.Logger {
 	var writer io.Writer = os.Stdout
 
-	// styles
-	format := cfg.Mod["format"]
 	if format != "json" {
 		writer = zerolog.ConsoleWriter{
 			Out: writer, TimeFormat: "15:04:05.000",
@@ -43,18 +52,12 @@ func Init() {
 
 	zerolog.TimeFieldFormat = zerolog.TimeFormatUnixMs
 
-	lvl, err := zerolog.ParseLevel(cfg.Mod["level"])
+	lvl, err := zerolog.ParseLevel(level)
 	if err != nil || lvl == zerolog.NoLevel {
 		lvl = zerolog.InfoLevel
 	}
 
-	log = zerolog.New(writer).With().Timestamp().Logger().Level(lvl)
+	return zerolog.New(writer).With().Timestamp().Logger().Level(lvl)
-
-	modules = cfg.Mod
-
-	path, _ := os.Getwd()
-	log.Debug().Str("os", runtime.GOOS).Str("arch", runtime.GOARCH).
-		Str("cwd", path).Int("conf_size", len(data)).Msgf("[app]")
 }
 
 func LoadConfig(v interface{}) {
@@ -68,15 +71,13 @@ func LoadConfig(v interface{}) {
 func GetLogger(module string) zerolog.Logger {
 	if s, ok := modules[module]; ok {
 		lvl, err := zerolog.ParseLevel(s)
-		if err != nil {
+		if err == nil {
-			log.Warn().Err(err).Msg("[log]")
+			return log.Level(lvl)
-			return log
 		}
-
+		log.Warn().Err(err).Caller().Send()
-		return log.Level(lvl)
 	}
 
-	return log
+	return log.Logger
 }
 
 // internal
@@ -84,8 +85,5 @@ func GetLogger(module string) zerolog.Logger {
 // data - config content
 var data []byte
 
-// log - main logger
-var log zerolog.Logger
-
 // modules log levels
 var modules map[string]string

cmd/debug/stack.go

@@ -21,6 +21,7 @@ var stackSkip = [][]byte{
 	[]byte("created by net/http.(*Server).Serve"), // TODO: why two?
 
 	[]byte("created by github.com/AlexxIT/go2rtc/cmd/rtsp.Init"),
+	[]byte("created by github.com/AlexxIT/go2rtc/cmd/srtp.Init"),
 
 	// webrtc/api.go
 	[]byte("created by github.com/pion/ice/v2.NewTCPMuxDefault"),

cmd/exec/exec.go

@@ -14,6 +14,7 @@ import (
 	"os"
 	"os/exec"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -23,22 +24,22 @@ func Init() {
 		return
 	}
 
-	rtsp.OnProducer = func(prod streamer.Producer) bool {
+	rtsp.HandleFunc(func(conn *pkg.Conn) bool {
-		if conn := prod.(*pkg.Conn); conn != nil {
+		waitersMu.Lock()
-			if waiter := waiters[conn.URL.Path]; waiter != nil {
+		waiter := waiters[conn.URL.Path]
-				waiter <- prod
+		waitersMu.Unlock()
-				return true
+
-			}
+		if waiter == nil {
+			return false
 		}
-		return false
+
-	}
+		waiter <- conn
+		return true
+	})
 
 	streams.HandleFunc("exec", Handle)
 
 	log = app.GetLogger("exec")
-
-	// TODO: add sync.Mutex
-	waiters = map[string]chan streamer.Producer{}
 }
 
 func Handle(url string) (streamer.Producer, error) {
@@ -60,8 +61,15 @@ func Handle(url string) (streamer.Producer, error) {
 
 	ch := make(chan streamer.Producer)
 
+	waitersMu.Lock()
 	waiters[path] = ch
-	defer delete(waiters, path)
+	waitersMu.Unlock()
+
+	defer func() {
+		waitersMu.Lock()
+		delete(waiters, path)
+		waitersMu.Unlock()
+	}()
 
 	log.Debug().Str("url", url).Msg("[exec] run")
 
@@ -86,4 +94,5 @@ func Handle(url string) (streamer.Producer, error) {
 // internal
 
 var log zerolog.Logger
-var waiters map[string]chan streamer.Producer
+var waiters = map[string]chan streamer.Producer{}
+var waitersMu sync.Mutex

cmd/ffmpeg/ffmpeg.go

@@ -23,7 +23,7 @@ func Init() {
 		// inputs
 		"file": "-re -stream_loop -1 -i {input}",
 		"http": "-fflags nobuffer -flags low_delay -i {input}",
-		"rtsp": "-fflags nobuffer -flags low_delay -rtsp_transport tcp -i {input}",
+		"rtsp": "-fflags nobuffer -flags low_delay -rtsp_transport tcp -timeout 5000000 -i {input}",
 
 		// output
 		"output": "-rtsp_transport tcp -f rtsp {output}",
@@ -121,7 +121,7 @@ func Init() {
 
 			for _, audio := range query["audio"] {
 				if audio == "copy" {
-					s += " -codec:v copy"
+					s += " -codec:a copy"
 				} else {
 					s += " " + tpl[audio]
 				}

cmd/hass/api.go

@@ -55,6 +55,10 @@ func initAPI() {
 		// /stream/{id}/channel/0/webrtc
 		default:
 			i := strings.IndexByte(r.RequestURI[8:], '/')
+			if i <= 0 {
+				log.Warn().Msgf("wrong request: %s", r.RequestURI)
+				return
+			}
 			name := r.RequestURI[8 : 8+i]
 
 			stream := streams.Get(name)

cmd/mjpeg/mjpeg.go

@@ -10,12 +10,47 @@ import (
 )
 
 func Init() {
-	api.HandleFunc("api/stream.mjpeg", handler)
+	api.HandleFunc("api/frame.jpeg", handlerKeyframe)
+	api.HandleFunc("api/stream.mjpeg", handlerStream)
+}
+
+func handlerKeyframe(w http.ResponseWriter, r *http.Request) {
+	src := r.URL.Query().Get("src")
+	stream := streams.GetOrNew(src)
+	if stream == nil {
+		return
+	}
+
+	exit := make(chan []byte)
+
+	cons := &mjpeg.Consumer{}
+	cons.Listen(func(msg interface{}) {
+		switch msg := msg.(type) {
+		case []byte:
+			exit <- msg
+		}
+	})
+
+	if err := stream.AddConsumer(cons); err != nil {
+		log.Error().Err(err).Caller().Send()
+		return
+	}
+
+	data := <-exit
+
+	stream.RemoveConsumer(cons)
+
+	w.Header().Set("Content-Type", "image/jpeg")
+	w.Header().Set("Content-Length", strconv.Itoa(len(data)))
+
+	if _, err := w.Write(data); err != nil {
+		log.Error().Err(err).Caller().Send()
+	}
 }
 
 const header = "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: "
 
-func handler(w http.ResponseWriter, r *http.Request) {
+func handlerStream(w http.ResponseWriter, r *http.Request) {
 	src := r.URL.Query().Get("src")
 	stream := streams.GetOrNew(src)
 	if stream == nil {

cmd/rtmp/rtmp.go

@@ -8,6 +8,8 @@ import (
 
 func Init() {
 	streams.HandleFunc("rtmp", handle)
+	streams.HandleFunc("http", handle)
+	streams.HandleFunc("https", handle)
 }
 
 func handle(url string) (streamer.Producer, error) {

cmd/rtsp/rtsp.go

@@ -32,20 +32,43 @@ func Init() {
 
 	// RTSP server support
 	address := conf.Mod.Listen
-	if address != "" {
+	if address == "" {
-		_, Port, _ = net.SplitHostPort(address)
+		return
-
-		go worker(address)
 	}
+
+	ln, err := net.Listen("tcp", address)
+	if err != nil {
+		log.Error().Err(err).Msg("[rtsp] listen")
+		return
+	}
+
+	_, Port, _ = net.SplitHostPort(address)
+
+	log.Info().Str("addr", address).Msg("[rtsp] listen")
+
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go tcpHandler(conn)
+		}
+	}()
+}
+
+type Handler func(conn *rtsp.Conn) bool
+
+func HandleFunc(handler Handler) {
+	handlers = append(handlers, handler)
 }
 
 var Port string
 
-var OnProducer func(conn streamer.Producer) bool // TODO: maybe rewrite...
-
 // internal
 
 var log zerolog.Logger
+var handlers []Handler
 
 func rtspHandler(url string) (streamer.Producer, error) {
 	backchannel := true
@@ -84,10 +107,10 @@ func rtspHandler(url string) (streamer.Producer, error) {
 		}
 
 		// second try without backchannel, we need to reconnect
+		conn.Backchannel = false
 		if err = conn.Dial(); err != nil {
 			return nil, err
 		}
-		conn.Backchannel = false
 		if err = conn.Describe(); err != nil {
 			return nil, err
 		}
@@ -96,101 +119,89 @@ func rtspHandler(url string) (streamer.Producer, error) {
 	return conn, nil
 }
 
-func worker(address string) {
+func tcpHandler(c net.Conn) {
-	srv, err := tcp.NewServer(address)
+	var name string
-	if err != nil {
+	var closer func()
-		log.Error().Err(err).Msg("[rtsp] listen")
-		return
-	}
 
-	log.Info().Str("addr", address).Msg("[rtsp] listen")
+	trace := log.Trace().Enabled()
 
-	srv.Listen(func(msg interface{}) {
+	conn := rtsp.NewServer(c)
-		switch msg.(type) {
+	conn.Listen(func(msg interface{}) {
-		case net.Conn:
+		if trace {
-			var name string
+			switch msg := msg.(type) {
-			var onDisconnect func()
+			case *tcp.Request:
+				log.Trace().Msgf("[rtsp] server request:\n%s", msg)
+			case *tcp.Response:
+				log.Trace().Msgf("[rtsp] server response:\n%s", msg)
+			}
+		}
 
-			trace := log.Trace().Enabled()
+		switch msg {
+		case rtsp.MethodDescribe:
+			name = conn.URL.Path[1:]
 
-			conn := rtsp.NewServer(msg.(net.Conn))
+			stream := streams.Get(name)
-			conn.Listen(func(msg interface{}) {
+			if stream == nil {
-				if trace {
-					switch msg := msg.(type) {
-					case *tcp.Request:
-						log.Trace().Msgf("[rtsp] server request:\n%s", msg)
-					case *tcp.Response:
-						log.Trace().Msgf("[rtsp] server response:\n%s", msg)
-					}
-				}
-
-				switch msg {
-				case rtsp.MethodDescribe:
-					name = conn.URL.Path[1:]
-
-					log.Debug().Str("stream", name).Msg("[rtsp] new consumer")
-
-					stream := streams.Get(name) // TODO: rewrite
-					if stream == nil {
-						return
-					}
-
-					initMedias(conn)
-
-					if err = stream.AddConsumer(conn); err != nil {
-						log.Warn().Err(err).Str("stream", name).Msg("[rtsp]")
-						return
-					}
-
-					onDisconnect = func() {
-						stream.RemoveConsumer(conn)
-					}
-
-				case rtsp.MethodAnnounce:
-					if OnProducer != nil {
-						if OnProducer(conn) {
-							return
-						}
-					}
-
-					name = conn.URL.Path[1:]
-
-					log.Debug().Str("stream", name).Msg("[rtsp] new producer")
-
-					stream := streams.Get(name)
-					if stream == nil {
-						return
-					}
-
-					stream.AddProducer(conn)
-
-					onDisconnect = func() {
-						stream.RemoveProducer(conn)
-					}
-
-				case streamer.StatePlaying:
-					log.Debug().Str("stream", name).Msg("[rtsp] start")
-				}
-			})
-
-			if err = conn.Accept(); err != nil {
-				log.Warn().Err(err).Msg("[rtsp] accept")
 				return
 			}
 
-			if err = conn.Handle(); err != nil {
+			log.Debug().Str("stream", name).Msg("[rtsp] new consumer")
-				//log.Warn().Err(err).Msg("[rtsp] handle server")
+
+			initMedias(conn)
+
+			if err := stream.AddConsumer(conn); err != nil {
+				log.Warn().Err(err).Str("stream", name).Msg("[rtsp]")
+				return
 			}
 
-			if onDisconnect != nil {
+			closer = func() {
-				onDisconnect()
+				stream.RemoveConsumer(conn)
 			}
 
-			log.Debug().Str("stream", name).Msg("[rtsp] disconnect")
+		case rtsp.MethodAnnounce:
+			name = conn.URL.Path[1:]
+
+			stream := streams.Get(name)
+			if stream == nil {
+				return
+			}
+
+			log.Debug().Str("stream", name).Msg("[rtsp] new producer")
+
+			stream.AddProducer(conn)
+
+			closer = func() {
+				stream.RemoveProducer(conn)
+			}
+
+		case streamer.StatePlaying:
+			log.Debug().Str("stream", name).Msg("[rtsp] start")
 		}
 	})
 
-	srv.Serve()
+	if err := conn.Accept(); err != nil {
+		log.Warn().Err(err).Caller().Send()
+		_ = conn.Close()
+		return
+	}
+
+	for _, handler := range handlers {
+		if handler(conn) {
+			return
+		}
+	}
+
+	if closer != nil {
+		if err := conn.Handle(); err != nil {
+			log.Debug().Err(err).Caller().Send()
+		}
+
+		closer()
+
+		log.Debug().Str("stream", name).Msg("[rtsp] disconnect")
+	}
+
+	_ = conn.Close()
 }
 
 func initMedias(conn *rtsp.Conn) {

cmd/streams/handlers.go

@@ -4,30 +4,36 @@ import (
 	"fmt"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"strings"
+	"sync"
 )
 
 type Handler func(url string) (streamer.Producer, error)
 
-var handlers map[string]Handler
+var handlers = map[string]Handler{}
+var handlersMu sync.Mutex
 
 func HandleFunc(scheme string, handler Handler) {
-	if handlers == nil {
+	handlersMu.Lock()
-		handlers = make(map[string]Handler)
-	}
 	handlers[scheme] = handler
+	handlersMu.Unlock()
+}
+
+func getHandler(url string) Handler {
+	i := strings.IndexByte(url, ':')
+	if i <= 0 { // TODO: i < 4 ?
+		return nil
+	}
+	handlersMu.Lock()
+	defer handlersMu.Unlock()
+	return handlers[url[:i]]
 }
 
 func HasProducer(url string) bool {
-	i := strings.IndexByte(url, ':')
+	return getHandler(url) != nil
-	if i <= 0 { // TODO: i < 4 ?
-		return false
-	}
-	return handlers[url[:i]] != nil
 }
 
 func GetProducer(url string) (streamer.Producer, error) {
-	i := strings.IndexByte(url, ':')
+	handler := getHandler(url)
-	handler := handlers[url[:i]]
 	if handler == nil {
 		return nil, fmt.Errorf("unsupported scheme: %s", url)
 	}

cmd/streams/producer.go

@@ -4,6 +4,7 @@ import (
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"strings"
 	"sync"
+	"time"
 )
 
 type state byte
@@ -24,8 +25,9 @@ type Producer struct {
 	element streamer.Producer
 	tracks  []*streamer.Track
 
-	state state
+	state   state
-	mx    sync.Mutex
+	mu      sync.Mutex
+	restart *time.Timer
 }
 
 func (p *Producer) SetSource(s string) {
@@ -36,16 +38,16 @@ func (p *Producer) SetSource(s string) {
 }
 
 func (p *Producer) GetMedias() []*streamer.Media {
-	p.mx.Lock()
+	p.mu.Lock()
-	defer p.mx.Unlock()
+	defer p.mu.Unlock()
 
 	if p.state == stateNone {
-		log.Debug().Str("url", p.url).Msg("[streams] probe producer")
+		log.Debug().Msgf("[streams] probe producer url=%s", p.url)
 
 		var err error
 		p.element, err = GetProducer(p.url)
 		if err != nil || p.element == nil {
-			log.Error().Err(err).Str("url", p.url).Msg("[streams] probe producer")
+			log.Error().Err(err).Caller().Send()
 			return nil
 		}
 
@@ -56,14 +58,17 @@ func (p *Producer) GetMedias() []*streamer.Media {
 }
 
 func (p *Producer) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.Track {
-	p.mx.Lock()
+	p.mu.Lock()
-	defer p.mx.Unlock()
+	defer p.mu.Unlock()
 
-	if p.state == stateMedias {
+	if p.state == stateNone {
-		p.state = stateTracks
+		return nil
 	}
 
 	track := p.element.GetTrack(media, codec)
+	if track == nil {
+		return nil
+	}
 
 	for _, t := range p.tracks {
 		if track == t {
@@ -71,6 +76,10 @@ func (p *Producer) GetTrack(media *streamer.Media, codec *streamer.Codec) *strea
 		}
 	}
 
+	if p.state == stateMedias {
+		p.state = stateTracks
+	}
+
 	p.tracks = append(p.tracks, track)
 
 	return track
@@ -79,36 +88,89 @@ func (p *Producer) GetTrack(media *streamer.Media, codec *streamer.Codec) *strea
 // internals
 
 func (p *Producer) start() {
-	p.mx.Lock()
+	p.mu.Lock()
-	defer p.mx.Unlock()
+	defer p.mu.Unlock()
 
 	if p.state != stateTracks {
 		return
 	}
 
-	log.Debug().Str("url", p.url).Msg("[streams] start producer")
+	log.Debug().Msgf("[streams] start producer url=%s", p.url)
 
 	p.state = stateStart
 	go func() {
+		// safe read element while mu locked
 		if err := p.element.Start(); err != nil {
-			log.Warn().Err(err).Str("url", p.url).Msg("[streams] start")
+			log.Warn().Err(err).Caller().Send()
 		}
+		p.reconnect()
+	}()
+}
+
+func (p *Producer) reconnect() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.state != stateStart {
+		log.Trace().Msgf("[streams] stop reconnect url=%s", p.url)
+		return
+	}
+
+	log.Debug().Msgf("[streams] reconnect to url=%s", p.url)
+
+	var err error
+	p.element, err = GetProducer(p.url)
+	if err != nil || p.element == nil {
+		log.Debug().Err(err).Caller().Send()
+		// TODO: dynamic timeout
+		p.restart = time.AfterFunc(30*time.Second, p.reconnect)
+		return
+	}
+
+	medias := p.element.GetMedias()
+
+	// convert all old producer tracks to new tracks
+	for i, oldTrack := range p.tracks {
+		// match new element medias with old track codec
+		for _, media := range medias {
+			codec := media.MatchCodec(oldTrack.Codec)
+			if codec == nil {
+				continue
+			}
+
+			// move sink from old track to new track
+			newTrack := p.element.GetTrack(media, codec)
+			newTrack.GetSink(oldTrack)
+			p.tracks[i] = newTrack
+
+			break
+		}
+	}
+
+	go func() {
+		if err = p.element.Start(); err != nil {
+			log.Debug().Err(err).Caller().Send()
+		}
+		p.reconnect()
 	}()
 }
 
 func (p *Producer) stop() {
-	p.mx.Lock()
+	p.mu.Lock()
 
-	log.Debug().Str("url", p.url).Msg("[streams] stop producer")
+	log.Debug().Msgf("[streams] stop producer url=%s", p.url)
 
 	if p.element != nil {
 		_ = p.element.Stop()
 		p.element = nil
-	} else {
-		log.Warn().Str("url", p.url).Msg("[streams] stop empty producer")
 	}
-	p.tracks = nil
+	if p.restart != nil {
-	p.state = stateNone
+		p.restart.Stop()
+		p.restart = nil
+	}
 
-	p.mx.Unlock()
+	p.state = stateNone
+	p.tracks = nil
+
+	p.mu.Unlock()
 }

cmd/streams/stream.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"sync"
 )
 
 type Consumer struct {
@@ -14,6 +15,7 @@ type Consumer struct {
 type Stream struct {
 	producers []*Producer
 	consumers []*Consumer
+	mu        sync.Mutex
 }
 
 func NewStream(source interface{}) *Stream {
@@ -51,18 +53,19 @@ func (s *Stream) AddConsumer(cons streamer.Consumer) (err error) {
 	ic := len(s.consumers)
 
 	consumer := &Consumer{element: cons}
+	var producers []*Producer // matched producers for consumer
 
 	// Step 1. Get consumer medias
 	for icc, consMedia := range cons.GetMedias() {
 		log.Trace().Stringer("media", consMedia).
-			Msgf("[streams] consumer:%d:%d candidate", ic, icc)
+			Msgf("[streams] consumer=%d candidate=%d", ic, icc)
 
 	producers:
 		for ip, prod := range s.producers {
 			// Step 2. Get producer medias (not tracks yet)
 			for ipc, prodMedia := range prod.GetMedias() {
 				log.Trace().Stringer("media", prodMedia).
-					Msgf("[streams] producer:%d:%d candidate", ip, ipc)
+					Msgf("[streams] producer=%d candidate=%d", ip, ipc)
 
 				// Step 3. Match consumer/producer codecs list
 				prodCodec := prodMedia.MatchMedia(consMedia)
@@ -81,20 +84,23 @@ func (s *Stream) AddConsumer(cons streamer.Consumer) (err error) {
 					consTrack := consumer.element.AddTrack(consMedia, prodTrack)
 
 					consumer.tracks = append(consumer.tracks, consTrack)
+					producers = append(producers, prod)
 					break producers
 				}
 			}
 		}
 	}
 
-	// can't match tracks for consumer
+	if len(producers) == 0 {
-	if len(consumer.tracks) == 0 {
 		return errors.New("couldn't find the matching tracks")
 	}
 
+	s.mu.Lock()
 	s.consumers = append(s.consumers, consumer)
+	s.mu.Unlock()
 
-	for _, prod := range s.producers {
+	// there may be duplicates, but that's not a problem
+	for _, prod := range producers {
 		prod.start()
 	}
 
@@ -102,6 +108,7 @@ func (s *Stream) AddConsumer(cons streamer.Consumer) (err error) {
 }
 
 func (s *Stream) RemoveConsumer(cons streamer.Consumer) {
+	s.mu.Lock()
 	for i, consumer := range s.consumers {
 		if consumer == nil {
 			log.Warn().Msgf("empty consumer: %+v\n", s)
@@ -127,7 +134,7 @@ func (s *Stream) RemoveConsumer(cons streamer.Consumer) {
 
 		var sink bool
 		for _, track := range producer.tracks {
-			if len(track.Sink) > 0 {
+			if track.HasSink() {
 				sink = true
 			}
 		}
@@ -135,38 +142,44 @@ func (s *Stream) RemoveConsumer(cons streamer.Consumer) {
 			producer.stop()
 		}
 	}
+	s.mu.Unlock()
 }
 
 func (s *Stream) AddProducer(prod streamer.Producer) {
 	producer := &Producer{element: prod, state: stateTracks}
+	s.mu.Lock()
 	s.producers = append(s.producers, producer)
+	s.mu.Unlock()
 }
 
 func (s *Stream) RemoveProducer(prod streamer.Producer) {
+	s.mu.Lock()
 	for i, producer := range s.producers {
 		if producer.element == prod {
 			s.removeProducer(i)
 			break
 		}
 	}
+	s.mu.Unlock()
 }
 
-func (s *Stream) Active() bool {
+//func (s *Stream) Active() bool {
-	if len(s.consumers) > 0 {
+//	if len(s.consumers) > 0 {
-		return true
+//		return true
-	}
+//	}
-
+//
-	for _, prod := range s.producers {
+//	for _, prod := range s.producers {
-		if prod.element != nil {
+//		if prod.element != nil {
-			return true
+//			return true
-		}
+//		}
-	}
+//	}
-
+//
-	return false
+//	return false
-}
+//}
 
 func (s *Stream) MarshalJSON() ([]byte, error) {
 	var v []interface{}
+	s.mu.Lock()
 	for _, prod := range s.producers {
 		if prod.element != nil {
 			v = append(v, prod.element)
@@ -176,6 +189,7 @@ func (s *Stream) MarshalJSON() ([]byte, error) {
 		// cons.element always not nil
 		v = append(v, cons.element)
 	}
+	s.mu.Unlock()
 	if len(v) == 0 {
 		v = nil
 	}

cmd/webrtc/webrtc.go

@@ -8,7 +8,9 @@ import (
 	"github.com/AlexxIT/go2rtc/pkg/webrtc"
 	pion "github.com/pion/webrtc/v3"
 	"github.com/rs/zerolog"
+	"io/ioutil"
 	"net"
+	"net/http"
 )
 
 func Init() {
@@ -55,6 +57,8 @@ func Init() {
 
 	api.HandleWS(webrtc.MsgTypeOffer, offerHandler)
 	api.HandleWS(webrtc.MsgTypeCandidate, candidateHandler)
+
+	api.HandleFunc("api/webrtc", syncHandler)
 }
 
 var Port string
@@ -137,6 +141,32 @@ func offerHandler(ctx *api.Context, msg *streamer.Message) {
 	ctx.Consumer = conn
 }
 
+func syncHandler(w http.ResponseWriter, r *http.Request) {
+	url := r.URL.Query().Get("src")
+	stream := streams.Get(url)
+	if stream == nil {
+		return
+	}
+
+	// get offer
+	offer, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		log.Error().Err(err).Caller().Send()
+		return
+	}
+
+	answer, err := ExchangeSDP(stream, string(offer), r.UserAgent())
+	if err != nil {
+		log.Error().Err(err).Caller().Send()
+		return
+	}
+
+	// send SDP to client
+	if _, err = w.Write([]byte(answer)); err != nil {
+		log.Error().Err(err).Caller().Send()
+	}
+}
+
 func ExchangeSDP(
 	stream *streams.Stream, offer string, userAgent string,
 ) (answer string, err error) {

pkg/h264/README.md

@@ -1,3 +1,11 @@
+# H264
+
+Access Unit (AU) can contain one or multiple NAL Unit:
+
+1. [SEI,] SPS, PPS, IFrame, [IFrame...]
+2. BFrame, [BFrame...]
+3. IFrame, [IFrame...]
+
 ## RTP H264
 
 Camera | NALu

pkg/h264/avc.go

@@ -12,49 +12,38 @@ func IsAVC(codec *streamer.Codec) bool {
 	return codec.PayloadType == PayloadTypeAVC
 }
 
-func EncodeAVC(raw []byte) (avc []byte) {
+func EncodeAVC(nals ...[]byte) (avc []byte) {
-	avc = make([]byte, len(raw)+4)
+	var i, n int
-	binary.BigEndian.PutUint32(avc, uint32(len(raw)))
+
-	copy(avc[4:], raw)
+	for _, nal := range nals {
+		if i = len(nal); i > 0 {
+			n += 4 + i
+		}
+	}
+
+	avc = make([]byte, n)
+
+	n = 0
+	for _, nal := range nals {
+		if i = len(nal); i > 0 {
+			binary.BigEndian.PutUint32(avc[n:], uint32(i))
+			n += 4 + copy(avc[n+4:], nal)
+		}
+	}
+
 	return
 }
 
 func RepairAVC(track *streamer.Track) streamer.WrapperFunc {
 	sps, pps := GetParameterSet(track.Codec.FmtpLine)
-	sps = EncodeAVC(sps)
+	ps := EncodeAVC(sps, pps)
-	pps = EncodeAVC(pps)
 
 	return func(push streamer.WriterFunc) streamer.WriterFunc {
 		return func(packet *rtp.Packet) (err error) {
-			naluType := NALUType(packet.Payload)
+			if NALUType(packet.Payload) == NALUTypeIFrame {
-			switch naluType {
+				packet.Payload = Join(ps, packet.Payload)
-			case NALUTypeSPS:
-				sps = packet.Payload
-				return
-			case NALUTypePPS:
-				pps = packet.Payload
-				return
 			}
-
+			return push(packet)
-			var clone rtp.Packet
-
-			if naluType == NALUTypeIFrame {
-				clone = *packet
-				clone.Payload = sps
-				if err = push(&clone); err != nil {
-					return
-				}
-
-				clone = *packet
-				clone.Payload = pps
-				if err = push(&clone); err != nil {
-					return
-				}
-			}
-
-			clone = *packet
-			clone.Payload = packet.Payload
-			return push(&clone)
 		}
 	}
 }
@@ -63,12 +52,12 @@ func SplitAVC(data []byte) [][]byte {
 	var nals [][]byte
 	for {
 		// get AVC length
-		size := int(binary.BigEndian.Uint32(data))
+		size := int(binary.BigEndian.Uint32(data)) + 4
 
 		// check if multiple items in one packet
-		if size+4 < len(data) {
+		if size < len(data) {
-			nals = append(nals, data[:size+4])
+			nals = append(nals, data[:size])
-			data = data[size+4:]
+			data = data[size:]
 		} else {
 			nals = append(nals, data)
 			break
@@ -76,3 +65,18 @@ func SplitAVC(data []byte) [][]byte {
 	}
 	return nals
 }
+
+func Types(data []byte) []byte {
+	var types []byte
+	for {
+		types = append(types, NALUType(data))
+
+		size := 4 + int(binary.BigEndian.Uint32(data))
+		if size < len(data) {
+			data = data[size:]
+		} else {
+			break
+		}
+	}
+	return types
+}

pkg/h264/helper.go

@@ -2,24 +2,49 @@ package h264
 
 import (
 	"encoding/base64"
+	"encoding/binary"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"strings"
 )
 
 const (
-	NALUTypePFrame = 1
+	NALUTypePFrame = 1 // Coded slice of a non-IDR picture
-	NALUTypeIFrame = 5
+	NALUTypeIFrame = 5 // Coded slice of an IDR picture
-	NALUTypeSEI    = 6
+	NALUTypeSEI    = 6 // Supplemental enhancement information (SEI)
-	NALUTypeSPS    = 7
+	NALUTypeSPS    = 7 // Sequence parameter set
-	NALUTypePPS    = 8
+	NALUTypePPS    = 8 // Picture parameter set
+	NALUTypeAUD    = 9 // Access unit delimiter
 )
 
 func NALUType(b []byte) byte {
 	return b[4] & 0x1F
 }
 
+// IsKeyframe - check if any NALU in one AU is Keyframe
 func IsKeyframe(b []byte) bool {
-	return NALUType(b) == NALUTypeIFrame
+	for {
+		switch NALUType(b) {
+		case NALUTypePFrame:
+			return false
+		case NALUTypeIFrame:
+			return true
+		}
+
+		size := int(binary.BigEndian.Uint32(b)) + 4
+		if size < len(b) {
+			b = b[size:]
+			continue
+		} else {
+			return false
+		}
+	}
+}
+
+func Join(ps, iframe []byte) []byte {
+	b := make([]byte, len(ps)+len(iframe))
+	i := copy(b, ps)
+	copy(b[i:], iframe)
+	return b
 }
 
 func GetProfileLevelID(fmtp string) string {

pkg/h264/rtp.go

@@ -13,97 +13,69 @@ func RTPDepay(track *streamer.Track) streamer.WrapperFunc {
 	depack := &codecs.H264Packet{IsAVC: true}
 
 	sps, pps := GetParameterSet(track.Codec.FmtpLine)
-	sps = EncodeAVC(sps)
+	ps := EncodeAVC(sps, pps)
-	pps = EncodeAVC(pps)
 
-	var buffer []byte
+	buf := make([]byte, 0, 512*1024) // 512K
 
 	return func(push streamer.WriterFunc) streamer.WriterFunc {
 		return func(packet *rtp.Packet) error {
-			//nalUnitType := packet.Payload[0] & 0x1F
+			//log.Printf("[RTP] codec: %s, nalu: %2d, size: %6d, ts: %10d, pt: %2d, ssrc: %d, seq: %d, %v", track.Codec.Name, packet.Payload[0]&0x1F, len(packet.Payload), packet.Timestamp, packet.PayloadType, packet.SSRC, packet.SequenceNumber, packet.Marker)
-			//fmt.Printf(
-			//	"[RTP] codec: %s, nalu: %2d, size: %6d, ts: %10d, pt: %2d, ssrc: %d, seq: %d\n",
-			//	track.Codec.Name, nalUnitType, len(packet.Payload), packet.Timestamp,
-			//	packet.PayloadType, packet.SSRC, packet.SequenceNumber,
-			//)
 
-			data, err := depack.Unmarshal(packet.Payload)
+			payload, err := depack.Unmarshal(packet.Payload)
-			if len(data) == 0 || err != nil {
+			if len(payload) == 0 || err != nil {
 				return nil
 			}
 
-			for {
+			// Fix TP-Link Tapo TC70: sends SPS and PPS with packet.Marker = true
-				unitType := NALUType(data)
+			if packet.Marker {
-				//fmt.Printf("[H264] nalu: %2d, size: %6d\n", unitType, len(data))
+				switch NALUType(payload) {
-
+				case NALUTypeSPS, NALUTypePPS:
-				// multiple 5 and 1 in one payload is OK
+					buf = append(buf, payload...)
-				if unitType != NALUTypeIFrame && unitType != NALUTypePFrame {
-					i := int(binary.BigEndian.Uint32(data)) + 4
-					if i < len(data) {
-						data0 := data[:i] // NAL Unit with AVC header
-						data = data[i:]
-						switch unitType {
-						case NALUTypeSPS:
-							sps = data0
-							continue
-						case NALUTypePPS:
-							pps = data0
-							continue
-						case NALUTypeSEI:
-							// some unnecessary text information
-							continue
-						}
-					}
-				}
-
-				switch unitType {
-				case NALUTypeSPS:
-					sps = data
-					return nil
-				case NALUTypePPS:
-					pps = data
-					return nil
-				case NALUTypeSEI:
-					// some unnecessary text information
 					return nil
 				}
-
-				// ffmpeg with `-tune zerolatency` enable option `-x264opts sliced-threads=1`
-				// and every NALU will be sliced to multiple NALUs
-				if !packet.Marker {
-					buffer = append(buffer, data...)
-					return nil
-				}
-
-				if buffer != nil {
-					buffer = append(buffer, data...)
-					data = buffer
-					buffer = nil
-				}
-
-				var clone rtp.Packet
-
-				if unitType == NALUTypeIFrame {
-					clone = *packet
-					clone.Version = RTPPacketVersionAVC
-					clone.Payload = sps
-					if err = push(&clone); err != nil {
-						return err
-					}
-
-					clone = *packet
-					clone.Version = RTPPacketVersionAVC
-					clone.Payload = pps
-					if err = push(&clone); err != nil {
-						return err
-					}
-				}
-
-				clone = *packet
-				clone.Version = RTPPacketVersionAVC
-				clone.Payload = data
-				return push(&clone)
 			}
+
+			if len(buf) == 0 {
+				// Amcrest IP4M-1051: 9, 7, 8, 6, 28...
+				// Amcrest IP4M-1051: 9, 6, 1
+				switch NALUType(payload) {
+				case NALUTypeIFrame:
+					// fix IFrame without SPS,PPS
+					buf = append(buf, ps...)
+				case NALUTypeSEI, NALUTypeAUD:
+					// fix ffmpeg with transcoding first frame
+					i := int(4 + binary.BigEndian.Uint32(payload))
+
+					// check if only one NAL (fix ffmpeg transcoding for Reolink RLC-510A)
+					if i == len(payload) {
+						return nil
+					}
+
+					payload = payload[i:]
+
+					if NALUType(payload) == NALUTypeIFrame {
+						buf = append(buf, ps...)
+					}
+				}
+			}
+
+			// collect all NALs for Access Unit
+			if !packet.Marker {
+				buf = append(buf, payload...)
+				return nil
+			}
+
+			if len(buf) > 0 {
+				payload = append(buf, payload...)
+				buf = buf[:0]
+			}
+
+			//log.Printf("[AVC] %v, len: %d", Types(payload), len(payload))
+
+			clone := *packet
+			clone.Version = RTPPacketVersionAVC
+			clone.Payload = payload
+			return push(&clone)
 		}
 	}
 }
@@ -117,11 +89,12 @@ func RTPPay(mtu uint16) streamer.WrapperFunc {
 		return func(packet *rtp.Packet) error {
 			if packet.Version == RTPPacketVersionAVC {
 				payloads := payloader.Payload(mtu, packet.Payload)
+				last := len(payloads) - 1
 				for i, payload := range payloads {
 					clone := rtp.Packet{
 						Header: rtp.Header{
 							Version: 2,
-							Marker:  i == len(payloads)-1,
+							Marker:  i == last,
 							//PayloadType:    packet.PayloadType,
 							SequenceNumber: sequencer.NextSequenceNumber(),
 							Timestamp:      packet.Timestamp,

pkg/httpflv/README.md

@@ -0,0 +1,3 @@
+## Useful links
+
+- https://medium.com/@nate510/don-t-use-go-s-default-http-client-4804cb19f779

pkg/httpflv/httpflv.go

@@ -0,0 +1,100 @@
+package httpflv
+
+import (
+	"bufio"
+	"errors"
+	"github.com/deepch/vdk/av"
+	"github.com/deepch/vdk/codec/h264parser"
+	"github.com/deepch/vdk/format/flv/flvio"
+	"github.com/deepch/vdk/utils/bits/pio"
+	"io"
+	"net/http"
+)
+
+func Dial(uri string) (*Conn, error) {
+	req, err := http.NewRequest("GET", uri, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	c := Conn{
+		conn:   res.Body,
+		reader: bufio.NewReaderSize(res.Body, pio.RecommendBufioSize),
+		buf:    make([]byte, 256),
+	}
+
+	if _, err = io.ReadFull(c.reader, c.buf[:flvio.FileHeaderLength]); err != nil {
+		return nil, err
+	}
+
+	flags, n, err := flvio.ParseFileHeader(c.buf)
+	if err != nil {
+		return nil, err
+	}
+
+	if flags&flvio.FILE_HAS_VIDEO == 0 {
+		return nil, errors.New("not supported")
+	}
+
+	if _, err = c.reader.Discard(n); err != nil {
+		return nil, err
+	}
+
+	return &c, nil
+}
+
+type Conn struct {
+	conn      io.ReadCloser
+	reader    *bufio.Reader
+	buf       []byte
+}
+
+func (c *Conn) Streams() ([]av.CodecData, error) {
+	for {
+		tag, _, err := flvio.ReadTag(c.reader, c.buf)
+		if err != nil {
+			return nil, err
+		}
+
+		if tag.Type != flvio.TAG_VIDEO || tag.AVCPacketType != flvio.AAC_SEQHDR {
+			continue
+		}
+
+		stream, err := h264parser.NewCodecDataFromAVCDecoderConfRecord(tag.Data)
+		if err != nil {
+			return nil, err
+		}
+
+		return []av.CodecData{stream}, nil
+	}
+}
+
+func (c *Conn) ReadPacket() (av.Packet, error) {
+	for {
+		tag, ts, err := flvio.ReadTag(c.reader, c.buf)
+		if err != nil {
+			return av.Packet{}, err
+		}
+
+		if tag.Type != flvio.TAG_VIDEO || tag.AVCPacketType != flvio.AVC_NALU {
+			continue
+		}
+
+		return av.Packet{
+			Idx:             0,
+			Data:            tag.Data,
+			CompositionTime: flvio.TsToTime(tag.CompositionTime),
+			IsKeyFrame:      tag.FrameType == flvio.FRAME_KEY,
+			Time:            flvio.TsToTime(ts),
+		}, nil
+	}
+}
+
+func (c *Conn) Close() (err error) {
+	return c.conn.Close()
+}

pkg/ivideon/client.go

@@ -245,14 +245,12 @@ func (c *Client) worker() {
 			time.Sleep(d)
 
 			// can be SPS, PPS and IFrame in one packet
-			for _, payload := range h264.SplitAVC(data[:entry.Size]) {
+			packet := &rtp.Packet{
-				packet := &rtp.Packet{
+				// ivideon clockrate=1000, RTP clockrate=90000
-					// ivideon clockrate=1000, RTP clockrate=90000
+				Header:  rtp.Header{Timestamp: ts * 90},
-					Header:  rtp.Header{Timestamp: ts * 90},
+				Payload: data[:entry.Size],
-					Payload: payload,
-				}
-				_ = track.WriteRTP(packet)
 			}
+			_ = track.WriteRTP(packet)
 
 			data = data[entry.Size:]
 			ts += entry.Duration

pkg/mjpeg/consumer.go

@@ -61,8 +61,16 @@ func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *strea
 				lqt, cqt = MakeTables(q)
 			}
 
+			// https://www.rfc-editor.org/rfc/rfc2435#section-3.1.5
+			// The maximum width is 2040 pixels.
 			w := uint16(packet.Payload[6]) << 3
 			h := uint16(packet.Payload[7]) << 3
+
+			// fix 2560x1920 and 2560x1440
+			if w == 512 && (h == 1920 || h == 1440) {
+				w = 2560
+			}
+
 			//fmt.Printf("t: %d, q: %d, w: %d, h: %d\n", t, q, w, h)
 			header = MakeHeaders(t, w, h, lqt, cqt)
 		}

pkg/mp4/consumer.go

@@ -53,15 +53,16 @@ func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *strea
 				return nil
 			}
 
-			switch h264.NALUType(packet.Payload) {
+			if c.muxer == nil {
-			case h264.NALUTypeIFrame:
+				return nil
-				c.start = true
+			}
-			case h264.NALUTypePFrame:
+
-				if !c.start {
+			if !c.start {
+				if h264.IsKeyframe(packet.Payload) {
+					c.start = true
+				} else {
 					return nil
 				}
-			default:
-				return nil
 			}
 
 			buf := c.muxer.Marshal(packet)
@@ -71,10 +72,13 @@ func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *strea
 			return nil
 		}
 
-		if !h264.IsAVC(codec) {
+		var wrapper streamer.WrapperFunc
-			wrapper := h264.RTPDepay(track)
+		if h264.IsAVC(codec) {
-			push = wrapper(push)
+			wrapper = h264.RepairAVC(track)
+		} else {
+			wrapper = h264.RTPDepay(track)
 		}
+		push = wrapper(push)
 
 		return track.Bind(push)
 

pkg/mp4/muxer.go

@@ -46,10 +46,11 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
 		case streamer.CodecH264:
 			sps, pps := h264.GetParameterSet(codec.FmtpLine)
 			if sps == nil {
-				return nil, fmt.Errorf("empty SPS: %#v", codec)
+				// some dummy SPS and PPS not a problem
+				sps = []byte{0x67, 0x42, 0x00, 0x0a, 0xf8, 0x41, 0xa2}
+				pps = []byte{0x68, 0xce, 0x38, 0x80}
 			}
 
-			// TODO: remove
 			codecData, err := h264parser.NewCodecDataFromSPSAndPPS(sps, pps)
 			if err != nil {
 				return nil, err

pkg/rtmp/client.go

@@ -5,15 +5,24 @@ import (
 	"encoding/hex"
 	"fmt"
 	"github.com/AlexxIT/go2rtc/pkg/h264"
+	"github.com/AlexxIT/go2rtc/pkg/httpflv"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/deepch/vdk/av"
 	"github.com/deepch/vdk/codec/aacparser"
 	"github.com/deepch/vdk/codec/h264parser"
 	"github.com/deepch/vdk/format/rtmp"
 	"github.com/pion/rtp"
+	"strings"
 	"time"
 )
 
+// Conn for RTMP and RTMPT (flv over HTTP)
+type Conn interface {
+	Streams() (streams []av.CodecData, err error)
+	ReadPacket() (pkt av.Packet, err error)
+	Close() (err error)
+}
+
 type Client struct {
 	streamer.Element
 
@@ -22,7 +31,7 @@ type Client struct {
 	medias []*streamer.Media
 	tracks []*streamer.Track
 
-	conn   *rtmp.Conn
+	conn   Conn
 	closed bool
 
 	receive int
@@ -33,7 +42,12 @@ func NewClient(uri string) *Client {
 }
 
 func (c *Client) Dial() (err error) {
-	c.conn, err = rtmp.Dial(c.URI)
+	if strings.HasPrefix(c.URI, "http") {
+		c.conn, err = httpflv.Dial(c.URI)
+	} else {
+		c.conn, err = rtmp.Dial(c.URI)
+	}
+
 	if err != nil {
 		return
 	}
@@ -47,10 +61,14 @@ func (c *Client) Dial() (err error) {
 	for _, stream := range streams {
 		switch stream.Type() {
 		case av.H264:
-			cd := stream.(h264parser.CodecData)
+			info := stream.(h264parser.CodecData).RecordInfo
-			fmtp := "sprop-parameter-sets=" +
+
-				base64.StdEncoding.EncodeToString(cd.RecordInfo.SPS[0]) + "," +
+			fmtp := fmt.Sprintf(
-				base64.StdEncoding.EncodeToString(cd.RecordInfo.PPS[0])
+				"profile-level-id=%02X%02X%02X;sprop-parameter-sets=%s,%s",
+				info.AVCProfileIndication, info.ProfileCompatibility, info.AVCLevelIndication,
+				base64.StdEncoding.EncodeToString(info.SPS[0]),
+				base64.StdEncoding.EncodeToString(info.PPS[0]),
+			)
 
 			codec := &streamer.Codec{
 				Name:        streamer.CodecH264,
@@ -129,22 +147,14 @@ func (c *Client) Handle() (err error) {
 
 		track := c.tracks[int(pkt.Idx)]
 
-		timestamp := uint32(pkt.Time / time.Duration(track.Codec.ClockRate))
+		// convert seconds to RTP timestamp
+		timestamp := uint32(pkt.Time * time.Duration(track.Codec.ClockRate) / time.Second)
 
-		var payloads [][]byte
+		packet := &rtp.Packet{
-		if track.Codec.Name == streamer.CodecH264 {
+			Header:  rtp.Header{Timestamp: timestamp},
-			payloads = h264.SplitAVC(pkt.Data)
+			Payload: pkt.Data,
-		} else {
-			payloads = [][]byte{pkt.Data}
-		}
-
-		for _, payload := range payloads {
-			packet := &rtp.Packet{
-				Header:  rtp.Header{Timestamp: timestamp},
-				Payload: payload,
-			}
-			_ = track.WriteRTP(packet)
 		}
+		_ = track.WriteRTP(packet)
 	}
 }
 

pkg/rtmp/streamer.go

@@ -32,7 +32,7 @@ func (c *Client) MarshalJSON() ([]byte, error) {
 	v := map[string]interface{}{
 		streamer.JSONReceive:    c.receive,
 		streamer.JSONType:       "RTMP client producer",
-		streamer.JSONRemoteAddr: c.conn.NetConn().RemoteAddr().String(),
+		//streamer.JSONRemoteAddr: c.conn.NetConn().RemoteAddr().String(),
 		"url":                   c.URI,
 	}
 	for i, media := range c.medias {

pkg/rtsp/conn.go

@@ -43,8 +43,6 @@ const (
 	ModeServerConsumer
 )
 
-const KeepAlive = time.Second * 25
-
 type Conn struct {
 	streamer.Element
 
@@ -60,6 +58,7 @@ type Conn struct {
 	// internal
 
 	auth     *tcp.Auth
+	closed   bool
 	conn     net.Conn
 	mode     Mode
 	reader   *bufio.Reader
@@ -115,9 +114,7 @@ func (c *Conn) Dial() (err error) {
 		_ = c.parseURI()
 	}
 
-	c.conn, err = net.DialTimeout(
+	c.conn, err = net.DialTimeout("tcp", c.URL.Host, time.Second*5)
-		"tcp", c.URL.Host, 10*time.Second,
-	)
 	if err != nil {
 		return
 	}
@@ -362,21 +359,25 @@ func (c *Conn) SetupMedia(
 	var res *tcp.Response
 	res, err = c.Do(req)
 	if err != nil {
-		// Dahua VTO2111D fail on this step because of backchannel
+		// some Dahua/Amcrest cameras fail here because two simultaneous
+		// backchannel connections
 		if c.Backchannel {
-			if err = c.Dial(); err != nil {
-				return nil, err
-			}
 			c.Backchannel = false
-			if err = c.Describe(); err != nil {
+			if err := c.Dial(); err != nil {
 				return nil, err
 			}
-			res, err = c.Do(req)
+			if err := c.Describe(); err != nil {
+				return nil, err
+			}
+
+			for _, newMedia := range c.Medias {
+				if newMedia.Control == media.Control {
+					return c.SetupMedia(newMedia, newMedia.Codecs[0])
+				}
+			}
 		}
 
-		if err != nil {
+		return nil, err
-			return nil, err
-		}
 	}
 
 	if c.Session == "" {
@@ -392,12 +393,16 @@ func (c *Conn) SetupMedia(
 	// we send our `interleaved`, but camera can answer with another
 
 	// Transport: RTP/AVP/TCP;unicast;interleaved=10-11;ssrc=10117CB7
-	// Transport: RTP/AVP/TCP;unicast;destination=192.168.1.123;source=192.168.10.12;interleaved=0
+	// Transport: RTP/AVP/TCP;unicast;destination=192.168.1.111;source=192.168.1.222;interleaved=0
 	// Transport: RTP/AVP/TCP;ssrc=22345682;interleaved=0-1
 	s := res.Header.Get("Transport")
 	// TODO: rewrite
 	if !strings.HasPrefix(s, "RTP/AVP/TCP;") {
-		return nil, fmt.Errorf("wrong transport: %s", s)
+		// Escam Q6 has a bug:
+		// Transport: RTP/AVP;unicast;destination=192.168.1.111;source=192.168.1.222;interleaved=0-1
+		if !strings.Contains(s, ";interleaved=") {
+			return nil, fmt.Errorf("wrong transport: %s", s)
+		}
 	}
 
 	i := strings.Index(s, "interleaved=")
@@ -451,24 +456,19 @@ func (c *Conn) Teardown() (err error) {
 }
 
 func (c *Conn) Close() error {
-	if c.conn == nil {
+	if c.closed {
 		return nil
 	}
 	if err := c.Teardown(); err != nil {
 		return err
 	}
-	conn := c.conn
+	c.closed = true
-	c.conn = nil
+	return c.conn.Close()
-	return conn.Close()
 }
 
 const transport = "RTP/AVP/TCP;unicast;interleaved="
 
 func (c *Conn) Accept() error {
-	//if c.state != StateServerInit {
-	//	panic("wrong state")
-	//}
-
 	for {
 		req, err := tcp.ReadRequest(c.reader)
 		if err != nil {
@@ -571,7 +571,7 @@ func (c *Conn) Accept() error {
 				Request: req,
 			}
 
-			if tr[:len(transport)] == transport {
+			if strings.HasPrefix(tr, transport) {
 				c.Session = "1" // TODO: fixme
 				res.Header.Set("Transport", tr[:len(transport)+3])
 			} else {
@@ -594,16 +594,44 @@ func (c *Conn) Accept() error {
 
 func (c *Conn) Handle() (err error) {
 	defer func() {
-		if c.conn == nil {
+		if c.closed {
 			err = nil
+		} else {
+			// may have gotten here because of the deadline
+			// so close the connection to stop keepalive
+			_ = c.conn.Close()
 		}
-		//c.Fire(streamer.StateNull)
 	}()
 
-	//c.Fire(streamer.StatePlaying)
+	var timeout time.Duration
-	ts := time.Now().Add(KeepAlive)
+
+	switch c.mode {
+	case ModeClientProducer:
+		// polling frames from remote RTSP Server (ex Camera)
+		timeout = time.Second * 5
+		go c.keepalive()
+
+	case ModeServerProducer:
+		// polling frames from remote RTSP Client (ex FFmpeg)
+		timeout = time.Second * 15
+
+	case ModeServerConsumer:
+		// pushing frames to remote RTSP Client (ex VLC)
+		timeout = time.Second * 60
+
+	default:
+		return fmt.Errorf("wrong RTSP conn mode: %d", c.mode)
+	}
 
 	for {
+		if c.closed {
+			return
+		}
+
+		if err = c.conn.SetReadDeadline(time.Now().Add(timeout)); err != nil {
+			return
+		}
+
 		// we can read:
 		// 1. RTP interleaved: `$` + 1B channel number + 2B size
 		// 2. RTSP response:   RTSP/1.0 200 OK
@@ -681,16 +709,19 @@ func (c *Conn) Handle() (err error) {
 
 			c.Fire(msg)
 		}
+	}
+}
 
-		// keep-alive
+func (c *Conn) keepalive() {
-		now := time.Now()
+	// TODO: rewrite to RTCP
-		if now.After(ts) {
+	req := &tcp.Request{Method: MethodOptions, URL: c.URL}
-			req := &tcp.Request{Method: MethodOptions, URL: c.URL}
+	for {
-			// don't need to wait respose on this request
+		time.Sleep(time.Second * 25)
-			if err = c.Request(req); err != nil {
+		if c.closed {
-				return err
+			return
-			}
+		}
-			ts = now.Add(KeepAlive)
+		if err := c.Request(req); err != nil {
+			return
 		}
 	}
 }
@@ -708,20 +739,16 @@ func (c *Conn) bindTrack(
 	track *streamer.Track, channel uint8, payloadType uint8,
 ) *streamer.Track {
 	push := func(packet *rtp.Packet) error {
-		if c.conn == nil {
+		if c.closed {
 			return nil
 		}
 		packet.Header.PayloadType = payloadType
-		//packet.Header.PayloadType = 100
-		//packet.Header.PayloadType = 8
-		//packet.Header.PayloadType = 106
 
 		size := packet.MarshalSize()
 
 		data := make([]byte, 4+size)
 		data[0] = '$'
 		data[1] = channel
-		//data[1] = 10
 		binary.BigEndian.PutUint16(data[2:], uint16(size))
 
 		if _, err := packet.MarshalTo(data[4:]); err != nil {

pkg/rtsp/streamer.go

@@ -2,6 +2,7 @@ package rtsp
 
 import (
 	"encoding/json"
+	"fmt"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"strconv"
 )
@@ -27,13 +28,16 @@ func (c *Conn) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.
 }
 
 func (c *Conn) Start() error {
-	if c.mode == ModeServerProducer {
+	switch c.mode {
-		return nil
+	case ModeClientProducer:
+		if err := c.Play(); err != nil {
+			return err
+		}
+	case ModeServerProducer:
+	default:
+		return fmt.Errorf("start wrong mode: %d", c.mode)
 	}
 
-	if err := c.Play(); err != nil {
-		return err
-	}
 	return c.Handle()
 }
 

pkg/streamer/media.go

@@ -75,13 +75,13 @@ func (m *Media) AV() bool {
 	return m.Kind == KindVideo || m.Kind == KindAudio
 }
 
-func (m *Media) MatchCodec(codec *Codec) bool {
+func (m *Media) MatchCodec(codec *Codec) *Codec {
 	for _, c := range m.Codecs {
 		if c.Match(codec) {
-			return true
+			return c
 		}
 	}
-	return false
+	return nil
 }
 
 func (m *Media) MatchMedia(media *Media) *Codec {

pkg/streamer/track.go

@@ -12,44 +12,54 @@ type WrapperFunc func(push WriterFunc) WriterFunc
 type Track struct {
 	Codec     *Codec
 	Direction string
-	Sink      map[*Track]WriterFunc
+	sink      map[*Track]WriterFunc
-	mx        sync.Mutex
+	sinkMu    sync.Mutex
 }
 
 func (t *Track) String() string {
 	s := t.Codec.String()
-	s += fmt.Sprintf(", sinks=%d", len(t.Sink))
+	s += fmt.Sprintf(", sinks=%d", len(t.sink))
 	return s
 }
 
 func (t *Track) WriteRTP(p *rtp.Packet) error {
-	t.mx.Lock()
+	t.sinkMu.Lock()
-	for _, f := range t.Sink {
+	for _, f := range t.sink {
 		_ = f(p)
 	}
-	t.mx.Unlock()
+	t.sinkMu.Unlock()
 	return nil
 }
 
 func (t *Track) Bind(w WriterFunc) *Track {
-	t.mx.Lock()
+	t.sinkMu.Lock()
 
-	if t.Sink == nil {
+	if t.sink == nil {
-		t.Sink = map[*Track]WriterFunc{}
+		t.sink = map[*Track]WriterFunc{}
 	}
 
 	clone := &Track{
-		Codec: t.Codec, Direction: t.Direction, Sink: t.Sink,
+		Codec: t.Codec, Direction: t.Direction, sink: t.sink,
 	}
-	t.Sink[clone] = w
+	t.sink[clone] = w
 
-	t.mx.Unlock()
+	t.sinkMu.Unlock()
 
 	return clone
 }
 
 func (t *Track) Unbind() {
-	t.mx.Lock()
+	t.sinkMu.Lock()
-	delete(t.Sink, t)
+	delete(t.sink, t)
-	t.mx.Unlock()
+	t.sinkMu.Unlock()
+}
+
+func (t *Track) GetSink(from *Track) {
+	t.sink = from.sink
+}
+
+func (t *Track) HasSink() bool {
+	t.sinkMu.Lock()
+	defer t.sinkMu.Unlock()
+	return len(t.sink) > 0
 }

pkg/webrtc/helper.go

@@ -9,6 +9,7 @@ import (
 	"net"
 	"strconv"
 	"strings"
+	"time"
 )
 
 func NewCandidate(address string) (string, error) {
@@ -38,7 +39,7 @@ func NewCandidate(address string) (string, error) {
 
 func LookupIP(address string) (string, error) {
 	if strings.HasPrefix(address, "stun:") {
-		ip, err := GetPublicIP()
+		ip, err := GetCachedPublicIP()
 		if err != nil {
 			return "", err
 		}
@@ -63,11 +64,20 @@ func LookupIP(address string) (string, error) {
 
 // GetPublicIP example from https://github.com/pion/stun
 func GetPublicIP() (net.IP, error) {
-	c, err := stun.Dial("udp", "stun.l.google.com:19302")
+	conn, err := net.Dial("udp", "stun.l.google.com:19302")
 	if err != nil {
 		return nil, err
 	}
 
+	c, err := stun.NewClient(conn)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = conn.SetDeadline(time.Now().Add(time.Second * 3)); err != nil {
+		return nil, err
+	}
+
 	var res stun.Event
 
 	message := stun.MustBuild(stun.TransactionID, stun.BindingRequest)
@@ -90,6 +100,24 @@ func GetPublicIP() (net.IP, error) {
 	return xorAddr.IP, nil
 }
 
+var cachedIP net.IP
+var cachedTS time.Time
+
+func GetCachedPublicIP() (net.IP, error) {
+	now := time.Now()
+	if now.After(cachedTS) {
+		newIP, err := GetPublicIP()
+		if err == nil {
+			cachedIP = newIP
+			cachedTS = now.Add(time.Minute * 5)
+		} else if cachedIP == nil {
+			return nil, err
+		}
+	}
+
+	return cachedIP, nil
+}
+
 func IsIP(host string) bool {
 	for _, i := range host {
 		if i >= 'A' {

www/index.html

@@ -70,6 +70,7 @@
         '<a href="api/stream.mp4?src={name}">mp4</a>',
         '<a href="api/frame.mp4?src={name}">frame</a>',
         `<a href="rtsp://${location.hostname}:8554/{name}">rtsp</a>`,
+        '<a href="api/stream.mjpeg?src={name}">mjpeg</a>',
         '<a href="api/streams?src={name}">info</a>',
     ];
 

www/webrtc.html

@@ -51,11 +51,6 @@
                 pc.addIceCandidate({candidate: msg.value, sdpMid: ''});
             } else if (msg.type === 'webrtc/answer') {
                 pc.setRemoteDescription({type: 'answer', sdp: msg.value});
-                pc.getTransceivers().forEach(t => {
-                    if (t.receiver.track.kind === 'audio') {
-                        t.currentDirection
-                    }
-                })
             }
         }