OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

move UDP support into gortsplib

Browse Source
This commit is contained in:
aler9
2021-01-06 12:40:18 +01:00
parent 39882ef654
commit 9fb844331e
12 changed files with 195 additions and 400 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
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-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=
internal/client/client.go
+29 -131
View File
@@ -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.
internal/clientman/clientman.go
+23 -16
View File
@@ -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():
internal/conf/conf.go
+8 -8
View File
@@ -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")
}
internal/servertcp/server.go → internal/serverplain/server.go
+5 -1
View File
@@ -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))
internal/serverudp/server.go
-176
View File
@@ -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)
}
internal/serverudpl/server.go
+34
View File
@@ -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
}
internal/sourcertmp/source.go
+14 -14
View File
@@ -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:
internal/sourcertsp/source.go
+2 -2
View File
@@ -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 {
main.go
+66 -38
View File
@@ -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 {
main_test.go
+11 -11
View File
@@ -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")