Merge pull request #1758 from seydx/rtsp-udp

Add RTSP UDP transport support

pkg/rtsp/client.go

@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/AlexxIT/go2rtc/pkg/tcp/websocket"
@@ -36,14 +37,22 @@ func (c *Conn) Dial() (err error) {
 
 	var conn net.Conn
 
-	if c.Transport == "" {
+	switch c.Transport {
-		timeout := core.ConnDialTimeout
+	case "", "tcp", "udp":
+		var timeout time.Duration
 		if c.Timeout != 0 {
 			timeout = time.Second * time.Duration(c.Timeout)
+		} else {
+			timeout = core.ConnDialTimeout
 		}
 		conn, err = tcp.Dial(c.URL, timeout)
-		c.Protocol = "rtsp+tcp"
+
-	} else {
+		if c.Transport != "udp" {
+			c.Protocol = "rtsp+tcp"
+		} else {
+			c.Protocol = "rtsp+udp"
+		}
+	default:
 		conn, err = websocket.Dial(c.Transport)
 		c.Protocol = "ws"
 	}
@@ -61,6 +70,9 @@ func (c *Conn) Dial() (err error) {
 	c.sequence = 0
 	c.state = StateConn
 
+	c.udpConn = nil
+	c.udpAddr = nil
+
 	c.Connection.RemoteAddr = conn.RemoteAddr().String()
 	c.Connection.Transport = conn
 	c.Connection.URL = c.uri
@@ -218,15 +230,27 @@ func (c *Conn) Record() (err error) {
 func (c *Conn) SetupMedia(media *core.Media) (byte, error) {
 	var transport string
 
-	// try to use media position as channel number
+	if c.Transport == "udp" {
-	for i, m := range c.Medias {
+		conn1, conn2, err := ListenUDPPair()
-		if m.Equal(media) {
+		if err != nil {
-			transport = fmt.Sprintf(
+			return 0, err
-				// i   - RTP (data channel)
+		}
-				// i+1 - RTCP (control channel)
+
-				"RTP/AVP/TCP;unicast;interleaved=%d-%d", i*2, i*2+1,
+		c.udpConn = append(c.udpConn, conn1, conn2)
-			)
+
-			break
+		port := conn1.LocalAddr().(*net.UDPAddr).Port
+		transport = fmt.Sprintf("RTP/AVP;unicast;client_port=%d-%d", port, port+1)
+	} else {
+		// try to use media position as channel number
+		for i, m := range c.Medias {
+			if m.Equal(media) {
+				transport = fmt.Sprintf(
+					// i   - RTP (data channel)
+					// i+1 - RTCP (control channel)
+					"RTP/AVP/TCP;unicast;interleaved=%d-%d", i*2, i*2+1,
+				)
+				break
+			}
 		}
 	}
 
@@ -286,27 +310,53 @@ func (c *Conn) SetupMedia(media *core.Media) (byte, error) {
 		}
 	}
 
-	// we send our `interleaved`, but camera can answer with another
+	// Parse server response
-
-	// Transport: RTP/AVP/TCP;unicast;interleaved=10-11;ssrc=10117CB7
-	// 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
 	transport = res.Header.Get("Transport")
-	if !strings.HasPrefix(transport, "RTP/AVP/TCP;") {
+
+	if c.Transport == "udp" {
+		channel := byte(len(c.udpConn) - 2)
+
+		// Dahua:   RTP/AVP/UDP;unicast;client_port=49292-49293;server_port=43670-43671;ssrc=7CB694B4
+		// OpenIPC: RTP/AVP/UDP;unicast;client_port=59612-59613
+		if s := core.Between(transport, "server_port=", ";"); s != "" {
+			s1, s2, _ := strings.Cut(s, "-")
+			port1 := core.Atoi(s1)
+			port2 := core.Atoi(s2)
+			// TODO: more smart handling empty server ports
+			if port1 > 0 && port2 > 0 {
+				remoteIP := c.conn.RemoteAddr().(*net.TCPAddr).IP
+				c.udpAddr = append(c.udpAddr,
+					&net.UDPAddr{IP: remoteIP, Port: port1},
+					&net.UDPAddr{IP: remoteIP, Port: port2},
+				)
+
+				go func() {
+					// Try to open a hole in the NAT router (to allow incoming UDP packets)
+					// by send a UDP packet for RTP and RTCP to the remote RTSP server.
+					// https://github.com/FFmpeg/FFmpeg/blob/aa91ae25b88e195e6af4248e0ab30605735ca1cd/libavformat/rtpdec.c#L416-L438
+					_, _ = c.WriteToUDP([]byte{0x80, 0x00, 0x00, 0x00}, channel)
+					_, _ = c.WriteToUDP([]byte{0x80, 0xC8, 0x00, 0x01}, channel+1)
+				}()
+			}
+		}
+
+		return channel, nil
+	} else {
+		// 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.111;source=192.168.1.222;interleaved=0
+		// Transport: RTP/AVP/TCP;ssrc=22345682;interleaved=0-1
 		// 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(transport, ";interleaved=") {
+		s := core.Between(transport, "interleaved=", "-")
+		i, err := strconv.Atoi(s)
+		if err != nil {
 			return 0, fmt.Errorf("wrong transport: %s", transport)
 		}
-	}
 
-	channel := core.Between(transport, "interleaved=", "-")
+		return byte(i), nil
-	i, err := strconv.Atoi(channel)
-	if err != nil {
-		return 0, err
 	}
-
-	return byte(i), nil
 }
 
 func (c *Conn) Play() (err error) {
@@ -327,5 +377,56 @@ func (c *Conn) Close() error {
 	if c.OnClose != nil {
 		_ = c.OnClose()
 	}
+	for _, conn := range c.udpConn {
+		_ = conn.Close()
+	}
 	return c.conn.Close()
 }
+
+func (c *Conn) WriteToUDP(b []byte, channel byte) (int, error) {
+	return c.udpConn[channel].WriteToUDP(b, c.udpAddr[channel])
+}
+
+const listenUDPAttemps = 10
+
+var listenUDPMu sync.Mutex
+
+func ListenUDPPair() (*net.UDPConn, *net.UDPConn, error) {
+	listenUDPMu.Lock()
+	defer listenUDPMu.Unlock()
+
+	for i := 0; i < listenUDPAttemps; i++ {
+		// Get a random even port from the OS
+		ln1, err := net.ListenUDP("udp", &net.UDPAddr{IP: nil, Port: 0})
+		if err != nil {
+			continue
+		}
+
+		var port1 = ln1.LocalAddr().(*net.UDPAddr).Port
+		var port2 int
+
+		// 11. RTP over Network and Transport Protocols (https://www.ietf.org/rfc/rfc3550.txt)
+		// For UDP and similar protocols,
+		// RTP SHOULD use an even destination port number and the corresponding
+		// RTCP stream SHOULD use the next higher (odd) destination port number
+		if port1&1 > 0 {
+			port2 = port1 - 1
+		} else {
+			port2 = port1 + 1
+		}
+
+		ln2, err := net.ListenUDP("udp", &net.UDPAddr{IP: nil, Port: port2})
+		if err != nil {
+			_ = ln1.Close()
+			continue
+		}
+
+		if port1 < port2 {
+			return ln1, ln2, nil
+		} else {
+			return ln2, ln1, nil
+		}
+	}
+
+	return nil, nil, fmt.Errorf("can't open two UDP ports")
+}

pkg/rtsp/conn.go

@@ -2,6 +2,7 @@ package rtsp
 
 import (
 	"bufio"
+	"context"
 	"encoding/binary"
 	"fmt"
 	"io"
@@ -13,7 +14,6 @@ import (
 
 	"github.com/AlexxIT/go2rtc/pkg/core"
 	"github.com/AlexxIT/go2rtc/pkg/tcp"
-	"github.com/pion/rtcp"
 	"github.com/pion/rtp"
 )
 
@@ -40,6 +40,7 @@ type Conn struct {
 	keepalive int
 	mode      core.Mode
 	playOK    bool
+	playErr   error
 	reader    *bufio.Reader
 	sequence  int
 	session   string
@@ -47,6 +48,9 @@ type Conn struct {
 
 	state   State
 	stateMu sync.Mutex
+
+	udpConn []*net.UDPConn
+	udpAddr []*net.UDPAddr
 }
 
 const (
@@ -68,7 +72,6 @@ func (s State) String() string {
 	case StateNone:
 		return "NONE"
 	case StateConn:
-
 		return "CONN"
 	case StateSetup:
 		return MethodSetup
@@ -88,23 +91,25 @@ const (
 func (c *Conn) Handle() (err error) {
 	var timeout time.Duration
 
-	var keepaliveDT time.Duration
-	var keepaliveTS time.Time
-
 	switch c.mode {
 	case core.ModeActiveProducer:
+		var keepaliveDT time.Duration
+
 		if c.keepalive > 5 {
 			keepaliveDT = time.Duration(c.keepalive-5) * time.Second
 		} else {
 			keepaliveDT = 25 * time.Second
 		}
-		keepaliveTS = time.Now().Add(keepaliveDT)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		go c.handleKeepalive(ctx, keepaliveDT)
+		defer cancel()
 
 		if c.Timeout == 0 {
 			// polling frames from remote RTSP Server (ex Camera)
 			timeout = time.Second * 5
 
-			if len(c.Receivers) == 0 {
+			if len(c.Receivers) == 0 || c.Transport == "udp" {
 				// if we only send audio to camera
 				// https://github.com/AlexxIT/go2rtc/issues/659
 				timeout += keepaliveDT
@@ -129,148 +134,190 @@ func (c *Conn) Handle() (err error) {
 		return fmt.Errorf("wrong RTSP conn mode: %d", c.mode)
 	}
 
+	for i := 0; i < len(c.udpConn); i++ {
+		go c.handleUDPData(byte(i))
+	}
+
 	for c.state != StateNone {
 		ts := time.Now()
 
-		if err = c.conn.SetReadDeadline(ts.Add(timeout)); err != nil {
+		_ = c.conn.SetReadDeadline(ts.Add(timeout))
+
+		if err = c.handleTCPData(); err != nil {
 			return
 		}
-
-		// we can read:
-		// 1. RTP interleaved: `$` + 1B channel number + 2B size
-		// 2. RTSP response:   RTSP/1.0 200 OK
-		// 3. RTSP request:    OPTIONS ...
-		var buf4 []byte // `$` + 1B channel number + 2B size
-		buf4, err = c.reader.Peek(4)
-		if err != nil {
-			return
-		}
-
-		var channelID byte
-		var size uint16
-
-		if buf4[0] != '$' {
-			switch string(buf4) {
-			case "RTSP":
-				var res *tcp.Response
-				if res, err = c.ReadResponse(); err != nil {
-					return
-				}
-				c.Fire(res)
-				// for playing backchannel only after OK response on play
-				c.playOK = true
-				continue
-
-			case "OPTI", "TEAR", "DESC", "SETU", "PLAY", "PAUS", "RECO", "ANNO", "GET_", "SET_":
-				var req *tcp.Request
-				if req, err = c.ReadRequest(); err != nil {
-					return
-				}
-				c.Fire(req)
-				if req.Method == MethodOptions {
-					res := &tcp.Response{Request: req}
-					if err = c.WriteResponse(res); err != nil {
-						return
-					}
-				}
-				continue
-
-			default:
-				c.Fire("RTSP wrong input")
-
-				for i := 0; ; i++ {
-					// search next start symbol
-					if _, err = c.reader.ReadBytes('$'); err != nil {
-						return err
-					}
-
-					if channelID, err = c.reader.ReadByte(); err != nil {
-						return err
-					}
-
-					// TODO: better check maximum good channel ID
-					if channelID >= 20 {
-						continue
-					}
-
-					buf4 = make([]byte, 2)
-					if _, err = io.ReadFull(c.reader, buf4); err != nil {
-						return err
-					}
-
-					// check if size good for RTP
-					size = binary.BigEndian.Uint16(buf4)
-					if size <= 1500 {
-						break
-					}
-
-					// 10 tries to find good packet
-					if i >= 10 {
-						return fmt.Errorf("RTSP wrong input")
-					}
-				}
-			}
-		} else {
-			// hope that the odd channels are always RTCP
-			channelID = buf4[1]
-
-			// get data size
-			size = binary.BigEndian.Uint16(buf4[2:])
-
-			// skip 4 bytes from c.reader.Peek
-			if _, err = c.reader.Discard(4); err != nil {
-				return
-			}
-		}
-
-		// init memory for data
-		buf := make([]byte, size)
-		if _, err = io.ReadFull(c.reader, buf); err != nil {
-			return
-		}
-
-		c.Recv += int(size)
-
-		if channelID&1 == 0 {
-			packet := &rtp.Packet{}
-			if err = packet.Unmarshal(buf); err != nil {
-				return
-			}
-
-			for _, receiver := range c.Receivers {
-				if receiver.ID == channelID {
-					receiver.WriteRTP(packet)
-					break
-				}
-			}
-		} else {
-			msg := &RTCP{Channel: channelID}
-
-			if err = msg.Header.Unmarshal(buf); err != nil {
-				continue
-			}
-
-			msg.Packets, err = rtcp.Unmarshal(buf)
-			if err != nil {
-				continue
-			}
-
-			c.Fire(msg)
-		}
-
-		if keepaliveDT != 0 && ts.After(keepaliveTS) {
-			req := &tcp.Request{Method: MethodOptions, URL: c.URL}
-			if err = c.WriteRequest(req); err != nil {
-				return
-			}
-
-			keepaliveTS = ts.Add(keepaliveDT)
-		}
 	}
 
 	return
 }
 
+func (c *Conn) handleKeepalive(ctx context.Context, d time.Duration) {
+	ticker := time.NewTicker(d)
+	for {
+		select {
+		case <-ticker.C:
+			req := &tcp.Request{Method: MethodOptions, URL: c.URL}
+			if err := c.WriteRequest(req); err != nil {
+				return
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (c *Conn) handleUDPData(channel byte) {
+	// TODO: handle timeouts and drop TCP connection after any error
+	conn := c.udpConn[channel]
+
+	for {
+		// TP-Link Tapo camera has crazy 10000 bytes packet size
+		buf := make([]byte, 10240)
+
+		n, _, err := conn.ReadFromUDP(buf)
+		if err != nil {
+			return
+		}
+
+		if err = c.handleRawPacket(channel, buf[:n]); err != nil {
+			return
+		}
+	}
+}
+
+func (c *Conn) handleTCPData() error {
+	// we can read:
+	// 1. RTP interleaved: `$` + 1B channel number + 2B size
+	// 2. RTSP response:   RTSP/1.0 200 OK
+	// 3. RTSP request:    OPTIONS ...
+	var buf4 []byte // `$` + 1B channel number + 2B size
+	var err error
+
+	buf4, err = c.reader.Peek(4)
+	if err != nil {
+		return err
+	}
+
+	var channel byte
+	var size uint16
+
+	if buf4[0] != '$' {
+		switch string(buf4) {
+		case "RTSP":
+			var res *tcp.Response
+			if res, err = c.ReadResponse(); err != nil {
+				return err
+			}
+			c.Fire(res)
+			// for playing backchannel only after OK response on play
+			c.playOK = true
+			return nil
+
+		case "OPTI", "TEAR", "DESC", "SETU", "PLAY", "PAUS", "RECO", "ANNO", "GET_", "SET_":
+			var req *tcp.Request
+			if req, err = c.ReadRequest(); err != nil {
+				return err
+			}
+			c.Fire(req)
+			if req.Method == MethodOptions {
+				res := &tcp.Response{Request: req}
+				if err = c.WriteResponse(res); err != nil {
+					return err
+				}
+			}
+			return nil
+
+		default:
+			c.Fire("RTSP wrong input")
+
+			for i := 0; ; i++ {
+				// search next start symbol
+				if _, err = c.reader.ReadBytes('$'); err != nil {
+					return err
+				}
+
+				if channel, err = c.reader.ReadByte(); err != nil {
+					return err
+				}
+
+				// TODO: better check maximum good channel ID
+				if channel >= 20 {
+					continue
+				}
+
+				buf4 = make([]byte, 2)
+				if _, err = io.ReadFull(c.reader, buf4); err != nil {
+					return err
+				}
+
+				// check if size good for RTP
+				size = binary.BigEndian.Uint16(buf4)
+				if size <= 1500 {
+					break
+				}
+
+				// 10 tries to find good packet
+				if i >= 10 {
+					return fmt.Errorf("RTSP wrong input")
+				}
+			}
+		}
+	} else {
+		// hope that the odd channels are always RTCP
+		channel = buf4[1]
+
+		// get data size
+		size = binary.BigEndian.Uint16(buf4[2:])
+
+		// skip 4 bytes from c.reader.Peek
+		if _, err = c.reader.Discard(4); err != nil {
+			return err
+		}
+	}
+
+	// init memory for data
+	buf := make([]byte, size)
+	if _, err = io.ReadFull(c.reader, buf); err != nil {
+		return err
+	}
+
+	c.Recv += int(size)
+
+	return c.handleRawPacket(channel, buf)
+}
+
+func (c *Conn) handleRawPacket(channel byte, buf []byte) error {
+	if channel&1 == 0 {
+		packet := &rtp.Packet{}
+		if err := packet.Unmarshal(buf); err != nil {
+			return err
+		}
+
+		for _, receiver := range c.Receivers {
+			if receiver.ID == channel {
+				receiver.WriteRTP(packet)
+				break
+			}
+		}
+	} else {
+		msg := &RTCP{Channel: channel}
+
+		if err := msg.Header.Unmarshal(buf); err != nil {
+			return nil
+		}
+
+		//var err error
+		//msg.Packets, err = rtcp.Unmarshal(buf)
+		//if err != nil {
+		//	return nil
+		//}
+
+		c.Fire(msg)
+	}
+
+	return nil
+}
+
 func (c *Conn) WriteRequest(req *tcp.Request) error {
 	if req.Proto == "" {
 		req.Proto = ProtoRTSP

pkg/rtsp/consumer.go

@@ -85,11 +85,8 @@ func (c *Conn) packetWriter(codec *core.Codec, channel, payloadType uint8) core.
 	}
 
 	flushBuf := func() {
-		if err := c.conn.SetWriteDeadline(time.Now().Add(Timeout)); err != nil {
-			return
-		}
 		//log.Printf("[rtsp] channel:%2d write_size:%6d buffer_size:%6d", channel, n, len(buf))
-		if _, err := c.conn.Write(buf[:n]); err == nil {
+		if err := c.writeInterleavedData(buf[:n]); err != nil {
 			c.Send += n
 		}
 		n = 0
@@ -177,3 +174,25 @@ func (c *Conn) packetWriter(codec *core.Codec, channel, payloadType uint8) core.
 
 	return handlerFunc
 }
+
+func (c *Conn) writeInterleavedData(data []byte) error {
+	if c.Transport != "udp" {
+		_ = c.conn.SetWriteDeadline(time.Now().Add(Timeout))
+		_, err := c.conn.Write(data)
+		return err
+	}
+
+	for len(data) >= 4 && data[0] == '$' {
+		channel := data[1]
+		size := uint16(data[2])<<8 | uint16(data[3])
+		rtpData := data[4 : 4+size]
+
+		if _, err := c.WriteToUDP(rtpData, channel); err != nil {
+			return err
+		}
+
+		data = data[4+size:]
+	}
+
+	return nil
+}