Update version to 0.1-rc.8

cmd/api/ws.go

@@ -6,6 +6,7 @@ import (
 	"net/url"
 	"strings"
 	"sync"
+	"time"
 )
 
 // Message - struct for data exchange in Web API
@@ -68,6 +69,8 @@ func apiWS(w http.ResponseWriter, r *http.Request) {
 
 	tr := &Transport{Request: r}
 	tr.OnWrite(func(msg interface{}) {
+		_ = ws.SetWriteDeadline(time.Now().Add(time.Second * 5))
+
 		if data, ok := msg.([]byte); ok {
 			_ = ws.WriteMessage(websocket.BinaryMessage, data)
 		} else {
@@ -101,7 +104,9 @@ type Transport struct {
 	Request  *http.Request
 	Consumer interface{} // TODO: rewrite
 
+	closed bool
 	mx     sync.Mutex
+	wrmx   sync.Mutex
 
 	onChange func()
 	onWrite  func(msg interface{})
@@ -118,21 +123,32 @@ func (t *Transport) OnWrite(f func(msg interface{})) {
 }
 
 func (t *Transport) Write(msg interface{}) {
-	t.mx.Lock()
+	t.wrmx.Lock()
 	t.onWrite(msg)
-	t.mx.Unlock()
+	t.wrmx.Unlock()
 }
 
 func (t *Transport) Close() {
+	t.mx.Lock()
 	for _, f := range t.onClose {
 		f()
 	}
+	t.closed = true
+	t.mx.Unlock()
 }
 
 func (t *Transport) OnChange(f func()) {
+	t.mx.Lock()
 	t.onChange = f
+	t.mx.Unlock()
 }
 
 func (t *Transport) OnClose(f func()) {
+	t.mx.Lock()
+	if t.closed {
+		f()
+	} else {
 		t.onClose = append(t.onClose, f)
+	}
+	t.mx.Unlock()
 }

cmd/app/app.go

@@ -10,7 +10,7 @@ import (
 	"runtime"
 )
 
-var Version = "0.1-rc.7"
+var Version = "0.1-rc.8"
 var UserAgent = "go2rtc/" + Version
 
 func Init() {

cmd/debug/stack.go

@@ -25,6 +25,7 @@ var stackSkip = [][]byte{
 
 	// webrtc/api.go
 	[]byte("created by github.com/pion/ice/v2.NewTCPMuxDefault"),
+	[]byte("created by github.com/pion/ice/v2.NewUDPMuxDefault"),
 }
 
 func stackHandler(w http.ResponseWriter, r *http.Request) {

cmd/exec/exec.go

@@ -34,8 +34,13 @@ func Init() {
 			return false
 		}
 
-		waiter <- conn
+		// unblocking write to channel
+		select {
+		case waiter <- conn:
 			return true
+		default:
+			return false
+		}
 	})
 
 	streams.HandleFunc("exec", Handle)
@@ -86,7 +91,13 @@ func Handle(url string) (streamer.Producer, error) {
 	chErr := make(chan error)
 
 	go func() {
-		chErr <- cmd.Wait()
+		err := cmd.Wait()
+		// unblocking write to channel
+		select {
+		case chErr <- err:
+		default:
+			log.Trace().Str("url", url).Msg("[exec] close")
+		}
 	}()
 
 	select {

cmd/mp4/ws.go

@@ -38,7 +38,7 @@ func handlerWSMSE(tr *api.Transport, msg *api.Message) error {
 	})
 
 	if err := stream.AddConsumer(cons); err != nil {
-		log.Warn().Err(err).Caller().Send()
+		log.Debug().Err(err).Msg("[mp4] add consumer")
 		return err
 	}
 

cmd/rtsp/rtsp.go

@@ -200,6 +200,9 @@ func tcpHandler(conn *rtsp.Conn) {
 
 	if err := conn.Accept(); err != nil {
 		log.Warn().Err(err).Caller().Send()
+		if closer != nil {
+			closer()
+		}
 		_ = conn.Close()
 		return
 	}
@@ -212,7 +215,7 @@ func tcpHandler(conn *rtsp.Conn) {
 
 	if closer != nil {
 		if err := conn.Handle(); err != nil {
-			log.Debug().Err(err).Caller().Send()
+			log.Debug().Msgf("[rtsp] handle=%s", err)
 		}
 
 		closer()

cmd/streams/producer.go

@@ -29,7 +29,7 @@ type Producer struct {
 
 	state    state
 	mu       sync.Mutex
-	restart *time.Timer
+	workerID int
 }
 
 func (p *Producer) SetSource(s string) {
@@ -104,20 +104,32 @@ func (p *Producer) start() {
 	log.Debug().Msgf("[streams] start producer url=%s", p.url)
 
 	p.state = stateStart
-	go func() {
+	p.workerID++
-		// safe read element while mu locked
+
-		if err := p.element.Start(); err != nil {
+	go p.worker(p.element, p.workerID)
-			log.Warn().Err(err).Str("url", p.url).Caller().Send()
-		}
-		p.reconnect()
-	}()
 }
 
-func (p *Producer) reconnect() {
+func (p *Producer) worker(element streamer.Producer, workerID int) {
+	if err := element.Start(); err != nil {
+		p.mu.Lock()
+		closed := p.workerID != workerID
+		p.mu.Unlock()
+
+		if closed {
+			return
+		}
+
+		log.Warn().Err(err).Str("url", p.url).Caller().Send()
+	}
+
+	p.reconnect(workerID)
+}
+
+func (p *Producer) reconnect(workerID int) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	if p.state != stateStart {
+	if p.workerID != workerID {
 		log.Trace().Msgf("[streams] stop reconnect url=%s", p.url)
 		return
 	}
@@ -126,9 +138,11 @@ func (p *Producer) reconnect() {
 
 	p.element, p.lastErr = GetProducer(p.url)
 	if p.lastErr != nil || p.element == nil {
-		log.Debug().Err(p.lastErr).Caller().Send()
+		log.Debug().Msgf("[streams] producer=%s", p.lastErr)
 		// TODO: dynamic timeout
-		p.restart = time.AfterFunc(30*time.Second, p.reconnect)
+		time.AfterFunc(30*time.Second, func() {
+			p.reconnect(workerID)
+		})
 		return
 	}
 
@@ -152,12 +166,7 @@ func (p *Producer) reconnect() {
 		}
 	}
 
-	go func() {
+	go p.worker(p.element, workerID)
-		if err := p.element.Start(); err != nil {
-			log.Debug().Err(err).Caller().Send()
-		}
-		p.reconnect()
-	}()
 }
 
 func (p *Producer) stop() {
@@ -171,6 +180,8 @@ func (p *Producer) stop() {
 	case stateNone:
 		log.Debug().Msgf("[streams] can't stop none producer")
 		return
+	case stateStart:
+		p.workerID++
 	}
 
 	log.Debug().Msgf("[streams] stop producer url=%s", p.url)
@@ -179,10 +190,6 @@ func (p *Producer) stop() {
 		_ = p.element.Stop()
 		p.element = nil
 	}
-	if p.restart != nil {
-		p.restart.Stop()
-		p.restart = nil
-	}
 
 	p.state = stateNone
 	p.tracks = nil

cmd/streams/stream.go

@@ -7,6 +7,7 @@ import (
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"strings"
 	"sync"
+	"sync/atomic"
 )
 
 type Consumer struct {
@@ -18,7 +19,7 @@ type Stream struct {
 	producers []*Producer
 	consumers []*Consumer
 	mu        sync.Mutex
-	wg        sync.WaitGroup
+	requests  int32
 }
 
 func NewStream(source interface{}) *Stream {
@@ -53,6 +54,9 @@ func (s *Stream) SetSource(source string) {
 }
 
 func (s *Stream) AddConsumer(cons streamer.Consumer) (err error) {
+	// support for multiple simultaneous requests from different consumers
+	atomic.AddInt32(&s.requests, 1)
+
 	ic := len(s.consumers)
 
 	consumer := &Consumer{element: cons}
@@ -60,9 +64,6 @@ func (s *Stream) AddConsumer(cons streamer.Consumer) (err error) {
 
 	var codecs string
 
-	// support for multiple simultaneous requests from different consumers
-	s.wg.Add(1)
-
 	// Step 1. Get consumer medias
 	for icc, consMedia := range cons.GetMedias() {
 		log.Trace().Stringer("media", consMedia).
@@ -86,7 +87,7 @@ func (s *Stream) AddConsumer(cons streamer.Consumer) (err error) {
 					// Step 4. Get producer track
 					prodTrack := prod.GetTrack(prodMedia, prodCodec)
 					if prodTrack == nil {
-						log.Warn().Str("url", prod.url).Msg("[stream] can't get track")
+						log.Warn().Str("url", prod.url).Msg("[streams] can't get track")
 						continue
 					}
 
@@ -101,12 +102,11 @@ func (s *Stream) AddConsumer(cons streamer.Consumer) (err error) {
 		}
 	}
 
-	s.wg.Done()
+	if atomic.AddInt32(&s.requests, -1) == 0 {
-	s.wg.Wait()
+		s.stopProducers()
+	}
 
 	if len(producers) == 0 {
-		s.stopProducers()
-
 		if len(codecs) > 0 {
 			return errors.New("codecs not match: " + codecs)
 		}
@@ -197,8 +197,12 @@ producers:
 //}
 
 func (s *Stream) MarshalJSON() ([]byte, error) {
+	if !s.mu.TryLock() {
+		log.Warn().Msgf("[streams] json locked")
+		return []byte(`null`), nil
+	}
+
 	var v []interface{}
-	s.mu.Lock()
 	for _, prod := range s.producers {
 		if prod.element != nil {
 			v = append(v, prod.element)
@@ -242,6 +246,10 @@ func (s *Stream) removeProducer(i int) {
 }
 
 func collectCodecs(media *streamer.Media, codecs *string) {
+	if media.Direction == streamer.DirectionRecvonly {
+		return
+	}
+
 	for _, codec := range media.Codecs {
 		name := codec.Name
 		if name == streamer.CodecAAC {

cmd/webrtc/candidates.go

@@ -7,6 +7,7 @@ import (
 )
 
 var candidates []string
+var networks = []string{"udp", "tcp"}
 
 func AddCandidate(address string) {
 	candidates = append(candidates, address)
@@ -20,7 +21,8 @@ func asyncCandidates(tr *api.Transport) {
 			continue
 		}
 
-		cand, err := webrtc.NewCandidate(address)
+		for _, network := range networks {
+			cand, err := webrtc.NewCandidate(network, address)
 			if err != nil {
 				log.Warn().Err(err).Caller().Send()
 				continue
@@ -30,6 +32,7 @@ func asyncCandidates(tr *api.Transport) {
 
 			tr.Write(&api.Message{Type: "webrtc/candidate", Value: cand})
 		}
+	}
 }
 
 func syncCanditates(answer string) (string, error) {
@@ -57,7 +60,8 @@ func syncCanditates(answer string) (string, error) {
 			continue
 		}
 
-		cand, err := webrtc.NewCandidate(address)
+		for _, network := range networks {
+			cand, err := webrtc.NewCandidate(network, address)
 			if err != nil {
 				log.Warn().Err(err).Msg("[webrtc] candidate")
 				continue
@@ -65,6 +69,7 @@ func syncCanditates(answer string) (string, error) {
 
 			md.WithPropertyAttribute(cand)
 		}
+	}
 
 	if end {
 		md.WithPropertyAttribute("end-of-candidates")

cmd/webrtc/webrtc.go

@@ -22,6 +22,7 @@ func Init() {
 		} `yaml:"webrtc"`
 	}
 
+	cfg.Mod.Listen = ":8555"
 	cfg.Mod.IceServers = []pion.ICEServer{
 		{URLs: []string{"stun:stun.l.google.com:19302"}},
 	}
@@ -112,7 +113,7 @@ func asyncHandler(tr *api.Transport, msg *api.Message) error {
 
 	// 2. AddConsumer, so we get new tracks
 	if err = stream.AddConsumer(conn); err != nil {
-		log.Warn().Err(err).Caller().Send()
+		log.Debug().Err(err).Msg("[webrtc] add consumer")
 		_ = conn.Conn.Close()
 		return err
 	}

pkg/ivideon/client.go

@@ -14,9 +14,18 @@ import (
 	"io"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 )
 
+type State byte
+
+const (
+	StateNone State = iota
+	StateConn
+	StateHandle
+)
+
 type Client struct {
 	streamer.Element
 
@@ -26,12 +35,12 @@ type Client struct {
 	medias []*streamer.Media
 	tracks map[byte]*streamer.Track
 
-	closed bool
-
 	msg *message
 	t0  time.Time
 
 	buffer chan []byte
+	state  State
+	mu     sync.Mutex
 }
 
 func NewClient(id string) *Client {
@@ -69,16 +78,26 @@ func (c *Client) Dial() (err error) {
 		return err
 	}
 
+	c.state = StateConn
+
 	return nil
 }
 
 func (c *Client) Handle() error {
-	c.buffer = make(chan []byte, 5)
 	// add delay to the stream for smooth playing (not a best solution)
 	c.t0 = time.Now().Add(time.Second)
 
+	c.mu.Lock()
+
+	if c.state == StateConn {
+		c.buffer = make(chan []byte, 5)
+		c.state = StateHandle
+
 		// processing stream in separate thread for lower delay between packets
-	go c.worker()
+		go c.worker(c.buffer)
+	}
+
+	c.mu.Unlock()
 
 	_, data, err := c.conn.ReadMessage()
 	if err != nil {
@@ -87,8 +106,12 @@ func (c *Client) Handle() error {
 
 	track := c.tracks[c.msg.Track]
 	if track != nil {
+		c.mu.Lock()
+		if c.state == StateHandle {
 			c.buffer <- data
 		}
+		c.mu.Unlock()
+	}
 
 	// we have one unprocessed msg after getTracks
 	for {
@@ -114,8 +137,12 @@ func (c *Client) Handle() error {
 
 			track = c.tracks[msg.Track]
 			if track != nil {
+				c.mu.Lock()
+				if c.state == StateHandle {
 					c.buffer <- data
 				}
+				c.mu.Unlock()
+			}
 
 		default:
 			return fmt.Errorf("wrong message type: %s", data)
@@ -124,13 +151,19 @@ func (c *Client) Handle() error {
 }
 
 func (c *Client) Close() error {
-	if c.conn == nil {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	switch c.state {
+	case StateNone:
 		return nil
-	}
+	case StateConn:
-	if c.buffer != nil {
+	case StateHandle:
 		close(c.buffer)
 	}
-	c.closed = true
+
+	c.state = StateNone
+
 	return c.conn.Close()
 }
 
@@ -210,13 +243,13 @@ func (c *Client) getTracks() error {
 	}
 }
 
-func (c *Client) worker() {
+func (c *Client) worker(buffer chan []byte) {
 	var track *streamer.Track
 	for _, track = range c.tracks {
 		break
 	}
 
-	for data := range c.buffer {
+	for data := range buffer {
 		moof := &fmp4io.MovieFrag{}
 		if _, err := moof.Unmarshal(data, 0); err != nil {
 			continue

pkg/ivideon/streamer.go

@@ -20,7 +20,7 @@ func (c *Client) GetTrack(media *streamer.Media, codec *streamer.Codec) *streame
 
 func (c *Client) Start() error {
 	err := c.Handle()
-	if c.closed {
+	if c.buffer == nil {
 		return nil
 	}
 	return err

pkg/mjpeg/client.go

@@ -64,6 +64,8 @@ func (c *Client) Start() error {
 }
 
 func (c *Client) Stop() error {
+	// important for close reader/writer gorutines
+	_ = c.res.Body.Close()
 	c.closed = true
 	return nil
 }

pkg/mp4f/consumer.go

@@ -89,7 +89,7 @@ func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *strea
 			return nil
 		}
 
-		if !codec.IsRAW() {
+		if codec.IsRTP() {
 			wrapper := h264.RTPDepay(track)
 			push = wrapper(push)
 		}

pkg/rtsp/conn.go

@@ -48,6 +48,22 @@ const (
 
 type State byte
 
+func (s State) String() string {
+	switch s {
+	case StateNone:
+		return "NONE"
+	case StateConn:
+		return "CONN"
+	case StateSetup:
+		return "SETUP"
+	case StatePlay:
+		return "PLAY"
+	case StateHandle:
+		return "HANDLE"
+	}
+	return strconv.Itoa(int(s))
+}
+
 const (
 	StateNone State = iota
 	StateConn
@@ -346,6 +362,10 @@ func (c *Conn) SetupMedia(
 	c.stateMu.Lock()
 	defer c.stateMu.Unlock()
 
+	if c.state != StateConn && c.state != StateSetup {
+		return nil, fmt.Errorf("RTSP SETUP from wrong state: %s", c.state)
+	}
+
 	ch := c.GetChannel(media)
 	if ch < 0 {
 		return nil, fmt.Errorf("wrong media: %v", media)
@@ -648,15 +668,17 @@ func (c *Conn) Handle() (err error) {
 	case StatePlay:
 		c.state = StateHandle
 	default:
-		err = fmt.Errorf("RTSP Handle from wrong state: %d", c.state)
+		err = fmt.Errorf("RTSP HANDLE from wrong state: %s", c.state)
 
 		c.state = StateNone
 		_ = c.conn.Close()
 	}
 
+	ok := c.state == StateHandle
+
 	c.stateMu.Unlock()
 
-	if c.state != StateHandle {
+	if !ok {
 		return
 	}
 

pkg/rtsp/streamer.go

@@ -119,7 +119,7 @@ func (c *Conn) MarshalJSON() ([]byte, error) {
 		v[k] = media.String()
 	}
 	for i, track := range c.tracks {
-		k := "track:" + strconv.Itoa(int(i>>1))
+		k := "track:" + strconv.Itoa(i)
 		v[k] = track.String()
 	}
 	//for i, track := range c.tracks {

pkg/webrtc/api.go

@@ -35,13 +35,17 @@ func NewAPI(address string) (*webrtc.API, error) {
 	s.SetICEMulticastDNSMode(ice.MulticastDNSModeDisabled)
 
 	if address != "" {
-		ln, err := net.Listen("tcp", address)
-		if err == nil {
 		s.SetNetworkTypes([]webrtc.NetworkType{
 			webrtc.NetworkTypeUDP4, webrtc.NetworkTypeUDP6,
 			webrtc.NetworkTypeTCP4, webrtc.NetworkTypeTCP6,
 		})
 
+		if ln, err := net.ListenPacket("udp", address); err == nil {
+			udpMux := webrtc.NewICEUDPMux(nil, ln)
+			s.SetICEUDPMux(udpMux)
+		}
+
+		if ln, err := net.Listen("tcp", address); err == nil {
 			tcpMux := webrtc.NewICETCPMux(nil, ln, 8)
 			s.SetICETCPMux(tcpMux)
 		}

pkg/webrtc/helper.go

@@ -13,7 +13,7 @@ import (
 	"time"
 )
 
-func NewCandidate(address string) (string, error) {
+func NewCandidate(network, address string) (string, error) {
 	i := strings.LastIndexByte(address, ':')
 	if i < 0 {
 		return "", errors.New("wrong candidate: " + address)
@@ -26,7 +26,7 @@ func NewCandidate(address string) (string, error) {
 	}
 
 	cand, err := ice.NewCandidateHost(&ice.CandidateHostConfig{
-		Network:   "tcp",
+		Network:   network,
 		Address:   host,
 		Port:      i,
 		Component: ice.ComponentRTP,