From 9fb844331ed22cc92beeaf263113a211176224d3 Mon Sep 17 00:00:00 2001 From: aler9 <46489434+aler9@users.noreply.github.com> Date: Wed, 6 Jan 2021 12:40:18 +0100 Subject: [PATCH] move UDP support into gortsplib --- go.mod | 2 +- go.sum | 4 +- internal/client/client.go | 160 +++------------- internal/clientman/clientman.go | 39 ++-- internal/conf/conf.go | 16 +- internal/{servertcp => serverplain}/server.go | 6 +- internal/serverudp/server.go | 176 ------------------ internal/serverudpl/server.go | 34 ++++ internal/sourcertmp/source.go | 28 +-- internal/sourcertsp/source.go | 4 +- main.go | 104 +++++++---- main_test.go | 22 +-- 12 files changed, 195 insertions(+), 400 deletions(-) rename internal/{servertcp => serverplain}/server.go (88%) delete mode 100644 internal/serverudp/server.go create mode 100644 internal/serverudpl/server.go diff --git a/go.mod b/go.mod index c196fe1b..a5f93931 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.15 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-20201231182741-9bd587e576f1 + github.com/aler9/gortsplib v0.0.0-20210106112607-8e70ac4d59c4 github.com/davecgh/go-spew v1.1.1 // indirect github.com/fsnotify/fsnotify v1.4.9 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 diff --git a/go.sum b/go.sum index 7610ac4f..a0344d79 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-20201231182741-9bd587e576f1 h1:RqOBalGfTyA43DigWv64uKNEiAIJT9IPiXrjkDbR/Lo= -github.com/aler9/gortsplib v0.0.0-20201231182741-9bd587e576f1/go.mod h1:8P09VjpiPJFyfkVosyF5/TY82jNwkMN165NS/7sc32I= +github.com/aler9/gortsplib v0.0.0-20210106112607-8e70ac4d59c4 h1:gfqoSKl2KyzzZjHfs//ImOtX6u1vgbcFcp0bUCd6Q2Q= +github.com/aler9/gortsplib v0.0.0-20210106112607-8e70ac4d59c4/go.mod h1:8P09VjpiPJFyfkVosyF5/TY82jNwkMN165NS/7sc32I= github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= diff --git a/internal/client/client.go b/internal/client/client.go index 57f45f0b..49979661 100644 --- a/internal/client/client.go +++ b/internal/client/client.go @@ -20,7 +20,6 @@ import ( "github.com/aler9/rtsp-simple-server/internal/conf" "github.com/aler9/rtsp-simple-server/internal/externalcmd" "github.com/aler9/rtsp-simple-server/internal/logger" - "github.com/aler9/rtsp-simple-server/internal/serverudp" "github.com/aler9/rtsp-simple-server/internal/stats" ) @@ -98,8 +97,6 @@ type Client struct { protocols map[gortsplib.StreamProtocol]struct{} wg *sync.WaitGroup stats *stats.Stats - serverUDPRtp *serverudp.Server - serverUDPRtcp *serverudp.Server conn *gortsplib.ServerConn parent Parent @@ -111,7 +108,7 @@ type Client struct { authFailures int streamProtocol gortsplib.StreamProtocol streamTracks map[int]*streamTrack - rtcpReceivers map[int]*rtcpreceiver.RtcpReceiver + rtcpReceivers map[int]*rtcpreceiver.RTCPReceiver udpLastFrameTimes []*int64 onReadCmd *externalcmd.Cmd onPublishCmd *externalcmd.Cmd @@ -134,8 +131,6 @@ func New( protocols map[gortsplib.StreamProtocol]struct{}, wg *sync.WaitGroup, stats *stats.Stats, - serverUDPRtp *serverudp.Server, - serverUDPRtcp *serverudp.Server, conn *gortsplib.ServerConn, parent Parent) *Client { @@ -147,13 +142,11 @@ func New( protocols: protocols, wg: wg, stats: stats, - serverUDPRtp: serverUDPRtp, - serverUDPRtcp: serverUDPRtcp, conn: conn, parent: parent, state: stateInitial, streamTracks: make(map[int]*streamTrack), - rtcpReceivers: make(map[int]*rtcpreceiver.RtcpReceiver), + rtcpReceivers: make(map[int]*rtcpreceiver.RTCPReceiver), terminate: make(chan struct{}), } @@ -187,10 +180,6 @@ func (c *Client) ip() net.IP { return c.conn.NetConn().RemoteAddr().(*net.TCPAddr).IP } -func (c *Client) zone() string { - return c.conn.NetConn().RemoteAddr().(*net.TCPAddr).Zone -} - var errTerminated = errors.New("terminated") func (c *Client) run() { @@ -462,21 +451,10 @@ func (c *Client) run() { rtcpPort: (*th.ClientPorts)[1], } - th := headers.Transport{ - Protocol: gortsplib.StreamProtocolUDP, - Delivery: func() *base.StreamDelivery { - v := base.StreamDeliveryUnicast - return &v - }(), - ClientPorts: th.ClientPorts, - ServerPorts: &[2]int{c.serverUDPRtp.Port(), c.serverUDPRtcp.Port()}, - } - return &base.Response{ StatusCode: base.StatusOK, Header: base.Header{ - "Transport": th.Write(), - "Session": base.HeaderValue{sessionID}, + "Session": base.HeaderValue{sessionID}, }, }, nil } @@ -528,18 +506,10 @@ func (c *Client) run() { rtcpPort: 0, } - interleavedIds := [2]int{trackID * 2, (trackID * 2) + 1} - - th := headers.Transport{ - Protocol: gortsplib.StreamProtocolTCP, - InterleavedIds: &interleavedIds, - } - return &base.Response{ StatusCode: base.StatusOK, Header: base.Header{ - "Transport": th.Write(), - "Session": base.HeaderValue{sessionID}, + "Session": base.HeaderValue{sessionID}, }, }, nil @@ -591,21 +561,10 @@ func (c *Client) run() { rtcpPort: (*th.ClientPorts)[1], } - th := headers.Transport{ - Protocol: gortsplib.StreamProtocolUDP, - Delivery: func() *base.StreamDelivery { - v := base.StreamDeliveryUnicast - return &v - }(), - ClientPorts: th.ClientPorts, - ServerPorts: &[2]int{c.serverUDPRtp.Port(), c.serverUDPRtcp.Port()}, - } - return &base.Response{ StatusCode: base.StatusOK, Header: base.Header{ - "Transport": th.Write(), - "Session": base.HeaderValue{sessionID}, + "Session": base.HeaderValue{sessionID}, }, }, nil } @@ -650,16 +609,10 @@ func (c *Client) run() { rtcpPort: 0, } - ht := headers.Transport{ - Protocol: gortsplib.StreamProtocolTCP, - InterleavedIds: &interleavedIds, - } - return &base.Response{ StatusCode: base.StatusOK, Header: base.Header{ - "Transport": ht.Write(), - "Session": base.HeaderValue{sessionID}, + "Session": base.HeaderValue{sessionID}, }, }, nil @@ -773,18 +726,11 @@ func (c *Client) run() { switch c.state { case statePlay: - if c.streamProtocol == gortsplib.StreamProtocolTCP { - c.conn.EnableFrames(false) - } c.stopPlay() c.state = statePrePlay c.path.OnClientPause(c) case stateRecord: - if c.streamProtocol == gortsplib.StreamProtocolTCP { - c.conn.EnableFrames(false) - c.conn.EnableReadTimeout(false) - } c.stopRecord() c.state = statePreRecord c.path.OnClientPause(c) @@ -798,14 +744,24 @@ func (c *Client) run() { }, nil } - onFrame := func(trackID int, streamType gortsplib.StreamType, content []byte) { - if c.state == stateRecord { + onFrame := func(trackID int, streamType gortsplib.StreamType, payload []byte) { + if c.state != stateRecord { + return + } + + if c.streamProtocol == gortsplib.StreamProtocolUDP { + now := time.Now() + atomic.StoreInt64(c.udpLastFrameTimes[trackID], now.Unix()) + c.rtcpReceivers[trackID].ProcessFrame(now, streamType, payload) + c.path.OnFrame(trackID, streamType, payload) + + } else { if trackID >= len(c.streamTracks) { return } - c.rtcpReceivers[trackID].ProcessFrame(time.Now(), streamType, content) - c.path.OnFrame(trackID, streamType, content) + c.rtcpReceivers[trackID].ProcessFrame(time.Now(), streamType, payload) + c.path.OnFrame(trackID, streamType, payload) } } @@ -974,10 +930,6 @@ func (c *Client) startPlay() { Port: strconv.FormatInt(int64(c.rtspPort), 10), }) } - - if c.streamProtocol == gortsplib.StreamProtocolTCP { - c.conn.EnableFrames(true) - } } func (c *Client) stopPlay() { @@ -1004,28 +956,13 @@ func (c *Client) startRecord() { c.udpLastFrameTimes[trackID] = &v } - for trackID, track := range c.streamTracks { - c.serverUDPRtp.AddPublisher(c.ip(), track.rtpPort, c, trackID) - c.serverUDPRtcp.AddPublisher(c.ip(), track.rtcpPort, c, trackID) - } - // open the firewall by sending packets to the counterpart - for _, track := range c.streamTracks { - c.serverUDPRtp.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, - }) + for trackID := range c.streamTracks { + c.conn.WriteFrame(trackID, gortsplib.StreamTypeRTP, + []byte{0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}) - c.serverUDPRtcp.Write( - []byte{0x80, 0xc9, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00}, - &net.UDPAddr{ - IP: c.ip(), - Zone: c.zone(), - Port: track.rtcpPort, - }) + c.conn.WriteFrame(trackID, gortsplib.StreamTypeRTCP, + []byte{0x80, 0xc9, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00}) } } @@ -1042,8 +979,6 @@ func (c *Client) startRecord() { if c.streamProtocol == gortsplib.StreamProtocolUDP { go c.backgroundRecordUDP() } else { - c.conn.EnableFrames(true) - c.conn.EnableReadTimeout(true) go c.backgroundRecordTCP() } } @@ -1052,13 +987,6 @@ func (c *Client) stopRecord() { close(c.backgroundRecordTerminate) <-c.backgroundRecordDone - if c.streamProtocol == gortsplib.StreamProtocolUDP { - for _, track := range c.streamTracks { - c.serverUDPRtp.RemovePublisher(c.ip(), track.rtpPort, c) - c.serverUDPRtcp.RemovePublisher(c.ip(), track.rtcpPort, c) - } - } - if c.path.Conf().RunOnPublish != "" { c.onPublishCmd.Close() } @@ -1092,11 +1020,7 @@ func (c *Client) backgroundRecordUDP() { now := time.Now() for trackID := range c.streamTracks { r := c.rtcpReceivers[trackID].Report(now) - c.serverUDPRtcp.Write(r, &net.UDPAddr{ - IP: c.ip(), - Zone: c.zone(), - Port: c.streamTracks[trackID].rtcpPort, - }) + c.conn.WriteFrame(trackID, gortsplib.StreamTypeRTP, r) } case <-c.backgroundRecordTerminate: @@ -1117,7 +1041,7 @@ func (c *Client) backgroundRecordTCP() { now := time.Now() for trackID := range c.streamTracks { r := c.rtcpReceivers[trackID].Report(now) - c.conn.WriteFrame(trackID, gortsplib.StreamTypeRtcp, r) + c.conn.WriteFrame(trackID, gortsplib.StreamTypeRTCP, r) } case <-c.backgroundRecordTerminate: @@ -1126,40 +1050,14 @@ func (c *Client) backgroundRecordTCP() { } } -// OnUDPPublisherFrame implements serverudp.Publisher. -func (c *Client) OnUDPPublisherFrame(trackID int, streamType base.StreamType, buf []byte) { - now := time.Now() - atomic.StoreInt64(c.udpLastFrameTimes[trackID], now.Unix()) - c.rtcpReceivers[trackID].ProcessFrame(now, streamType, buf) - c.path.OnFrame(trackID, streamType, buf) -} - // OnReaderFrame implements path.Reader. func (c *Client) OnReaderFrame(trackID int, streamType base.StreamType, buf []byte) { - track, ok := c.streamTracks[trackID] + _, ok := c.streamTracks[trackID] if !ok { return } - if c.streamProtocol == gortsplib.StreamProtocolUDP { - if streamType == gortsplib.StreamTypeRtp { - c.serverUDPRtp.Write(buf, &net.UDPAddr{ - IP: c.ip(), - Zone: c.zone(), - Port: track.rtpPort, - }) - - } else { - c.serverUDPRtcp.Write(buf, &net.UDPAddr{ - IP: c.ip(), - Zone: c.zone(), - Port: track.rtcpPort, - }) - } - - } else { - c.conn.WriteFrame(trackID, streamType, buf) - } + c.conn.WriteFrame(trackID, streamType, buf) } // OnPathDescribeData is called by path.Path. diff --git a/internal/clientman/clientman.go b/internal/clientman/clientman.go index 9f52c31e..827b774e 100644 --- a/internal/clientman/clientman.go +++ b/internal/clientman/clientman.go @@ -10,9 +10,8 @@ import ( "github.com/aler9/rtsp-simple-server/internal/client" "github.com/aler9/rtsp-simple-server/internal/logger" "github.com/aler9/rtsp-simple-server/internal/pathman" - "github.com/aler9/rtsp-simple-server/internal/servertcp" + "github.com/aler9/rtsp-simple-server/internal/serverplain" "github.com/aler9/rtsp-simple-server/internal/servertls" - "github.com/aler9/rtsp-simple-server/internal/serverudp" "github.com/aler9/rtsp-simple-server/internal/stats" ) @@ -29,10 +28,8 @@ type ClientManager struct { runOnConnectRestart bool protocols map[base.StreamProtocol]struct{} stats *stats.Stats - serverUDPRtp *serverudp.Server - serverUDPRtcp *serverudp.Server pathMan *pathman.PathManager - serverTCP *servertcp.Server + serverTCP *serverplain.Server serverTLS *servertls.Server parent Parent @@ -55,10 +52,8 @@ func New( runOnConnectRestart bool, protocols map[base.StreamProtocol]struct{}, stats *stats.Stats, - serverUDPRtp *serverudp.Server, - serverUDPRtcp *serverudp.Server, pathMan *pathman.PathManager, - serverTCP *servertcp.Server, + serverTCP *serverplain.Server, serverTLS *servertls.Server, parent Parent) *ClientManager { @@ -69,8 +64,6 @@ func New( runOnConnectRestart: runOnConnectRestart, protocols: protocols, stats: stats, - serverUDPRtp: serverUDPRtp, - serverUDPRtcp: serverUDPRtcp, pathMan: pathMan, serverTCP: serverTCP, serverTLS: serverTLS, @@ -117,15 +110,29 @@ outer: for { select { case conn := <-tcpAccept: - c := client.New(false, cm.rtspPort, cm.readTimeout, - cm.runOnConnect, cm.runOnConnectRestart, cm.protocols, &cm.wg, - cm.stats, cm.serverUDPRtp, cm.serverUDPRtcp, conn, cm) + c := client.New(false, + cm.rtspPort, + cm.readTimeout, + cm.runOnConnect, + cm.runOnConnectRestart, + cm.protocols, + &cm.wg, + cm.stats, + conn, + cm) cm.clients[c] = struct{}{} case conn := <-tlsAccept: - c := client.New(true, cm.rtspPort, cm.readTimeout, - cm.runOnConnect, cm.runOnConnectRestart, cm.protocols, &cm.wg, - cm.stats, cm.serverUDPRtp, cm.serverUDPRtcp, conn, cm) + c := client.New(true, + cm.rtspPort, + cm.readTimeout, + cm.runOnConnect, + cm.runOnConnectRestart, + cm.protocols, + &cm.wg, + cm.stats, + conn, + cm) cm.clients[c] = struct{}{} case c := <-cm.pathMan.ClientClose(): diff --git a/internal/conf/conf.go b/internal/conf/conf.go index 05aa2843..89d9e25a 100644 --- a/internal/conf/conf.go +++ b/internal/conf/conf.go @@ -36,8 +36,8 @@ type Conf struct { EncryptionParsed Encryption `yaml:"-" json:"-"` RtspPort int `yaml:"rtspPort"` RtspsPort int `yaml:"rtspsPort"` - RtpPort int `yaml:"rtpPort"` - RtcpPort int `yaml:"rtcpPort"` + RTPPort int `yaml:"rtpPort"` + RTCPPort int `yaml:"rtcpPort"` ServerKey string `yaml:"serverKey"` ServerCert string `yaml:"serverCert"` AuthMethods []string `yaml:"authMethods"` @@ -140,16 +140,16 @@ func (conf *Conf) fillAndCheck() error { if conf.RtspsPort == 0 { conf.RtspsPort = 8555 } - if conf.RtpPort == 0 { - conf.RtpPort = 8000 + if conf.RTPPort == 0 { + conf.RTPPort = 8000 } - if (conf.RtpPort % 2) != 0 { + if (conf.RTPPort % 2) != 0 { return fmt.Errorf("rtp port must be even") } - if conf.RtcpPort == 0 { - conf.RtcpPort = 8001 + if conf.RTCPPort == 0 { + conf.RTCPPort = 8001 } - if conf.RtcpPort != (conf.RtpPort + 1) { + if conf.RTCPPort != (conf.RTPPort + 1) { return fmt.Errorf("rtcp and rtp ports must be consecutive") } diff --git a/internal/servertcp/server.go b/internal/serverplain/server.go similarity index 88% rename from internal/servertcp/server.go rename to internal/serverplain/server.go index fc3f2be6..39227d55 100644 --- a/internal/servertcp/server.go +++ b/internal/serverplain/server.go @@ -1,4 +1,4 @@ -package servertcp +package serverplain import ( "strconv" @@ -29,12 +29,16 @@ type Server struct { func New(port int, readTimeout time.Duration, writeTimeout time.Duration, + udpRTPListener *gortsplib.ServerUDPListener, + udpRTCPListener *gortsplib.ServerUDPListener, parent Parent) (*Server, error) { conf := gortsplib.ServerConf{ ReadTimeout: readTimeout, WriteTimeout: writeTimeout, ReadBufferCount: 1, + UDPRTPListener: udpRTPListener, + UDPRTCPListener: udpRTCPListener, } srv, err := conf.Serve(":" + strconv.FormatInt(int64(port), 10)) diff --git a/internal/serverudp/server.go b/internal/serverudp/server.go deleted file mode 100644 index b0a7ab33..00000000 --- a/internal/serverudp/server.go +++ /dev/null @@ -1,176 +0,0 @@ -package serverudp - -import ( - "net" - "sync" - "time" - - "github.com/aler9/gortsplib" - "github.com/aler9/gortsplib/pkg/base" - "github.com/aler9/gortsplib/pkg/multibuffer" - "github.com/aler9/rtsp-simple-server/internal/logger" -) - -const ( - // use the same buffer size as gstreamer's rtspsrc - kernelReadBufferSize = 0x80000 - - readBufferSize = 2048 -) - -// Publisher is implemented by client.Client. -type Publisher interface { - OnUDPPublisherFrame(int, base.StreamType, []byte) -} - -type publisherData struct { - publisher Publisher - trackID int -} - -// Parent is implemented by program. -type Parent interface { - Log(logger.Level, string, ...interface{}) -} - -type publisherAddr struct { - ip [net.IPv6len]byte // use a fixed-size array to enable the equality operator - port int -} - -func (p *publisherAddr) fill(ip net.IP, port int) { - p.port = port - - if len(ip) == net.IPv4len { - copy(p.ip[0:], []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}) // v4InV6Prefix - copy(p.ip[12:], ip) - } else { - copy(p.ip[:], ip) - } -} - -// Server is a UDP server that can be used to send and receive both RTP and RTCP. -type Server struct { - writeTimeout time.Duration - streamType gortsplib.StreamType - - pc *net.UDPConn - readBuf *multibuffer.MultiBuffer - publishersMutex sync.RWMutex - publishers map[publisherAddr]*publisherData - writeMutex sync.Mutex - - // out - done chan struct{} -} - -// New allocates a Server. -func New(writeTimeout time.Duration, - port int, - streamType gortsplib.StreamType, - parent Parent) (*Server, error) { - - pc, err := net.ListenUDP("udp", &net.UDPAddr{ - Port: port, - }) - if err != nil { - return nil, err - } - - err = pc.SetReadBuffer(kernelReadBufferSize) - if err != nil { - return nil, err - } - - s := &Server{ - writeTimeout: writeTimeout, - streamType: streamType, - pc: pc, - readBuf: multibuffer.New(1, readBufferSize), - publishers: make(map[publisherAddr]*publisherData), - done: make(chan struct{}), - } - - var label string - if s.streamType == gortsplib.StreamTypeRtp { - label = "RTP" - } else { - label = "RTCP" - } - parent.Log(logger.Info, "[UDP/"+label+" server] opened on :%d", port) - - go s.run() - return s, nil -} - -// Close closes a Server. -func (s *Server) Close() { - s.pc.Close() - <-s.done -} - -func (s *Server) run() { - defer close(s.done) - - for { - buf := s.readBuf.Next() - n, addr, err := s.pc.ReadFromUDP(buf) - if err != nil { - break - } - - func() { - s.publishersMutex.RLock() - defer s.publishersMutex.RUnlock() - - // find publisher data - var pubAddr publisherAddr - pubAddr.fill(addr.IP, addr.Port) - pubData, ok := s.publishers[pubAddr] - if !ok { - return - } - - pubData.publisher.OnUDPPublisherFrame(pubData.trackID, s.streamType, buf[:n]) - }() - } -} - -// Port returns the server local port. -func (s *Server) Port() int { - return s.pc.LocalAddr().(*net.UDPAddr).Port -} - -// Write writes a UDP packet. -func (s *Server) Write(buf []byte, addr *net.UDPAddr) { - s.writeMutex.Lock() - defer s.writeMutex.Unlock() - - s.pc.SetWriteDeadline(time.Now().Add(s.writeTimeout)) - s.pc.WriteTo(buf, addr) -} - -// AddPublisher adds a publisher. -func (s *Server) AddPublisher(ip net.IP, port int, publisher Publisher, trackID int) { - s.publishersMutex.Lock() - defer s.publishersMutex.Unlock() - - var addr publisherAddr - addr.fill(ip, port) - - s.publishers[addr] = &publisherData{ - publisher: publisher, - trackID: trackID, - } -} - -// RemovePublisher removes a publisher. -func (s *Server) RemovePublisher(ip net.IP, port int, publisher Publisher) { - s.publishersMutex.Lock() - defer s.publishersMutex.Unlock() - - var addr publisherAddr - addr.fill(ip, port) - - delete(s.publishers, addr) -} diff --git a/internal/serverudpl/server.go b/internal/serverudpl/server.go new file mode 100644 index 00000000..0a4d9b85 --- /dev/null +++ b/internal/serverudpl/server.go @@ -0,0 +1,34 @@ +package serverudpl + +import ( + "strconv" + + "github.com/aler9/gortsplib" + "github.com/aler9/rtsp-simple-server/internal/logger" +) + +// Parent is implemented by program. +type Parent interface { + Log(logger.Level, string, ...interface{}) +} + +// New allocates a gortsplib.ServerUDPListener. +func New(port int, + streamType gortsplib.StreamType, + parent Parent) (*gortsplib.ServerUDPListener, error) { + + listener, err := gortsplib.NewServerUDPListener(":" + strconv.FormatInt(int64(port), 10)) + if err != nil { + return nil, err + } + + label := func() string { + if streamType == gortsplib.StreamTypeRTP { + return "RTP" + } + return "RTCP" + }() + parent.Log(logger.Info, "[UDP/"+label+" listener] opened on :%d", port) + + return listener, nil +} diff --git a/internal/sourcertmp/source.go b/internal/sourcertmp/source.go index ca4ca585..c3205133 100644 --- a/internal/sourcertmp/source.go +++ b/internal/sourcertmp/source.go @@ -191,11 +191,11 @@ func (s *Source) runInner() bool { var tracks gortsplib.Tracks var videoTrack *gortsplib.Track - var videoRtcpSender *rtcpsender.RtcpSender + var videoRTCPSender *rtcpsender.RTCPSender var h264Encoder *rtph264.Encoder var audioTrack *gortsplib.Track - var audioRtcpSender *rtcpsender.RtcpSender + var audioRTCPSender *rtcpsender.RTCPSender var aacEncoder *rtpaac.Encoder if h264Sps != nil { @@ -206,7 +206,7 @@ func (s *Source) runInner() bool { } clockRate, _ := videoTrack.ClockRate() - videoRtcpSender = rtcpsender.New(clockRate) + videoRTCPSender = rtcpsender.New(clockRate) h264Encoder, err = rtph264.NewEncoder(96) if err != nil { @@ -225,7 +225,7 @@ func (s *Source) runInner() bool { } clockRate, _ := audioTrack.ClockRate() - audioRtcpSender = rtcpsender.New(clockRate) + audioRTCPSender = rtcpsender.New(clockRate) aacEncoder, err = rtpaac.NewEncoder(96, clockRate) if err != nil { @@ -262,17 +262,17 @@ func (s *Source) runInner() bool { case <-t.C: now := time.Now() - if videoRtcpSender != nil { - r := videoRtcpSender.Report(now) + if videoRTCPSender != nil { + r := videoRTCPSender.Report(now) if r != nil { - s.parent.OnFrame(videoTrack.ID, gortsplib.StreamTypeRtcp, r) + s.parent.OnFrame(videoTrack.ID, gortsplib.StreamTypeRTCP, r) } } - if audioRtcpSender != nil { - r := audioRtcpSender.Report(now) + if audioRTCPSender != nil { + r := audioRTCPSender.Report(now) if r != nil { - s.parent.OnFrame(audioTrack.ID, gortsplib.StreamTypeRtcp, r) + s.parent.OnFrame(audioTrack.ID, gortsplib.StreamTypeRTCP, r) } } @@ -313,8 +313,8 @@ func (s *Source) runInner() bool { } for _, f := range frames { - videoRtcpSender.ProcessFrame(time.Now(), gortsplib.StreamTypeRtp, f) - s.parent.OnFrame(videoTrack.ID, gortsplib.StreamTypeRtp, f) + videoRTCPSender.ProcessFrame(time.Now(), gortsplib.StreamTypeRTP, f) + s.parent.OnFrame(videoTrack.ID, gortsplib.StreamTypeRTP, f) } case av.AAC: @@ -330,8 +330,8 @@ func (s *Source) runInner() bool { } for _, f := range frames { - audioRtcpSender.ProcessFrame(time.Now(), gortsplib.StreamTypeRtp, f) - s.parent.OnFrame(audioTrack.ID, gortsplib.StreamTypeRtp, f) + audioRTCPSender.ProcessFrame(time.Now(), gortsplib.StreamTypeRTP, f) + s.parent.OnFrame(audioTrack.ID, gortsplib.StreamTypeRTP, f) } default: diff --git a/internal/sourcertsp/source.go b/internal/sourcertsp/source.go index 98b552a9..83d436fe 100644 --- a/internal/sourcertsp/source.go +++ b/internal/sourcertsp/source.go @@ -149,8 +149,8 @@ func (s *Source) runInner() bool { s.parent.OnSourceSetReady(tracks) defer s.parent.OnSourceSetNotReady() - done := conn.ReadFrames(func(trackID int, streamType gortsplib.StreamType, content []byte) { - s.parent.OnFrame(trackID, streamType, content) + done := conn.ReadFrames(func(trackID int, streamType gortsplib.StreamType, payload []byte) { + s.parent.OnFrame(trackID, streamType, payload) }) for { diff --git a/main.go b/main.go index 6aafecc3..0424f538 100644 --- a/main.go +++ b/main.go @@ -16,9 +16,9 @@ import ( "github.com/aler9/rtsp-simple-server/internal/metrics" "github.com/aler9/rtsp-simple-server/internal/pathman" "github.com/aler9/rtsp-simple-server/internal/pprof" - "github.com/aler9/rtsp-simple-server/internal/servertcp" + "github.com/aler9/rtsp-simple-server/internal/serverplain" "github.com/aler9/rtsp-simple-server/internal/servertls" - "github.com/aler9/rtsp-simple-server/internal/serverudp" + "github.com/aler9/rtsp-simple-server/internal/serverudpl" "github.com/aler9/rtsp-simple-server/internal/stats" ) @@ -32,9 +32,9 @@ type program struct { logger *logger.Logger metrics *metrics.Metrics pprof *pprof.Pprof - serverUDPRtp *serverudp.Server - serverUDPRtcp *serverudp.Server - serverTCP *servertcp.Server + serverUDPRTP *gortsplib.ServerUDPListener + serverUDPRTCP *gortsplib.ServerUDPListener + serverTCP *serverplain.Server serverTLS *servertls.Server pathMan *pathman.PathManager clientMan *clientman.ClientManager @@ -146,7 +146,9 @@ func (p *program) createResources(initial bool) error { } if p.logger == nil { - p.logger, err = logger.New(p.conf.LogLevelParsed, p.conf.LogDestinationsParsed, + p.logger, err = logger.New( + p.conf.LogLevelParsed, + p.conf.LogDestinationsParsed, p.conf.LogFile) if err != nil { return err @@ -179,17 +181,21 @@ func (p *program) createResources(initial bool) error { } if _, ok := p.conf.ProtocolsParsed[gortsplib.StreamProtocolUDP]; ok { - if p.serverUDPRtp == nil { - p.serverUDPRtp, err = serverudp.New(p.conf.WriteTimeout, - p.conf.RtpPort, gortsplib.StreamTypeRtp, p) + if p.serverUDPRTP == nil { + p.serverUDPRTP, err = serverudpl.New( + p.conf.RTPPort, + gortsplib.StreamTypeRTP, + p) if err != nil { return err } } - if p.serverUDPRtcp == nil { - p.serverUDPRtcp, err = serverudp.New(p.conf.WriteTimeout, - p.conf.RtcpPort, gortsplib.StreamTypeRtcp, p) + if p.serverUDPRTCP == nil { + p.serverUDPRTCP, err = serverudpl.New( + p.conf.RTCPPort, + gortsplib.StreamTypeRTCP, + p) if err != nil { return err } @@ -198,8 +204,13 @@ func (p *program) createResources(initial bool) error { if p.serverTCP == nil { if p.conf.EncryptionParsed == conf.EncryptionNo || p.conf.EncryptionParsed == conf.EncryptionOptional { - p.serverTCP, err = servertcp.New(p.conf.RtspPort, p.conf.ReadTimeout, - p.conf.WriteTimeout, p) + p.serverTCP, err = serverplain.New( + p.conf.RtspPort, + p.conf.ReadTimeout, + p.conf.WriteTimeout, + p.serverUDPRTP, + p.serverUDPRTCP, + p) if err != nil { return err } @@ -208,8 +219,13 @@ func (p *program) createResources(initial bool) error { if p.serverTLS == nil { if p.conf.EncryptionParsed == conf.EncryptionStrict || p.conf.EncryptionParsed == conf.EncryptionOptional { - p.serverTLS, err = servertls.New(p.conf.RtspsPort, p.conf.ReadTimeout, - p.conf.WriteTimeout, p.conf.ServerKey, p.conf.ServerCert, p) + p.serverTLS, err = servertls.New( + p.conf.RtspsPort, + p.conf.ReadTimeout, + p.conf.WriteTimeout, + p.conf.ServerKey, + p.conf.ServerCert, + p) if err != nil { return err } @@ -217,16 +233,28 @@ func (p *program) createResources(initial bool) error { } if p.pathMan == nil { - p.pathMan = pathman.New(p.conf.RtspPort, p.conf.ReadTimeout, - p.conf.WriteTimeout, p.conf.AuthMethodsParsed, p.conf.Paths, - p.stats, p) + p.pathMan = pathman.New( + p.conf.RtspPort, + p.conf.ReadTimeout, + p.conf.WriteTimeout, + p.conf.AuthMethodsParsed, + p.conf.Paths, + p.stats, + p) } if p.clientMan == nil { - p.clientMan = clientman.New(p.conf.RtspPort, p.conf.ReadTimeout, - p.conf.RunOnConnect, p.conf.RunOnConnectRestart, - p.conf.ProtocolsParsed, p.stats, p.serverUDPRtp, p.serverUDPRtcp, - p.pathMan, p.serverTCP, p.serverTLS, p) + p.clientMan = clientman.New( + p.conf.RtspPort, + p.conf.ReadTimeout, + p.conf.RunOnConnect, + p.conf.RunOnConnectRestart, + p.conf.ProtocolsParsed, + p.stats, + p.pathMan, + p.serverTCP, + p.serverTLS, + p) } return nil @@ -252,20 +280,20 @@ func (p *program) closeResources(newConf *conf.Conf) { closePprof = true } - closeServerUDPRtp := false + closeServerUDPRTP := false if newConf == nil || !reflect.DeepEqual(newConf.ProtocolsParsed, p.conf.ProtocolsParsed) || newConf.WriteTimeout != p.conf.WriteTimeout || - newConf.RtpPort != p.conf.RtpPort { - closeServerUDPRtp = true + newConf.RTPPort != p.conf.RTPPort { + closeServerUDPRTP = true } - closeServerUDPRtcp := false + closeServerUDPRTCP := false if newConf == nil || !reflect.DeepEqual(newConf.ProtocolsParsed, p.conf.ProtocolsParsed) || newConf.WriteTimeout != p.conf.WriteTimeout || - newConf.RtcpPort != p.conf.RtcpPort { - closeServerUDPRtcp = true + newConf.RTCPPort != p.conf.RTCPPort { + closeServerUDPRTCP = true } closeServerTCP := false @@ -273,7 +301,9 @@ func (p *program) closeResources(newConf *conf.Conf) { newConf.EncryptionParsed != p.conf.EncryptionParsed || newConf.RtspPort != p.conf.RtspPort || newConf.ReadTimeout != p.conf.ReadTimeout || - newConf.WriteTimeout != p.conf.WriteTimeout { + newConf.WriteTimeout != p.conf.WriteTimeout || + closeServerUDPRTP || + closeServerUDPRTCP { closeServerTCP = true } @@ -299,8 +329,6 @@ func (p *program) closeResources(newConf *conf.Conf) { closeClientMan := false if newConf == nil || - closeServerUDPRtp || - closeServerUDPRtcp || closeServerTCP || closeServerTLS || closePathMan || @@ -337,14 +365,14 @@ func (p *program) closeResources(newConf *conf.Conf) { p.serverTCP = nil } - if closeServerUDPRtcp && p.serverUDPRtcp != nil { - p.serverUDPRtcp.Close() - p.serverUDPRtcp = nil + if closeServerUDPRTCP && p.serverUDPRTCP != nil { + p.serverUDPRTCP.Close() + p.serverUDPRTCP = nil } - if closeServerUDPRtp && p.serverUDPRtp != nil { - p.serverUDPRtp.Close() - p.serverUDPRtp = nil + if closeServerUDPRTP && p.serverUDPRTP != nil { + p.serverUDPRTP.Close() + p.serverUDPRTP = nil } if closePprof && p.pprof != nil { diff --git a/main_test.go b/main_test.go index 74cf9cee..2a2eb1fd 100644 --- a/main_test.go +++ b/main_test.go @@ -298,7 +298,7 @@ y++U32uuSFiXDcSLarfIsE992MEJLSAynbF1Rsgsr3gXbGiuToJRyxbIeVy7gwzD `) func TestPublishRead(t *testing.T) { - for _, conf := range []struct { + for _, ca := range []struct { encrypted bool publisherSoft string publisherProto string @@ -321,17 +321,17 @@ func TestPublishRead(t *testing.T) { {true, "gstreamer", "tcp", "ffmpeg", "tcp"}, } { encryptedStr := func() string { - if conf.encrypted { + if ca.encrypted { return "encrypted" } return "plain" }() - t.Run(encryptedStr+"_"+conf.publisherSoft+"_"+conf.publisherProto+"_"+ - conf.readerSoft+"_"+conf.readerProto, func(t *testing.T) { + t.Run(encryptedStr+"_"+ca.publisherSoft+"_"+ca.publisherProto+"_"+ + ca.readerSoft+"_"+ca.readerProto, func(t *testing.T) { var proto string var port string - if !conf.encrypted { + if !ca.encrypted { proto = "rtsp" port = "8554" @@ -362,7 +362,7 @@ func TestPublishRead(t *testing.T) { time.Sleep(1 * time.Second) - switch conf.publisherSoft { + switch ca.publisherSoft { case "ffmpeg": cnt1, err := newContainer("ffmpeg", "source", []string{ "-re", @@ -370,7 +370,7 @@ func TestPublishRead(t *testing.T) { "-i", "emptyvideo.ts", "-c", "copy", "-f", "rtsp", - "-rtsp_transport", conf.publisherProto, + "-rtsp_transport", ca.publisherProto, proto + "://" + ownDockerIP + ":" + port + "/teststream", }) require.NoError(t, err) @@ -382,7 +382,7 @@ func TestPublishRead(t *testing.T) { cnt1, err := newContainer("gstreamer", "source", []string{ "filesrc location=emptyvideo.ts ! tsdemux ! video/x-h264 ! rtspclientsink " + "location=" + proto + "://" + ownDockerIP + ":" + port + "/teststream " + - "protocols=" + conf.publisherProto + " tls-validation-flags=0 latency=0 timeout=0 rtx-time=0", + "protocols=" + ca.publisherProto + " tls-validation-flags=0 latency=0 timeout=0 rtx-time=0", }) require.NoError(t, err) defer cnt1.close() @@ -392,10 +392,10 @@ func TestPublishRead(t *testing.T) { time.Sleep(1 * time.Second) - switch conf.readerSoft { + switch ca.readerSoft { case "ffmpeg": cnt2, err := newContainer("ffmpeg", "dest", []string{ - "-rtsp_transport", conf.readerProto, + "-rtsp_transport", ca.readerProto, "-i", proto + "://" + ownDockerIP + ":" + port + "/teststream", "-vframes", "1", "-f", "image2", @@ -416,7 +416,7 @@ func TestPublishRead(t *testing.T) { case "vlc": args := []string{} - if conf.readerProto == "tcp" { + if ca.readerProto == "tcp" { args = append(args, "--rtsp-tcp") } args = append(args, proto+"://"+ownDockerIP+":"+port+"/teststream")