From dd151e99786f29de3821f06e442efd4fc78b3766 Mon Sep 17 00:00:00 2001 From: aler9 <46489434+aler9@users.noreply.github.com> Date: Wed, 2 Sep 2020 10:59:02 +0200 Subject: [PATCH] open the firewall in case of udp connections --- client.go | 34 +++++++++++++++++++++++++--------- go.mod | 2 +- go.sum | 4 ++-- main.go | 23 +++++++++-------------- server-udp.go | 32 ++++++++++++++++---------------- 5 files changed, 53 insertions(+), 42 deletions(-) diff --git a/client.go b/client.go index dd4763b9..2b9a3bd0 100644 --- a/client.go +++ b/client.go @@ -1074,6 +1074,25 @@ func (c *client) runRecord() bool { } func (c *client) runRecordUdp() { + // open the firewall by sending packets to every channel + for _, track := range c.streamTracks { + c.p.serverRtp.write( + []byte{0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}, + &net.UDPAddr{ + IP: c.ip(), + Zone: c.zone(), + Port: track.rtpPort, + }) + + c.p.serverRtcp.write( + []byte{0x80, 0xc9, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00}, + &net.UDPAddr{ + IP: c.ip(), + Zone: c.zone(), + Port: track.rtcpPort, + }) + } + readDone := make(chan error) go func() { for { @@ -1111,7 +1130,7 @@ func (c *client) runRecordUdp() { case <-checkStreamTicker.C: for trackId := range c.streamTracks { if time.Since(c.rtcpReceivers[trackId].LastFrameTime()) >= c.p.conf.ReadTimeout { - c.log("ERR: stream is dead") + c.log("ERR: no packets received recently (maybe there's a firewall/NAT)") c.conn.Close() <-readDone c.p.clientClose <- c @@ -1123,14 +1142,11 @@ func (c *client) runRecordUdp() { case <-receiverReportTicker.C: for trackId := range c.streamTracks { frame := c.rtcpReceivers[trackId].Report() - c.p.serverRtcp.writeChan <- &udpAddrBufPair{ - addr: &net.UDPAddr{ - IP: c.ip(), - Zone: c.zone(), - Port: c.streamTracks[trackId].rtcpPort, - }, - buf: frame, - } + c.p.serverRtcp.write(frame, &net.UDPAddr{ + IP: c.ip(), + Zone: c.zone(), + Port: c.streamTracks[trackId].rtcpPort, + }) } case <-c.terminate: diff --git a/go.mod b/go.mod index 9d016b25..6d227652 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.12 require ( github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d // indirect - github.com/aler9/gortsplib v0.0.0-20200831134901-89f0204be80e + github.com/aler9/gortsplib v0.0.0-20200902092059-0e827e1bc452 github.com/aler9/sdp/v3 v3.0.0-20200719093237-2c3d108a7436 github.com/davecgh/go-spew v1.1.1 // indirect github.com/stretchr/testify v1.6.1 diff --git a/go.sum b/go.sum index b6b56fab..893e4c76 100644 --- a/go.sum +++ b/go.sum @@ -2,8 +2,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafo github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d h1:UQZhZ2O0vMHr2cI+DC1Mbh0TJxzA3RcLoMsFw+aXw7E= github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho= -github.com/aler9/gortsplib v0.0.0-20200831134901-89f0204be80e h1:aCZXDnNPDFltytaHe/yCUmJmd7/fglhwLwStH47U6X8= -github.com/aler9/gortsplib v0.0.0-20200831134901-89f0204be80e/go.mod h1:kBMvjIdOHRjLdV+oT28JD72JUPpJuwxOc9u72GG8GpY= +github.com/aler9/gortsplib v0.0.0-20200902092059-0e827e1bc452 h1:w7XwvxvynGYa8YEH3QEHFdbDTlGkOZIaJAl2moEOLIA= +github.com/aler9/gortsplib v0.0.0-20200902092059-0e827e1bc452/go.mod h1:kBMvjIdOHRjLdV+oT28JD72JUPpJuwxOc9u72GG8GpY= github.com/aler9/sdp/v3 v3.0.0-20200719093237-2c3d108a7436 h1:W0iNErWKvSAyJBNVx+qQoyFrWOFVgS6f/WEME/D3EZc= github.com/aler9/sdp/v3 v3.0.0-20200719093237-2c3d108a7436/go.mod h1:OnlEK3QI7YtM+ShZWtGajmOHLZ3bjU80AcIS5e34i1U= github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= diff --git a/main.go b/main.go index 6ab49daa..e9b136ac 100644 --- a/main.go +++ b/main.go @@ -427,22 +427,17 @@ func (p *program) forwardFrame(path *path, trackId int, streamType gortsplib.Str if c.streamProtocol == gortsplib.StreamProtocolUdp { if streamType == gortsplib.StreamTypeRtp { - p.serverRtp.write(&udpAddrBufPair{ - addr: &net.UDPAddr{ - IP: c.ip(), - Zone: c.zone(), - Port: track.rtpPort, - }, - buf: frame, + p.serverRtp.write(frame, &net.UDPAddr{ + IP: c.ip(), + Zone: c.zone(), + Port: track.rtpPort, }) + } else { - p.serverRtcp.write(&udpAddrBufPair{ - addr: &net.UDPAddr{ - IP: c.ip(), - Zone: c.zone(), - Port: track.rtcpPort, - }, - buf: frame, + p.serverRtcp.write(frame, &net.UDPAddr{ + IP: c.ip(), + Zone: c.zone(), + Port: track.rtcpPort, }) } diff --git a/server-udp.go b/server-udp.go index 13a44b9b..8e0ec075 100644 --- a/server-udp.go +++ b/server-udp.go @@ -7,23 +7,23 @@ import ( "github.com/aler9/gortsplib" ) -type udpAddrBufPair struct { - addr *net.UDPAddr +type udpBufAddrPair struct { buf []byte + addr *net.UDPAddr } type serverUdp struct { p *program - conn *net.UDPConn + pc *net.UDPConn streamType gortsplib.StreamType readBuf *multiBuffer - writeChan chan *udpAddrBufPair - done chan struct{} + writec chan udpBufAddrPair + done chan struct{} } func newServerUdp(p *program, port int, streamType gortsplib.StreamType) (*serverUdp, error) { - conn, err := net.ListenUDP("udp", &net.UDPAddr{ + pc, err := net.ListenUDP("udp", &net.UDPAddr{ Port: port, }) if err != nil { @@ -32,10 +32,10 @@ func newServerUdp(p *program, port int, streamType gortsplib.StreamType) (*serve l := &serverUdp{ p: p, - conn: conn, + pc: pc, streamType: streamType, readBuf: newMultiBuffer(3, clientUdpReadBufferSize), - writeChan: make(chan *udpAddrBufPair), + writec: make(chan udpBufAddrPair), done: make(chan struct{}), } @@ -57,15 +57,15 @@ func (l *serverUdp) run() { writeDone := make(chan struct{}) go func() { defer close(writeDone) - for w := range l.writeChan { - l.conn.SetWriteDeadline(time.Now().Add(l.p.conf.WriteTimeout)) - l.conn.WriteTo(w.buf, w.addr) + for w := range l.writec { + l.pc.SetWriteDeadline(time.Now().Add(l.p.conf.WriteTimeout)) + l.pc.WriteTo(w.buf, w.addr) } }() for { buf := l.readBuf.next() - n, addr, err := l.conn.ReadFromUDP(buf) + n, addr, err := l.pc.ReadFromUDP(buf) if err != nil { break } @@ -77,17 +77,17 @@ func (l *serverUdp) run() { } } - close(l.writeChan) + close(l.writec) <-writeDone close(l.done) } func (l *serverUdp) close() { - l.conn.Close() + l.pc.Close() <-l.done } -func (l *serverUdp) write(pair *udpAddrBufPair) { - l.writeChan <- pair +func (l *serverUdp) write(data []byte, addr *net.UDPAddr) { + l.writec <- udpBufAddrPair{data, addr} }