CEF/mediamtx
1
0
Fork 0
forked from External/mediamtx
Code Pull requests Activity

open the firewall in case of udp connections

Browse source
This commit is contained in:
aler9 2020-09-02 10:59:02 +02:00
parent 3c48518d99
commit dd151e9978
5 changed files with 53 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

34
client.go
View file

@ -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:

2
go.mod
View file

@ -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

4
go.sum
View file

@ -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=

23
main.go
View file

@ -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,
})
}

32
server-udp.go
View file

@ -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}
}