Update version to 0.1-rc.8

Speedup container building using Golang cross-building

Dockerfile

@@ -1,24 +1,31 @@
+# syntax=docker/dockerfile:labs
+
 # 0. Prepare images
 ARG PYTHON_VERSION="3.11"
 ARG GO_VERSION="1.19"
 ARG NGROK_VERSION="3"
 
 FROM python:${PYTHON_VERSION}-alpine AS base
-FROM golang:${GO_VERSION}-alpine AS go
 FROM ngrok/ngrok:${NGROK_VERSION}-alpine AS ngrok
 
 
 # 1. Build go2rtc binary
-FROM go AS build
+FROM --platform=$BUILDPLATFORM golang:${GO_VERSION}-alpine AS build
+ARG TARGETPLATFORM
+ARG TARGETOS
+ARG TARGETARCH
+
+ENV GOOS=${TARGETOS}
+ENV GOARCH=${TARGETARCH}
 
 WORKDIR /build
 
 # Cache dependencies
 COPY go.mod go.sum ./
-RUN go mod download
+RUN --mount=type=cache,target=/root/.cache/go-build go mod download
 
 COPY . .
-RUN CGO_ENABLED=0 go build -ldflags "-s -w" -trimpath
+RUN --mount=type=cache,target=/root/.cache/go-build CGO_ENABLED=0 go build -ldflags "-s -w" -trimpath
 
 
 # 2. Collect all files

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
 
-	mx sync.Mutex
+	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.onClose = append(t.onClose, f)
+	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.6"
+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
-		return true
+		// 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

@@ -27,9 +27,9 @@ type Producer struct {
 	lastErr error
 	tracks  []*streamer.Track
 
-	state   state
-	mu      sync.Mutex
-	restart *time.Timer
+	state    state
+	mu       sync.Mutex
+	workerID int
 }
 
 func (p *Producer) SetSource(s string) {
@@ -48,13 +48,18 @@ func (p *Producer) GetMedias() []*streamer.Media {
 
 		p.element, p.lastErr = GetProducer(p.url)
 		if p.lastErr != nil || p.element == nil {
-			log.Error().Err(p.lastErr).Caller().Send()
+			log.Error().Err(p.lastErr).Str("url", p.url).Caller().Send()
 			return nil
 		}
 
 		p.state = stateMedias
 	}
 
+	// if element in reconnect state
+	if p.element == nil {
+		return nil
+	}
+
 	return p.element.GetMedias()
 }
 
@@ -99,32 +104,45 @@ func (p *Producer) start() {
 	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).Caller().Send()
-		}
-		p.reconnect()
-	}()
+	p.workerID++
+
+	go p.worker(p.element, p.workerID)
 }
 
-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
 	}
 
 	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()
+	p.element, p.lastErr = GetProducer(p.url)
+	if p.lastErr != nil || p.element == nil {
+		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
 	}
 
@@ -148,12 +166,7 @@ func (p *Producer) reconnect() {
 		}
 	}
 
-	go func() {
-		if err = p.element.Start(); err != nil {
-			log.Debug().Err(err).Caller().Send()
-		}
-		p.reconnect()
-	}()
+	go p.worker(p.element, workerID)
 }
 
 func (p *Producer) stop() {
@@ -167,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)
@@ -175,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,6 +19,7 @@ type Stream struct {
 	producers []*Producer
 	consumers []*Consumer
 	mu        sync.Mutex
+	requests  int32
 }
 
 func NewStream(source interface{}) *Stream {
@@ -52,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}
@@ -82,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().Msg("[stream] can't get track")
+						log.Warn().Str("url", prod.url).Msg("[streams] can't get track")
 						continue
 					}
 
@@ -97,9 +102,11 @@ func (s *Stream) AddConsumer(cons streamer.Consumer) (err error) {
 		}
 	}
 
-	if len(producers) == 0 {
+	if atomic.AddInt32(&s.requests, -1) == 0 {
 		s.stopProducers()
+	}
 
+	if len(producers) == 0 {
 		if len(codecs) > 0 {
 			return errors.New("codecs not match: " + codecs)
 		}
@@ -190,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)
@@ -235,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,15 +21,17 @@ func asyncCandidates(tr *api.Transport) {
 			continue
 		}
 
-		cand, err := webrtc.NewCandidate(address)
-		if err != nil {
-			log.Warn().Err(err).Caller().Send()
-			continue
+		for _, network := range networks {
+			cand, err := webrtc.NewCandidate(network, address)
+			if err != nil {
+				log.Warn().Err(err).Caller().Send()
+				continue
+			}
+
+			log.Trace().Str("candidate", cand).Msg("[webrtc] config")
+
+			tr.Write(&api.Message{Type: "webrtc/candidate", Value: cand})
 		}
-
-		log.Trace().Str("candidate", cand).Msg("[webrtc] config")
-
-		tr.Write(&api.Message{Type: "webrtc/candidate", Value: cand})
 	}
 }
 
@@ -57,13 +60,15 @@ func syncCanditates(answer string) (string, error) {
 			continue
 		}
 
-		cand, err := webrtc.NewCandidate(address)
-		if err != nil {
-			log.Warn().Err(err).Msg("[webrtc] candidate")
-			continue
-		}
+		for _, network := range networks {
+			cand, err := webrtc.NewCandidate(network, address)
+			if err != nil {
+				log.Warn().Err(err).Msg("[webrtc] candidate")
+				continue
+			}
 
-		md.WithPropertyAttribute(cand)
+			md.WithPropertyAttribute(cand)
+		}
 	}
 
 	if end {

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/h264/helper.go

@@ -3,6 +3,7 @@ package h264
 import (
 	"encoding/base64"
 	"encoding/binary"
+	"fmt"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"strings"
 )
@@ -51,6 +52,16 @@ func GetProfileLevelID(fmtp string) string {
 	if fmtp == "" {
 		return ""
 	}
+
+	// some cameras has wrong profile-level-id
+	// https://github.com/AlexxIT/go2rtc/issues/155
+	if s := streamer.Between(fmtp, "sprop-parameter-sets=", ","); s != "" {
+		sps, _ := base64.StdEncoding.DecodeString(s)
+		if len(sps) >= 4 {
+			return fmt.Sprintf("%06X", sps[1:4])
+		}
+	}
+
 	return streamer.Between(fmtp, "profile-level-id=", ";")
 }
 

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)
 
-	// processing stream in separate thread for lower delay between packets
-	go c.worker()
+	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(c.buffer)
+	}
+
+	c.mu.Unlock()
 
 	_, data, err := c.conn.ReadMessage()
 	if err != nil {
@@ -87,7 +106,11 @@ func (c *Client) Handle() error {
 
 	track := c.tracks[c.msg.Track]
 	if track != nil {
-		c.buffer <- data
+		c.mu.Lock()
+		if c.state == StateHandle {
+			c.buffer <- data
+		}
+		c.mu.Unlock()
 	}
 
 	// we have one unprocessed msg after getTracks
@@ -114,7 +137,11 @@ func (c *Client) Handle() error {
 
 			track = c.tracks[msg.Track]
 			if track != nil {
-				c.buffer <- data
+				c.mu.Lock()
+				if c.state == StateHandle {
+					c.buffer <- data
+				}
+				c.mu.Unlock()
 			}
 
 		default:
@@ -124,11 +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:
+	case StateHandle:
+		close(c.buffer)
 	}
-	close(c.buffer)
-	c.closed = true
+
+	c.state = StateNone
+
 	return c.conn.Close()
 }
 
@@ -208,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

@@ -20,6 +20,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -47,11 +48,28 @@ 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
 	StateSetup
 	StatePlay
+	StateHandle
 )
 
 type Conn struct {
@@ -72,6 +90,7 @@ type Conn struct {
 	conn     net.Conn
 	mode     Mode
 	state    State
+	stateMu  sync.Mutex
 	reader   *bufio.Reader
 	sequence int
 	uri      string
@@ -340,6 +359,13 @@ func (c *Conn) Setup() error {
 func (c *Conn) SetupMedia(
 	media *streamer.Media, codec *streamer.Codec,
 ) (*streamer.Track, error) {
+	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)
@@ -461,12 +487,19 @@ func (c *Conn) SetupMedia(
 }
 
 func (c *Conn) Play() (err error) {
+	c.stateMu.Lock()
+	defer c.stateMu.Unlock()
+
 	if c.state != StateSetup {
 		return fmt.Errorf("RTSP PLAY from wrong state: %s", c.state)
 	}
 
 	req := &tcp.Request{Method: MethodPlay, URL: c.URL}
-	return c.Request(req)
+	if err = c.Request(req); err == nil {
+		c.state = StatePlay
+	}
+
+	return
 }
 
 func (c *Conn) Teardown() (err error) {
@@ -476,12 +509,14 @@ func (c *Conn) Teardown() (err error) {
 }
 
 func (c *Conn) Close() error {
+	c.stateMu.Lock()
+	defer c.stateMu.Unlock()
+
 	if c.state == StateNone {
 		return nil
 	}
-	if err := c.Teardown(); err != nil {
-		return err
-	}
+
+	_ = c.Teardown()
 	c.state = StateNone
 	return c.conn.Close()
 }
@@ -614,7 +649,10 @@ func (c *Conn) Accept() error {
 
 		case MethodRecord, MethodPlay:
 			res := &tcp.Response{Request: req}
-			return c.Response(res)
+			if err = c.Response(res); err == nil {
+				c.state = StatePlay
+			}
+			return err
 
 		default:
 			return fmt.Errorf("unsupported method: %s", req.Method)
@@ -623,13 +661,31 @@ func (c *Conn) Accept() error {
 }
 
 func (c *Conn) Handle() (err error) {
-	if c.state != StateSetup {
-		return fmt.Errorf("RTSP Handle from wrong state: %d", c.state)
+	c.stateMu.Lock()
+
+	switch c.state {
+	case StateNone: // Close after PLAY and before Handle is OK (because SETUP after PLAY)
+	case StatePlay:
+		c.state = StateHandle
+	default:
+		err = fmt.Errorf("RTSP HANDLE from wrong state: %s", c.state)
+
+		c.state = StateNone
+		_ = c.conn.Close()
 	}
 
-	c.state = StatePlay
+	ok := c.state == StateHandle
+
+	c.stateMu.Unlock()
+
+	if !ok {
+		return
+	}
 
 	defer func() {
+		c.stateMu.Lock()
+		defer c.stateMu.Unlock()
+
 		if c.state == StateNone {
 			err = nil
 			return

pkg/rtsp/streamer.go

@@ -21,7 +21,8 @@ func (c *Conn) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.
 	}
 
 	// can't setup new tracks from play state - forcing a reconnection feature
-	if c.state == StatePlay {
+	switch c.state {
+	case StatePlay, StateHandle:
 		go c.Close()
 		return streamer.NewTrack(codec, media.Direction)
 	}
@@ -118,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,
-			})
+		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

@@ -1,6 +1,7 @@
 package webrtc
 
 import (
+	"errors"
 	"fmt"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/pion/ice/v2"
@@ -12,11 +13,12 @@ import (
 	"time"
 )
 
-func NewCandidate(address string) (string, error) {
-	host, port, err := net.SplitHostPort(address)
-	if err != nil {
-		return "", err
+func NewCandidate(network, address string) (string, error) {
+	i := strings.LastIndexByte(address, ':')
+	if i < 0 {
+		return "", errors.New("wrong candidate: " + address)
 	}
+	host, port := address[:i], address[i+1:]
 
 	i, err := strconv.Atoi(port)
 	if err != nil {
@@ -24,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,

www/video-stream.js

@@ -1,19 +1,23 @@
 import {VideoRTC} from "./video-rtc.js";
 
 class VideoStream extends VideoRTC {
-    constructor() {
-        super();
+    set divMode(value) {
+        this.querySelector(".mode").innerText = value;
+        this.querySelector(".status").innerText = "";
+    }
 
-        /** @type {HTMLDivElement} */
-        this.divMode = null;
-        /** @type {HTMLDivElement} */
-        this.divStatus = null;
+    set divError(value) {
+        const state = this.querySelector(".mode").innerText;
+        if (state !== "loading") return;
+        this.querySelector(".mode").innerText = "error";
+        this.querySelector(".status").innerText = value;
     }
 
     /**
      * Custom GUI
      */
     oninit() {
+        console.debug("stream.oninit");
         super.oninit();
 
         this.innerHTML = `
@@ -36,35 +40,36 @@ class VideoStream extends VideoRTC {
         </div>
         `;
 
-        this.divStatus = this.querySelector(".status");
-        this.divMode = this.querySelector(".mode");
-
         const info = this.querySelector(".info")
         this.insertBefore(this.video, info);
     }
 
     onconnect() {
+        console.debug("stream.onconnect");
         const result = super.onconnect();
-        if (result) {
-            this.divMode.innerText = "loading";
-        }
+        if (result) this.divMode = "loading";
         return result;
     }
 
+    ondisconnect() {
+        console.debug("stream.ondisconnect");
+        super.ondisconnect();
+    }
+
     onopen() {
+        console.debug("stream.onopen");
         const result = super.onopen();
 
         this.onmessage["stream"] = msg => {
+            console.debug("stream.onmessge", msg);
             switch (msg.type) {
                 case "error":
-                    this.divMode.innerText = "error";
-                    this.divStatus.innerText = msg.value;
+                    this.divError = msg.value;
                     break;
                 case "mse":
                 case "mp4":
                 case "mjpeg":
-                    this.divMode.innerText = msg.type.toUpperCase();
-                    this.divStatus.innerText = "";
+                    this.divMode = msg.type.toUpperCase();
                     break;
             }
         }
@@ -72,12 +77,17 @@ class VideoStream extends VideoRTC {
         return result;
     }
 
+    onclose() {
+        console.debug("stream.onclose");
+        return super.onclose();
+    }
+
     onpcvideo(ev) {
+        console.debug("stream.onpcvideo");
         super.onpcvideo(ev);
 
         if (this.pcState !== WebSocket.CLOSED) {
-            this.divMode.innerText = "RTC";
-            this.divStatus.innerText = "";
+            this.divMode = "RTC";
         }
     }
 }