diff --git a/pkg/rtsp/client.go b/pkg/rtsp/client.go index 7fc134fc..4e891213 100644 --- a/pkg/rtsp/client.go +++ b/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 == "" { - timeout := core.ConnDialTimeout + switch c.Transport { + 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 - 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 + if c.Transport == "udp" { + conn1, conn2, err := ListenUDPPair() + if err != nil { + return 0, err + } + + c.udpConn = append(c.udpConn, conn1, conn2) + + 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 - - // 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 + // Parse server response 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=", "-") - i, err := strconv.Atoi(channel) - if err != nil { - return 0, err + return byte(i), nil } - - 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") +} diff --git a/pkg/rtsp/conn.go b/pkg/rtsp/conn.go index 0c2009d7..2984c781 100644 --- a/pkg/rtsp/conn.go +++ b/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 diff --git a/pkg/rtsp/consumer.go b/pkg/rtsp/consumer.go index 860ed113..e6525d96 100644 --- a/pkg/rtsp/consumer.go +++ b/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 +}