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

switch to golangci-lint

Browse Source
This commit is contained in:
aler9
2020-12-05 20:42:59 +01:00
parent b40b2f2463
commit 19440f2486
17 changed files with 367 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/lint.yml
+25
View File
@@ -0,0 +1,25 @@
name: lint
on:
push:
branches: [ master ]
pull_request:
branches: [ master ]
jobs:
lint:
name: lint
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- uses: golangci/golangci-lint-action@v2
with:
version: v1.33
args: >
--disable=errcheck
--enable=gofmt
--enable=golint
--enable=misspell
Makefile
+11 -1
View File
@@ -10,7 +10,8 @@ help:
@echo ""
@echo " mod-tidy run go mod tidy"
@echo " format format source files"
@echo " test run available tests"
@echo " test run tests"
@echo " lint run linters"
@echo " stress NAME=n run stress environment"
@echo " run run app"
@echo " release build release assets"
@@ -53,6 +54,15 @@ test-nodocker:
docker build -q testimages/$(IMG) -t rtsp-simple-server-test-$(IMG)$(NL))
go test -race -v .
lint:
docker run --rm -v $(PWD):/app -w /app \
golangci/golangci-lint:v1.33.0 \
golangci-lint run -v \
--disable=errcheck \
--enable=gofmt \
--enable=golint \
--enable=misspell
stress:
docker build -q . -f stress/$(NAME)/Dockerfile -t temp
docker run --rm -it --network=host temp
README.md
+1 -1
View File
@@ -2,7 +2,7 @@
# rtsp-simple-server
[![Test](https://github.com/aler9/rtsp-simple-server/workflows/test/badge.svg)](https://github.com/aler9/rtsp-simple-server/actions)
[![Go Report Card](https://goreportcard.com/badge/github.com/aler9/rtsp-simple-server)](https://goreportcard.com/report/github.com/aler9/rtsp-simple-server)
[![Lint](https://github.com/aler9/rtsp-simple-server/workflows/lint/badge.svg)](https://github.com/aler9/rtsp-simple-server/actions)
[![Docker Hub](https://img.shields.io/badge/docker-aler9%2Frtsp--simple--server-blue)](https://hub.docker.com/r/aler9/rtsp-simple-server)
_rtsp-simple-server_ is a simple, ready-to-use and zero-dependency RTSP server and RTSP proxy, a software that allows multiple users to publish, read and proxy live video and audio streams over time. RTSP is a standard protocol that describes how to perform these operations with the help of a server, that is contacted by both publishers and readers and relays the publisher's streams to the readers.
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-20201130140921-6b4fe73270ee
github.com/aler9/gortsplib v0.0.0-20201205174405-71861bd1325c
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-20201130140921-6b4fe73270ee h1:CpjcwYjyiHM/RT1wV1eYwjRCrA1O0EOuyiQk6DTocdQ=
github.com/aler9/gortsplib v0.0.0-20201130140921-6b4fe73270ee/go.mod h1:8P09VjpiPJFyfkVosyF5/TY82jNwkMN165NS/7sc32I=
github.com/aler9/gortsplib v0.0.0-20201205174405-71861bd1325c h1:LMCNoh4s3NnExhrZRFtrOyrUzaCLBTgg68uDBr76bQo=
github.com/aler9/gortsplib v0.0.0-20201205174405-71861bd1325c/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
+202 -207
View File
@@ -26,7 +26,7 @@ import (
const (
checkStreamInterval = 5 * time.Second
receiverReportInterval = 10 * time.Second
sessionId = "12345678"
sessionID = "12345678"
)
type readReq struct {
@@ -104,8 +104,8 @@ type Client struct {
protocols map[gortsplib.StreamProtocol]struct{}
wg *sync.WaitGroup
stats *stats.Stats
serverUdpRtp *serverudp.Server
serverUdpRtcp *serverudp.Server
serverUDPRtp *serverudp.Server
serverUDPRtcp *serverudp.Server
conn *gortsplib.ConnServer
parent Parent
@@ -120,7 +120,7 @@ type Client struct {
rtcpReceivers map[int]*rtcpreceiver.RtcpReceiver
udpLastFrameTimes []*int64
describeCSeq base.HeaderValue
describeUrl string
describeURL string
tcpWriteMutex sync.Mutex
tcpWriteOk bool
@@ -139,8 +139,8 @@ func New(
protocols map[gortsplib.StreamProtocol]struct{},
wg *sync.WaitGroup,
stats *stats.Stats,
serverUdpRtp *serverudp.Server,
serverUdpRtcp *serverudp.Server,
serverUDPRtp *serverudp.Server,
serverUDPRtcp *serverudp.Server,
nconn net.Conn,
parent Parent) *Client {
@@ -152,8 +152,8 @@ func New(
protocols: protocols,
wg: wg,
stats: stats,
serverUdpRtp: serverUdpRtp,
serverUdpRtcp: serverUdpRtcp,
serverUDPRtp: serverUDPRtp,
serverUDPRtcp: serverUDPRtcp,
conn: gortsplib.NewConnServer(gortsplib.ConnServerConf{
Conn: nconn,
ReadTimeout: readTimeout,
@@ -181,7 +181,7 @@ func (c *Client) Close() {
close(c.terminate)
}
// IsSource implementes path.source.
// IsSource implements path.source.
func (c *Client) IsSource() {}
func (c *Client) log(format string, args ...interface{}) {
@@ -272,14 +272,14 @@ func (c *Client) Authenticate(authMethods []headers.AuthMethod, ips []interface{
err := c.authHelper.ValidateHeader(req.Header["Authorization"], req.Method, req.URL)
if err != nil {
c.authFailures += 1
c.authFailures++
// vlc with login prompt sends 4 requests:
// 1) without credentials
// 2) with password but without username
// 3) without credentials
// 4) with password and username
// hence we must allow up to 3 failures
// therefore we must allow up to 3 failures
if c.authFailures > 3 {
c.log("ERR: unauthorized: %s", err)
@@ -290,20 +290,19 @@ func (c *Client) Authenticate(authMethods []headers.AuthMethod, ips []interface{
"WWW-Authenticate": c.authHelper.GenerateHeader(),
},
}}
} else {
if c.authFailures > 1 {
c.log("WARN: unauthorized: %s", err)
}
return errAuthNotCritical{&base.Response{
StatusCode: base.StatusUnauthorized,
Header: base.Header{
"CSeq": req.Header["CSeq"],
"WWW-Authenticate": c.authHelper.GenerateHeader(),
},
}}
}
if c.authFailures > 1 {
c.log("WARN: unauthorized: %s", err)
}
return errAuthNotCritical{&base.Response{
StatusCode: base.StatusUnauthorized,
Header: base.Header{
"CSeq": req.Header["CSeq"],
"WWW-Authenticate": c.authHelper.GenerateHeader(),
},
}}
}
}
@@ -347,27 +346,27 @@ func (c *Client) handleRequest(req *base.Request) error {
}
switch req.Method {
case base.OPTIONS:
case base.Options:
c.conn.WriteResponse(&base.Response{
StatusCode: base.StatusOK,
Header: base.Header{
"CSeq": cseq,
"Public": base.HeaderValue{strings.Join([]string{
string(base.GET_PARAMETER),
string(base.DESCRIBE),
string(base.ANNOUNCE),
string(base.SETUP),
string(base.PLAY),
string(base.RECORD),
string(base.PAUSE),
string(base.TEARDOWN),
string(base.GetParameter),
string(base.Describe),
string(base.Announce),
string(base.Setup),
string(base.Play),
string(base.Record),
string(base.Pause),
string(base.Teardown),
}, ", ")},
},
})
return nil
// GET_PARAMETER is used like a ping
case base.GET_PARAMETER:
case base.GetParameter:
c.conn.WriteResponse(&base.Response{
StatusCode: base.StatusOK,
Header: base.Header{
@@ -378,7 +377,7 @@ func (c *Client) handleRequest(req *base.Request) error {
})
return nil
case base.DESCRIBE:
case base.Describe:
err := c.checkState(map[state]struct{}{
stateInitial: {},
})
@@ -417,11 +416,11 @@ func (c *Client) handleRequest(req *base.Request) error {
c.path = path
c.state = stateWaitingDescribe
c.describeCSeq = cseq
c.describeUrl = req.URL.String()
c.describeURL = req.URL.String()
return errStateWaitingDescribe
case base.ANNOUNCE:
case base.Announce:
err := c.checkState(map[state]struct{}{
stateInitial: {},
})
@@ -458,10 +457,10 @@ func (c *Client) handleRequest(req *base.Request) error {
return errStateTerminate
}
for trackId, t := range tracks {
for trackID, t := range tracks {
_, err := t.ClockRate()
if err != nil {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("unable to get clock rate of track %d", trackId))
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("unable to get clock rate of track %d", trackID))
return errStateTerminate
}
}
@@ -483,9 +482,9 @@ func (c *Client) handleRequest(req *base.Request) error {
}
}
for trackId, t := range tracks {
for trackID, t := range tracks {
clockRate, _ := t.ClockRate()
c.rtcpReceivers[trackId] = rtcpreceiver.New(nil, clockRate)
c.rtcpReceivers[trackID] = rtcpreceiver.New(nil, clockRate)
}
c.path = path
@@ -499,7 +498,7 @@ func (c *Client) handleRequest(req *base.Request) error {
})
return nil
case base.SETUP:
case base.Setup:
th, err := headers.ReadTransport(req.Header["Transport"])
if err != nil {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("transport header: %s", err))
@@ -540,10 +539,10 @@ func (c *Client) handleRequest(req *base.Request) error {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("invalid track id (%s)", controlPath))
return errStateTerminate
}
trackId := int(tmp)
trackID := int(tmp)
if _, ok := c.streamTracks[trackId]; ok {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("track %d has already been setup", trackId))
if _, ok := c.streamTracks[trackID]; ok {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("track %d has already been setup", trackID))
return errStateTerminate
}
@@ -564,7 +563,7 @@ func (c *Client) handleRequest(req *base.Request) error {
return errStateTerminate
}
path, err := c.parent.OnClientSetupPlay(c, basePath, trackId, req)
path, err := c.parent.OnClientSetupPlay(c, basePath, trackID, req)
if err != nil {
switch terr := err.(type) {
case errAuthNotCritical:
@@ -585,7 +584,7 @@ func (c *Client) handleRequest(req *base.Request) error {
c.state = statePrePlay
c.streamProtocol = gortsplib.StreamProtocolUDP
c.streamTracks[trackId] = &streamTrack{
c.streamTracks[trackID] = &streamTrack{
rtpPort: (*th.ClientPorts)[0],
rtcpPort: (*th.ClientPorts)[1],
}
@@ -597,7 +596,7 @@ func (c *Client) handleRequest(req *base.Request) error {
return &v
}(),
ClientPorts: th.ClientPorts,
ServerPorts: &[2]int{c.serverUdpRtp.Port(), c.serverUdpRtcp.Port()},
ServerPorts: &[2]int{c.serverUDPRtp.Port(), c.serverUDPRtcp.Port()},
}
c.conn.WriteResponse(&base.Response{
@@ -605,67 +604,66 @@ func (c *Client) handleRequest(req *base.Request) error {
Header: base.Header{
"CSeq": cseq,
"Transport": th.Write(),
"Session": base.HeaderValue{sessionId},
},
})
return nil
// play with TCP
} else {
if _, ok := c.protocols[gortsplib.StreamProtocolTCP]; !ok {
c.writeResError(cseq, base.StatusUnsupportedTransport, fmt.Errorf("TCP streaming is disabled"))
return nil
}
if len(c.streamTracks) > 0 && c.streamProtocol != gortsplib.StreamProtocolTCP {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("can't receive tracks with different protocols"))
return errStateTerminate
}
path, err := c.parent.OnClientSetupPlay(c, basePath, trackId, req)
if err != nil {
switch terr := err.(type) {
case errAuthNotCritical:
c.conn.WriteResponse(terr.Response)
return nil
case errAuthCritical:
c.conn.WriteResponse(terr.Response)
return errStateTerminate
default:
c.writeResError(cseq, base.StatusBadRequest, err)
return errStateTerminate
}
}
c.path = path
c.state = statePrePlay
c.streamProtocol = gortsplib.StreamProtocolTCP
c.streamTracks[trackId] = &streamTrack{
rtpPort: 0,
rtcpPort: 0,
}
interleavedIds := [2]int{trackId * 2, (trackId * 2) + 1}
th := &headers.Transport{
Protocol: gortsplib.StreamProtocolTCP,
InterleavedIds: &interleavedIds,
}
c.conn.WriteResponse(&base.Response{
StatusCode: base.StatusOK,
Header: base.Header{
"CSeq": cseq,
"Transport": th.Write(),
"Session": base.HeaderValue{sessionId},
"Session": base.HeaderValue{sessionID},
},
})
return nil
}
// play with TCP
if _, ok := c.protocols[gortsplib.StreamProtocolTCP]; !ok {
c.writeResError(cseq, base.StatusUnsupportedTransport, fmt.Errorf("TCP streaming is disabled"))
return nil
}
if len(c.streamTracks) > 0 && c.streamProtocol != gortsplib.StreamProtocolTCP {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("can't receive tracks with different protocols"))
return errStateTerminate
}
path, err := c.parent.OnClientSetupPlay(c, basePath, trackID, req)
if err != nil {
switch terr := err.(type) {
case errAuthNotCritical:
c.conn.WriteResponse(terr.Response)
return nil
case errAuthCritical:
c.conn.WriteResponse(terr.Response)
return errStateTerminate
default:
c.writeResError(cseq, base.StatusBadRequest, err)
return errStateTerminate
}
}
c.path = path
c.state = statePrePlay
c.streamProtocol = gortsplib.StreamProtocolTCP
c.streamTracks[trackID] = &streamTrack{
rtpPort: 0,
rtcpPort: 0,
}
interleavedIds := [2]int{trackID * 2, (trackID * 2) + 1}
th := &headers.Transport{
Protocol: gortsplib.StreamProtocolTCP,
InterleavedIds: &interleavedIds,
}
c.conn.WriteResponse(&base.Response{
StatusCode: base.StatusOK,
Header: base.Header{
"CSeq": cseq,
"Transport": th.Write(),
"Session": base.HeaderValue{sessionID},
},
})
return nil
// record
case statePreRecord:
if th.Mode == nil || *th.Mode != headers.TransportModeRecord {
@@ -702,8 +700,8 @@ func (c *Client) handleRequest(req *base.Request) error {
}
c.streamProtocol = gortsplib.StreamProtocolUDP
trackId := len(c.streamTracks)
c.streamTracks[trackId] = &streamTrack{
trackID := len(c.streamTracks)
c.streamTracks[trackID] = &streamTrack{
rtpPort: (*th.ClientPorts)[0],
rtcpPort: (*th.ClientPorts)[1],
}
@@ -715,7 +713,7 @@ func (c *Client) handleRequest(req *base.Request) error {
return &v
}(),
ClientPorts: th.ClientPorts,
ServerPorts: &[2]int{c.serverUdpRtp.Port(), c.serverUdpRtcp.Port()},
ServerPorts: &[2]int{c.serverUDPRtp.Port(), c.serverUDPRtcp.Port()},
}
c.conn.WriteResponse(&base.Response{
@@ -723,69 +721,68 @@ func (c *Client) handleRequest(req *base.Request) error {
Header: base.Header{
"CSeq": cseq,
"Transport": th.Write(),
"Session": base.HeaderValue{sessionId},
},
})
return nil
// record with TCP
} else {
if _, ok := c.protocols[gortsplib.StreamProtocolTCP]; !ok {
c.writeResError(cseq, base.StatusUnsupportedTransport, fmt.Errorf("TCP streaming is disabled"))
return nil
}
if len(c.streamTracks) > 0 && c.streamProtocol != gortsplib.StreamProtocolTCP {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("can't publish tracks with different protocols"))
return errStateTerminate
}
interleavedIds := [2]int{len(c.streamTracks) * 2, 1 + len(c.streamTracks)*2}
if th.InterleavedIds == nil {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("transport header does not contain the interleaved field"))
return errStateTerminate
}
if (*th.InterleavedIds)[0] != interleavedIds[0] || (*th.InterleavedIds)[1] != interleavedIds[1] {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("wrong interleaved ids, expected %v, got %v", interleavedIds, *th.InterleavedIds))
return errStateTerminate
}
if len(c.streamTracks) >= c.path.SourceTrackCount() {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("all the tracks have already been setup"))
return errStateTerminate
}
c.streamProtocol = gortsplib.StreamProtocolTCP
trackId := len(c.streamTracks)
c.streamTracks[trackId] = &streamTrack{
rtpPort: 0,
rtcpPort: 0,
}
ht := &headers.Transport{
Protocol: gortsplib.StreamProtocolTCP,
InterleavedIds: &interleavedIds,
}
c.conn.WriteResponse(&base.Response{
StatusCode: base.StatusOK,
Header: base.Header{
"CSeq": cseq,
"Transport": ht.Write(),
"Session": base.HeaderValue{sessionId},
"Session": base.HeaderValue{sessionID},
},
})
return nil
}
// record with TCP
if _, ok := c.protocols[gortsplib.StreamProtocolTCP]; !ok {
c.writeResError(cseq, base.StatusUnsupportedTransport, fmt.Errorf("TCP streaming is disabled"))
return nil
}
if len(c.streamTracks) > 0 && c.streamProtocol != gortsplib.StreamProtocolTCP {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("can't publish tracks with different protocols"))
return errStateTerminate
}
interleavedIds := [2]int{len(c.streamTracks) * 2, 1 + len(c.streamTracks)*2}
if th.InterleavedIds == nil {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("transport header does not contain the interleaved field"))
return errStateTerminate
}
if (*th.InterleavedIds)[0] != interleavedIds[0] || (*th.InterleavedIds)[1] != interleavedIds[1] {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("wrong interleaved ids, expected %v, got %v", interleavedIds, *th.InterleavedIds))
return errStateTerminate
}
if len(c.streamTracks) >= c.path.SourceTrackCount() {
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("all the tracks have already been setup"))
return errStateTerminate
}
c.streamProtocol = gortsplib.StreamProtocolTCP
trackID := len(c.streamTracks)
c.streamTracks[trackID] = &streamTrack{
rtpPort: 0,
rtcpPort: 0,
}
ht := &headers.Transport{
Protocol: gortsplib.StreamProtocolTCP,
InterleavedIds: &interleavedIds,
}
c.conn.WriteResponse(&base.Response{
StatusCode: base.StatusOK,
Header: base.Header{
"CSeq": cseq,
"Transport": ht.Write(),
"Session": base.HeaderValue{sessionID},
},
})
return nil
default:
c.writeResError(cseq, base.StatusBadRequest, fmt.Errorf("client is in state '%s'", c.state))
return errStateTerminate
}
case base.PLAY:
case base.Play:
// play can be sent twice, allow calling it even if we're already playing
err := c.checkState(map[state]struct{}{
statePrePlay: {},
@@ -824,7 +821,7 @@ func (c *Client) handleRequest(req *base.Request) error {
StatusCode: base.StatusOK,
Header: base.Header{
"CSeq": cseq,
"Session": base.HeaderValue{sessionId},
"Session": base.HeaderValue{sessionID},
},
})
@@ -833,7 +830,7 @@ func (c *Client) handleRequest(req *base.Request) error {
}
return nil
case base.RECORD:
case base.Record:
err := c.checkState(map[state]struct{}{
statePreRecord: {},
})
@@ -865,12 +862,12 @@ func (c *Client) handleRequest(req *base.Request) error {
StatusCode: base.StatusOK,
Header: base.Header{
"CSeq": cseq,
"Session": base.HeaderValue{sessionId},
"Session": base.HeaderValue{sessionID},
},
})
return errStateRecord
case base.PAUSE:
case base.Pause:
err := c.checkState(map[state]struct{}{
statePrePlay: {},
statePlay: {},
@@ -886,7 +883,7 @@ func (c *Client) handleRequest(req *base.Request) error {
StatusCode: base.StatusOK,
Header: base.Header{
"CSeq": cseq,
"Session": base.HeaderValue{sessionId},
"Session": base.HeaderValue{sessionID},
},
})
@@ -895,7 +892,7 @@ func (c *Client) handleRequest(req *base.Request) error {
}
return nil
case base.TEARDOWN:
case base.Teardown:
// close connection silently
return errStateTerminate
@@ -983,7 +980,7 @@ func (c *Client) runWaitingDescribe() bool {
StatusCode: base.StatusOK,
Header: base.Header{
"CSeq": c.describeCSeq,
"Content-Base": base.HeaderValue{c.describeUrl + "/"},
"Content-Base": base.HeaderValue{c.describeURL + "/"},
"Content-Type": base.HeaderValue{"application/sdp"},
},
Content: res.sdp,
@@ -1033,9 +1030,8 @@ func (c *Client) runPlay() bool {
if c.streamProtocol == gortsplib.StreamProtocolUDP {
return c.runPlayUDP()
} else {
return c.runPlayTCP()
}
return c.runPlayTCP()
}
func (c *Client) runPlayUDP() bool {
@@ -1209,19 +1205,19 @@ func (c *Client) runRecord() bool {
if c.streamProtocol == gortsplib.StreamProtocolUDP {
c.udpLastFrameTimes = make([]*int64, len(c.streamTracks))
for trackId := range c.streamTracks {
for trackID := range c.streamTracks {
v := time.Now().Unix()
c.udpLastFrameTimes[trackId] = &v
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)
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(
c.serverUDPRtp.Write(
[]byte{0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00},
&net.UDPAddr{
IP: c.ip(),
@@ -1229,7 +1225,7 @@ func (c *Client) runRecord() bool {
Port: track.rtpPort,
})
c.serverUdpRtcp.Write(
c.serverUDPRtcp.Write(
[]byte{0x80, 0xc9, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00},
&net.UDPAddr{
IP: c.ip(),
@@ -1249,9 +1245,8 @@ func (c *Client) runRecord() bool {
if c.streamProtocol == gortsplib.StreamProtocolUDP {
return c.runRecordUDP()
} else {
return c.runRecordTCP()
}
return c.runRecordTCP()
}
func (c *Client) runRecordUDP() bool {
@@ -1286,8 +1281,8 @@ func (c *Client) runRecordUDP() bool {
onError := func(err error) bool {
if err == errStateInitial {
for _, track := range c.streamTracks {
c.serverUdpRtp.RemovePublisher(c.ip(), track.rtpPort, c)
c.serverUdpRtcp.RemovePublisher(c.ip(), track.rtcpPort, c)
c.serverUDPRtp.RemovePublisher(c.ip(), track.rtpPort, c)
c.serverUDPRtcp.RemovePublisher(c.ip(), track.rtcpPort, c)
}
c.state = statePreRecord
@@ -1301,8 +1296,8 @@ func (c *Client) runRecordUDP() bool {
}
for _, track := range c.streamTracks {
c.serverUdpRtp.RemovePublisher(c.ip(), track.rtpPort, c)
c.serverUdpRtcp.RemovePublisher(c.ip(), track.rtcpPort, c)
c.serverUDPRtp.RemovePublisher(c.ip(), track.rtpPort, c)
c.serverUDPRtcp.RemovePublisher(c.ip(), track.rtcpPort, c)
}
c.path.OnClientRemove(c)
@@ -1345,8 +1340,8 @@ func (c *Client) runRecordUDP() bool {
<-readerDone
for _, track := range c.streamTracks {
c.serverUdpRtp.RemovePublisher(c.ip(), track.rtpPort, c)
c.serverUdpRtcp.RemovePublisher(c.ip(), track.rtcpPort, c)
c.serverUDPRtp.RemovePublisher(c.ip(), track.rtpPort, c)
c.serverUDPRtcp.RemovePublisher(c.ip(), track.rtcpPort, c)
}
c.path.OnClientRemove(c)
@@ -1360,12 +1355,12 @@ func (c *Client) runRecordUDP() bool {
case <-receiverReportTicker.C:
now := time.Now()
for trackId := range c.streamTracks {
r := c.rtcpReceivers[trackId].Report(now)
c.serverUdpRtcp.Write(r, &net.UDPAddr{
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,
Port: c.streamTracks[trackID].rtcpPort,
})
}
@@ -1380,8 +1375,8 @@ func (c *Client) runRecordUDP() bool {
<-readerDone
for _, track := range c.streamTracks {
c.serverUdpRtp.RemovePublisher(c.ip(), track.rtpPort, c)
c.serverUdpRtcp.RemovePublisher(c.ip(), track.rtcpPort, c)
c.serverUDPRtp.RemovePublisher(c.ip(), track.rtpPort, c)
c.serverUDPRtcp.RemovePublisher(c.ip(), track.rtcpPort, c)
}
c.path.OnClientRemove(c)
@@ -1406,13 +1401,13 @@ func (c *Client) runRecordTCP() bool {
switch recvt := recv.(type) {
case *base.InterleavedFrame:
if recvt.TrackId >= len(c.streamTracks) {
readerDone <- fmt.Errorf("invalid track id '%d'", recvt.TrackId)
if recvt.TrackID >= len(c.streamTracks) {
readerDone <- fmt.Errorf("invalid track id '%d'", recvt.TrackID)
return
}
c.rtcpReceivers[recvt.TrackId].ProcessFrame(time.Now(), recvt.StreamType, recvt.Content)
c.path.OnFrame(recvt.TrackId, recvt.StreamType, recvt.Content)
c.rtcpReceivers[recvt.TrackID].ProcessFrame(time.Now(), recvt.StreamType, recvt.Content)
c.path.OnFrame(recvt.TrackID, recvt.StreamType, recvt.Content)
case *base.Request:
okc := make(chan bool)
@@ -1465,9 +1460,9 @@ func (c *Client) runRecordTCP() bool {
case <-receiverReportTicker.C:
now := time.Now()
for trackId := range c.streamTracks {
r := c.rtcpReceivers[trackId].Report(now)
c.conn.WriteFrameTCP(trackId, gortsplib.StreamTypeRtcp, r)
for trackID := range c.streamTracks {
r := c.rtcpReceivers[trackID].Report(now)
c.conn.WriteFrameTCP(trackID, gortsplib.StreamTypeRtcp, r)
}
case <-c.terminate:
@@ -1487,31 +1482,31 @@ func (c *Client) runRecordTCP() bool {
}
}
// OnUdpPublisherFrame implements serverudp.Publisher.
func (c *Client) OnUdpPublisherFrame(trackId int, streamType base.StreamType, buf []byte) {
// 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)
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]
func (c *Client) OnReaderFrame(trackID int, streamType base.StreamType, buf []byte) {
track, ok := c.streamTracks[trackID]
if !ok {
return
}
if c.streamProtocol == gortsplib.StreamProtocolUDP {
if streamType == gortsplib.StreamTypeRtp {
c.serverUdpRtp.Write(buf, &net.UDPAddr{
c.serverUDPRtp.Write(buf, &net.UDPAddr{
IP: c.ip(),
Zone: c.zone(),
Port: track.rtpPort,
})
} else {
c.serverUdpRtcp.Write(buf, &net.UDPAddr{
c.serverUDPRtcp.Write(buf, &net.UDPAddr{
IP: c.ip(),
Zone: c.zone(),
Port: track.rtcpPort,
@@ -1521,7 +1516,7 @@ func (c *Client) OnReaderFrame(trackId int, streamType base.StreamType, buf []by
} else {
c.tcpWriteMutex.Lock()
if c.tcpWriteOk {
c.conn.WriteFrameTCP(trackId, streamType, buf)
c.conn.WriteFrameTCP(trackID, streamType, buf)
}
c.tcpWriteMutex.Unlock()
}
internal/clientman/clientman.go
+14 -17
View File
@@ -28,10 +28,10 @@ type ClientManager struct {
runOnConnectRestart bool
protocols map[base.StreamProtocol]struct{}
stats *stats.Stats
serverUdpRtp *serverudp.Server
serverUdpRtcp *serverudp.Server
serverUDPRtp *serverudp.Server
serverUDPRtcp *serverudp.Server
pathMan *pathman.PathManager
serverTcp *servertcp.Server
serverTCP *servertcp.Server
parent Parent
clients map[*client.Client]struct{}
@@ -54,10 +54,10 @@ func New(
runOnConnectRestart bool,
protocols map[base.StreamProtocol]struct{},
stats *stats.Stats,
serverUdpRtp *serverudp.Server,
serverUdpRtcp *serverudp.Server,
serverUDPRtp *serverudp.Server,
serverUDPRtcp *serverudp.Server,
pathMan *pathman.PathManager,
serverTcp *servertcp.Server,
serverTCP *servertcp.Server,
parent Parent) *ClientManager {
cm := &ClientManager{
@@ -68,10 +68,10 @@ func New(
runOnConnectRestart: runOnConnectRestart,
protocols: protocols,
stats: stats,
serverUdpRtp: serverUdpRtp,
serverUdpRtcp: serverUdpRtcp,
serverUDPRtp: serverUDPRtp,
serverUDPRtcp: serverUDPRtcp,
pathMan: pathMan,
serverTcp: serverTcp,
serverTCP: serverTCP,
parent: parent,
clients: make(map[*client.Client]struct{}),
clientClose: make(chan *client.Client),
@@ -100,10 +100,10 @@ func (cm *ClientManager) run() {
outer:
for {
select {
case conn := <-cm.serverTcp.Accept():
case conn := <-cm.serverTCP.Accept():
c := client.New(cm.rtspPort, cm.readTimeout, cm.writeTimeout,
cm.runOnConnect, cm.runOnConnectRestart, cm.protocols, &cm.wg,
cm.stats, cm.serverUdpRtp, cm.serverUdpRtcp, conn, cm)
cm.stats, cm.serverUDPRtp, cm.serverUDPRtcp, conn, cm)
cm.clients[c] = struct{}{}
case c := <-cm.pathMan.ClientClose():
@@ -126,10 +126,7 @@ outer:
}
go func() {
for {
select {
case <-cm.clientClose:
}
for range cm.clientClose {
}
}()
@@ -157,6 +154,6 @@ func (cm *ClientManager) OnClientAnnounce(c *client.Client, pathName string, tra
}
// OnClientSetupPlay is called by client.Client.
func (cm *ClientManager) OnClientSetupPlay(c *client.Client, pathName string, trackId int, req *base.Request) (client.Path, error) {
return cm.pathMan.OnClientSetupPlay(c, pathName, trackId, req)
func (cm *ClientManager) OnClientSetupPlay(c *client.Client, pathName string, trackID int, req *base.Request) (client.Path, error) {
return cm.pathMan.OnClientSetupPlay(c, pathName, trackID, req)
}
internal/conf/pathconf.go
+5 -5
View File
@@ -12,7 +12,7 @@ import (
"github.com/aler9/gortsplib/pkg/base"
)
var reUserPass = regexp.MustCompile("^[a-zA-Z0-9!\\$\\(\\)\\*\\+\\.;<=>\\[\\]\\^_\\-\\{\\}]+$")
var reUserPass = regexp.MustCompile(`^[a-zA-Z0-9!\$\(\)\*\+\.;<=>\[\]\^_\-\{\}]+$`)
const userPassSupportedChars = "A-Z,0-9,!,$,(,),*,+,.,;,<,=,>,[,],^,_,-,{,}"
@@ -89,7 +89,7 @@ func (pconf *PathConf) fillAndCheck(name string) error {
user := u.User.Username()
if user != "" && pass == "" ||
user == "" && pass != "" {
fmt.Errorf("username and password must be both provided")
return fmt.Errorf("username and password must be both provided")
}
}
@@ -130,7 +130,7 @@ func (pconf *PathConf) fillAndCheck(name string) error {
user := u.User.Username()
if user != "" && pass == "" ||
user == "" && pass != "" {
fmt.Errorf("username and password must be both provided")
return fmt.Errorf("username and password must be both provided")
}
}
@@ -177,7 +177,7 @@ func (pconf *PathConf) fillAndCheck(name string) error {
if len(pconf.PublishIps) > 0 {
var err error
pconf.PublishIpsParsed, err = parseIpCidrList(pconf.PublishIps)
pconf.PublishIpsParsed, err = parseIPCidrList(pconf.PublishIps)
if err != nil {
return err
}
@@ -205,7 +205,7 @@ func (pconf *PathConf) fillAndCheck(name string) error {
if len(pconf.ReadIps) > 0 {
var err error
pconf.ReadIpsParsed, err = parseIpCidrList(pconf.ReadIps)
pconf.ReadIpsParsed, err = parseIPCidrList(pconf.ReadIps)
if err != nil {
return err
}
internal/conf/utils.go
+2 -2
View File
@@ -6,7 +6,7 @@ import (
"regexp"
)
var rePathName = regexp.MustCompile("^[0-9a-zA-Z_\\-/]+$")
var rePathName = regexp.MustCompile(`^[0-9a-zA-Z_\-/]+$`)
// CheckPathName check if a path name is valid.
func CheckPathName(name string) error {
@@ -29,7 +29,7 @@ func CheckPathName(name string) error {
return nil
}
func parseIpCidrList(in []string) ([]interface{}, error) {
func parseIPCidrList(in []string) ([]interface{}, error) {
if len(in) == 0 {
return nil, nil
}
internal/path/path.go
+7 -7
View File
@@ -94,7 +94,7 @@ type ClientSetupPlayReq struct {
Res chan ClientSetupPlayRes
Client *client.Client
PathName string
TrackId int
TrackID int
Req *base.Request
}
@@ -310,7 +310,7 @@ outer:
pa.onClientDescribe(req.Client)
case req := <-pa.clientSetupPlay:
err := pa.onClientSetupPlay(req.Client, req.TrackId)
err := pa.onClientSetupPlay(req.Client, req.TrackID)
if err != nil {
req.Res <- ClientSetupPlayRes{nil, err}
continue
@@ -655,13 +655,13 @@ func (pa *Path) onClientDescribe(c *client.Client) {
}
}
func (pa *Path) onClientSetupPlay(c *client.Client, trackId int) error {
func (pa *Path) onClientSetupPlay(c *client.Client, trackID int) error {
if pa.sourceState != sourceStateReady {
return fmt.Errorf("no one is publishing to path '%s'", pa.name)
}
if trackId >= pa.sourceTrackCount {
return fmt.Errorf("track %d does not exist", trackId)
if trackID >= pa.sourceTrackCount {
return fmt.Errorf("track %d does not exist", trackID)
}
if _, ok := pa.clients[c]; !ok {
@@ -873,6 +873,6 @@ func (pa *Path) OnClientPause(c *client.Client) {
}
// OnFrame is called by a source or by a client.Client.
func (pa *Path) OnFrame(trackId int, streamType gortsplib.StreamType, buf []byte) {
pa.readers.forwardFrame(trackId, streamType, buf)
func (pa *Path) OnFrame(trackID int, streamType gortsplib.StreamType, buf []byte) {
pa.readers.forwardFrame(trackID, streamType, buf)
}
internal/path/readersmap.go
+2 -2
View File
@@ -36,11 +36,11 @@ func (m *readersMap) remove(reader reader) {
delete(m.ma, reader)
}
func (m *readersMap) forwardFrame(trackId int, streamType gortsplib.StreamType, buf []byte) {
func (m *readersMap) forwardFrame(trackID int, streamType gortsplib.StreamType, buf []byte) {
m.mutex.RLock()
defer m.mutex.RUnlock()
for c := range m.ma {
c.OnReaderFrame(trackId, streamType, buf)
c.OnReaderFrame(trackID, streamType, buf)
}
}
internal/pathman/pathman.go
+14 -14
View File
@@ -146,14 +146,14 @@ outer:
case req := <-pm.clientDescribe:
pathName, pathConf, err := pm.findPathConf(req.PathName)
if err != nil {
req.Res <- path.ClientDescribeRes{nil, err}
req.Res <- path.ClientDescribeRes{nil, err} //nolint:govet
continue
}
err = req.Client.Authenticate(pm.authMethods, pathConf.ReadIpsParsed,
pathConf.ReadUser, pathConf.ReadPass, req.Req)
if err != nil {
req.Res <- path.ClientDescribeRes{nil, err}
req.Res <- path.ClientDescribeRes{nil, err} //nolint:govet
continue
}
@@ -169,14 +169,14 @@ outer:
case req := <-pm.clientAnnounce:
pathName, pathConf, err := pm.findPathConf(req.PathName)
if err != nil {
req.Res <- path.ClientAnnounceRes{nil, err}
req.Res <- path.ClientAnnounceRes{nil, err} //nolint:govet
continue
}
err = req.Client.Authenticate(pm.authMethods,
pathConf.PublishIpsParsed, pathConf.PublishUser, pathConf.PublishPass, req.Req)
if err != nil {
req.Res <- path.ClientAnnounceRes{nil, err}
req.Res <- path.ClientAnnounceRes{nil, err} //nolint:govet
continue
}
@@ -191,20 +191,20 @@ outer:
case req := <-pm.clientSetupPlay:
if _, ok := pm.paths[req.PathName]; !ok {
req.Res <- path.ClientSetupPlayRes{nil, fmt.Errorf("no one is publishing to path '%s'", req.PathName)}
req.Res <- path.ClientSetupPlayRes{nil, fmt.Errorf("no one is publishing to path '%s'", req.PathName)} //nolint:govet
continue
}
_, pathConf, err := pm.findPathConf(req.PathName)
if err != nil {
req.Res <- path.ClientSetupPlayRes{nil, err}
req.Res <- path.ClientSetupPlayRes{nil, err} //nolint:govet
continue
}
err = req.Client.Authenticate(pm.authMethods,
pathConf.ReadIpsParsed, pathConf.ReadUser, pathConf.ReadPass, req.Req)
if err != nil {
req.Res <- path.ClientSetupPlayRes{nil, err}
req.Res <- path.ClientSetupPlayRes{nil, err} //nolint:govet
continue
}
@@ -229,13 +229,13 @@ outer:
}
case req := <-pm.clientDescribe:
req.Res <- path.ClientDescribeRes{nil, fmt.Errorf("terminated")}
req.Res <- path.ClientDescribeRes{nil, fmt.Errorf("terminated")} //nolint:govet
case req := <-pm.clientAnnounce:
req.Res <- path.ClientAnnounceRes{nil, fmt.Errorf("terminated")}
req.Res <- path.ClientAnnounceRes{nil, fmt.Errorf("terminated")} //nolint:govet
case req := <-pm.clientSetupPlay:
req.Res <- path.ClientSetupPlayRes{nil, fmt.Errorf("terminated")}
req.Res <- path.ClientSetupPlayRes{nil, fmt.Errorf("terminated")} //nolint:govet
}
}
}()
@@ -302,7 +302,7 @@ func (pm *PathManager) OnPathClientClose(c *client.Client) {
// OnClientDescribe is called by client.Client.
func (pm *PathManager) OnClientDescribe(c *client.Client, pathName string, req *base.Request) (client.Path, error) {
res := make(chan path.ClientDescribeRes)
pm.clientDescribe <- path.ClientDescribeReq{res, c, pathName, req}
pm.clientDescribe <- path.ClientDescribeReq{res, c, pathName, req} //nolint:govet
re := <-res
return re.Path, re.Err
}
@@ -310,15 +310,15 @@ func (pm *PathManager) OnClientDescribe(c *client.Client, pathName string, req *
// OnClientAnnounce is called by client.Client.
func (pm *PathManager) OnClientAnnounce(c *client.Client, pathName string, tracks gortsplib.Tracks, req *base.Request) (client.Path, error) {
res := make(chan path.ClientAnnounceRes)
pm.clientAnnounce <- path.ClientAnnounceReq{res, c, pathName, tracks, req}
pm.clientAnnounce <- path.ClientAnnounceReq{res, c, pathName, tracks, req} //nolint:govet
re := <-res
return re.Path, re.Err
}
// OnClientSetupPlay is called by client.Client.
func (pm *PathManager) OnClientSetupPlay(c *client.Client, pathName string, trackId int, req *base.Request) (client.Path, error) {
func (pm *PathManager) OnClientSetupPlay(c *client.Client, pathName string, trackID int, req *base.Request) (client.Path, error) {
res := make(chan path.ClientSetupPlayRes)
pm.clientSetupPlay <- path.ClientSetupPlayReq{res, c, pathName, trackId, req}
pm.clientSetupPlay <- path.ClientSetupPlayReq{res, c, pathName, trackID, req} //nolint:govet
re := <-res
return re.Path, re.Err
}
internal/serverudp/server.go
+5 -5
View File
@@ -19,12 +19,12 @@ const (
// Publisher is implemented by client.Client.
type Publisher interface {
OnUdpPublisherFrame(int, base.StreamType, []byte)
OnUDPPublisherFrame(int, base.StreamType, []byte)
}
type publisherData struct {
publisher Publisher
trackId int
trackID int
}
// Parent is implemented by program.
@@ -130,7 +130,7 @@ func (s *Server) run() {
return
}
pubData.publisher.OnUdpPublisherFrame(pubData.trackId, s.streamType, buf[:n])
pubData.publisher.OnUDPPublisherFrame(pubData.trackID, s.streamType, buf[:n])
}()
}
}
@@ -150,7 +150,7 @@ func (s *Server) Write(buf []byte, addr *net.UDPAddr) {
}
// AddPublisher adds a publisher.
func (s *Server) AddPublisher(ip net.IP, port int, publisher Publisher, trackId int) {
func (s *Server) AddPublisher(ip net.IP, port int, publisher Publisher, trackID int) {
s.publishersMutex.Lock()
defer s.publishersMutex.Unlock()
@@ -159,7 +159,7 @@ func (s *Server) AddPublisher(ip net.IP, port int, publisher Publisher, trackId
s.publishers[addr] = &publisherData{
publisher: publisher,
trackId: trackId,
trackID: trackID,
}
}
internal/sourcertmp/source.go
+5 -6
View File
@@ -34,7 +34,6 @@ type Parent interface {
// Source is a RTMP source.
type Source struct {
ur string
state bool
wg *sync.WaitGroup
stats *stats.Stats
parent Parent
@@ -238,7 +237,7 @@ func (s *Source) runInner() bool {
}
for i, t := range tracks {
t.Id = i
t.ID = i
}
s.parent.Log("rtmp source ready")
@@ -261,14 +260,14 @@ func (s *Source) runInner() bool {
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 r != nil {
s.parent.OnFrame(audioTrack.Id, gortsplib.StreamTypeRtcp, r)
s.parent.OnFrame(audioTrack.ID, gortsplib.StreamTypeRtcp, r)
}
}
@@ -310,7 +309,7 @@ func (s *Source) runInner() bool {
for _, f := range frames {
videoRtcpSender.ProcessFrame(time.Now(), gortsplib.StreamTypeRtp, f)
s.parent.OnFrame(videoTrack.Id, gortsplib.StreamTypeRtp, f)
s.parent.OnFrame(videoTrack.ID, gortsplib.StreamTypeRtp, f)
}
case av.AAC:
@@ -327,7 +326,7 @@ func (s *Source) runInner() bool {
for _, f := range frames {
audioRtcpSender.ProcessFrame(time.Now(), gortsplib.StreamTypeRtp, f)
s.parent.OnFrame(audioTrack.Id, gortsplib.StreamTypeRtp, f)
s.parent.OnFrame(audioTrack.ID, gortsplib.StreamTypeRtp, f)
}
default:
internal/sourcertsp/source.go
+2 -5
View File
@@ -34,9 +34,6 @@ type Source struct {
// in
terminate chan struct{}
// out
done chan struct{}
}
// New allocates a Source.
@@ -140,8 +137,8 @@ func (s *Source) runInner() bool {
s.parent.OnSourceSetReady(tracks)
defer s.parent.OnSourceSetNotReady()
readerDone := conn.OnFrame(func(trackId int, streamType gortsplib.StreamType, content []byte) {
s.parent.OnFrame(trackId, streamType, content)
readerDone := conn.OnFrame(func(trackID int, streamType gortsplib.StreamType, content []byte) {
s.parent.OnFrame(trackID, streamType, content)
})
for {
main.go
+38 -38
View File
@@ -31,9 +31,9 @@ type program struct {
logHandler *loghandler.LogHandler
metrics *metrics.Metrics
pprof *pprof.Pprof
serverUdpRtp *serverudp.Server
serverUdpRtcp *serverudp.Server
serverTcp *servertcp.Server
serverUDPRtp *serverudp.Server
serverUDPRtcp *serverudp.Server
serverTCP *servertcp.Server
pathMan *pathman.PathManager
clientMan *clientman.ClientManager
confWatcher *confwatcher.ConfWatcher
@@ -51,7 +51,7 @@ func newProgram(args []string) (*program, error) {
kingpin.MustParse(k.Parse(args))
if *argVersion == true {
if *argVersion {
fmt.Println(version)
os.Exit(0)
}
@@ -95,8 +95,8 @@ func (p *program) Log(format string, args ...interface{}) {
countPublishers := atomic.LoadInt64(p.stats.CountPublishers)
countReaders := atomic.LoadInt64(p.stats.CountReaders)
log.Printf(fmt.Sprintf("[%d/%d/%d] "+format, append([]interface{}{countClients,
countPublishers, countReaders}, args...)...))
log.Printf("[%d/%d/%d] "+format, append([]interface{}{countClients,
countPublishers, countReaders}, args...)...)
}
func (p *program) run() {
@@ -157,16 +157,16 @@ func (p *program) createDynamicResources(initial bool) error {
}
if _, ok := p.conf.ProtocolsParsed[gortsplib.StreamProtocolUDP]; ok {
if p.serverUdpRtp == nil {
p.serverUdpRtp, err = serverudp.New(p.conf.WriteTimeout,
if p.serverUDPRtp == nil {
p.serverUDPRtp, err = serverudp.New(p.conf.WriteTimeout,
p.conf.RtpPort, gortsplib.StreamTypeRtp, p)
if err != nil {
return err
}
}
if p.serverUdpRtcp == nil {
p.serverUdpRtcp, err = serverudp.New(p.conf.WriteTimeout,
if p.serverUDPRtcp == nil {
p.serverUDPRtcp, err = serverudp.New(p.conf.WriteTimeout,
p.conf.RtcpPort, gortsplib.StreamTypeRtcp, p)
if err != nil {
return err
@@ -174,8 +174,8 @@ func (p *program) createDynamicResources(initial bool) error {
}
}
if p.serverTcp == nil {
p.serverTcp, err = servertcp.New(p.conf.RtspPort, p)
if p.serverTCP == nil {
p.serverTCP, err = servertcp.New(p.conf.RtspPort, p)
if err != nil {
return err
}
@@ -190,8 +190,8 @@ func (p *program) createDynamicResources(initial bool) error {
if p.clientMan == nil {
p.clientMan = clientman.New(p.conf.RtspPort, p.conf.ReadTimeout,
p.conf.WriteTimeout, p.conf.RunOnConnect, p.conf.RunOnConnectRestart,
p.conf.ProtocolsParsed, p.stats, p.serverUdpRtp, p.serverUdpRtcp,
p.pathMan, p.serverTcp, p)
p.conf.ProtocolsParsed, p.stats, p.serverUDPRtp, p.serverUDPRtcp,
p.pathMan, p.serverTCP, p)
}
return nil
@@ -210,16 +210,16 @@ func (p *program) closeAllResources() {
p.pathMan.Close()
}
if p.serverTcp != nil {
p.serverTcp.Close()
if p.serverTCP != nil {
p.serverTCP.Close()
}
if p.serverUdpRtcp != nil {
p.serverUdpRtcp.Close()
if p.serverUDPRtcp != nil {
p.serverUDPRtcp.Close()
}
if p.serverUdpRtp != nil {
p.serverUdpRtp.Close()
if p.serverUDPRtp != nil {
p.serverUDPRtp.Close()
}
if p.metrics != nil {
@@ -259,23 +259,23 @@ func (p *program) reloadConf() error {
closePprof = true
}
closeServerUdpRtp := false
closeServerUDPRtp := false
if !reflect.DeepEqual(conf.ProtocolsParsed, p.conf.ProtocolsParsed) ||
conf.WriteTimeout != p.conf.WriteTimeout ||
conf.RtpPort != p.conf.RtpPort {
closeServerUdpRtp = true
closeServerUDPRtp = true
}
closeServerUdpRtcp := false
closeServerUDPRtcp := false
if !reflect.DeepEqual(conf.ProtocolsParsed, p.conf.ProtocolsParsed) ||
conf.WriteTimeout != p.conf.WriteTimeout ||
conf.RtcpPort != p.conf.RtcpPort {
closeServerUdpRtcp = true
closeServerUDPRtcp = true
}
closeServerTcp := false
closeServerTCP := false
if conf.RtspPort != p.conf.RtspPort {
closeServerTcp = true
closeServerTCP = true
}
closePathMan := false
@@ -289,9 +289,9 @@ func (p *program) reloadConf() error {
}
closeClientMan := false
if closeServerUdpRtp ||
closeServerUdpRtcp ||
closeServerTcp ||
if closeServerUDPRtp ||
closeServerUDPRtcp ||
closeServerTCP ||
closePathMan ||
conf.RtspPort != p.conf.RtspPort ||
conf.ReadTimeout != p.conf.ReadTimeout ||
@@ -312,19 +312,19 @@ func (p *program) reloadConf() error {
p.pathMan = nil
}
if closeServerTcp {
p.serverTcp.Close()
p.serverTcp = nil
if closeServerTCP {
p.serverTCP.Close()
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
+31 -31
View File
@@ -17,7 +17,7 @@ import (
"github.com/aler9/rtsp-simple-server/internal/conf"
)
var ownDockerIp = func() string {
var ownDockerIP = func() string {
out, err := exec.Command("docker", "network", "inspect", "bridge",
"-f", "{{range .IPAM.Config}}{{.Subnet}}{{end}}").Output()
if err != nil {
@@ -259,7 +259,7 @@ func TestPublish(t *testing.T) {
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", conf.publishProto,
"rtsp://" + ownDockerIp + ":8554/teststream",
"rtsp://" + ownDockerIP + ":8554/teststream",
})
require.NoError(t, err)
defer cnt1.close()
@@ -267,7 +267,7 @@ func TestPublish(t *testing.T) {
default:
cnt1, err := newContainer("gstreamer", "source", []string{
"filesrc location=emptyvideo.ts ! tsdemux ! queue ! video/x-h264 ! h264parse config-interval=1 ! rtspclientsink " +
"location=rtsp://" + ownDockerIp + ":8554/teststream protocols=" + conf.publishProto + " latency=0",
"location=rtsp://" + ownDockerIP + ":8554/teststream protocols=" + conf.publishProto + " latency=0",
})
require.NoError(t, err)
defer cnt1.close()
@@ -277,7 +277,7 @@ func TestPublish(t *testing.T) {
cnt2, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", "udp",
"-i", "rtsp://" + ownDockerIp + ":8554/teststream",
"-i", "rtsp://" + ownDockerIP + ":8554/teststream",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -315,7 +315,7 @@ func TestRead(t *testing.T) {
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", "udp",
"rtsp://" + ownDockerIp + ":8554/teststream",
"rtsp://" + ownDockerIP + ":8554/teststream",
})
require.NoError(t, err)
defer cnt1.close()
@@ -326,7 +326,7 @@ func TestRead(t *testing.T) {
case "ffmpeg":
cnt2, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", conf.readProto,
"-i", "rtsp://" + ownDockerIp + ":8554/teststream",
"-i", "rtsp://" + ownDockerIP + ":8554/teststream",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -342,7 +342,7 @@ func TestRead(t *testing.T) {
if conf.readProto == "tcp" {
args = append(args, "--rtsp-tcp")
}
args = append(args, "rtsp://"+ownDockerIp+":8554/teststream")
args = append(args, "rtsp://"+ownDockerIP+":8554/teststream")
cnt2, err := newContainer("vlc", "dest", args)
require.NoError(t, err)
@@ -369,14 +369,14 @@ func TestTCPOnly(t *testing.T) {
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", "tcp",
"rtsp://" + ownDockerIp + ":8554/teststream",
"rtsp://" + ownDockerIP + ":8554/teststream",
})
require.NoError(t, err)
defer cnt1.close()
cnt2, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", "tcp",
"-i", "rtsp://" + ownDockerIp + ":8554/teststream",
"-i", "rtsp://" + ownDockerIP + ":8554/teststream",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -402,14 +402,14 @@ func TestPathWithSlash(t *testing.T) {
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", "udp",
"rtsp://" + ownDockerIp + ":8554/test/stream",
"rtsp://" + ownDockerIP + ":8554/test/stream",
})
require.NoError(t, err)
defer cnt1.close()
cnt2, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", "udp",
"-i", "rtsp://" + ownDockerIp + ":8554/test/stream",
"-i", "rtsp://" + ownDockerIP + ":8554/test/stream",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -435,14 +435,14 @@ func TestPathWithQuery(t *testing.T) {
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", "udp",
"rtsp://" + ownDockerIp + ":8554/test?param1=val&param2=val",
"rtsp://" + ownDockerIP + ":8554/test?param1=val&param2=val",
})
require.NoError(t, err)
defer cnt1.close()
cnt2, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", "udp",
"-i", "rtsp://" + ownDockerIp + ":8554/test?param3=otherval",
"-i", "rtsp://" + ownDockerIP + ":8554/test?param3=otherval",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -473,7 +473,7 @@ func TestAuth(t *testing.T) {
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", "udp",
"rtsp://testuser:test!$()*+.;<=>[]^_-{}@" + ownDockerIp + ":8554/test/stream",
"rtsp://testuser:test!$()*+.;<=>[]^_-{}@" + ownDockerIP + ":8554/test/stream",
})
require.NoError(t, err)
defer cnt1.close()
@@ -482,7 +482,7 @@ func TestAuth(t *testing.T) {
cnt2, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", "udp",
"-i", "rtsp://" + ownDockerIp + ":8554/test/stream",
"-i", "rtsp://" + ownDockerIP + ":8554/test/stream",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -516,7 +516,7 @@ func TestAuth(t *testing.T) {
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", "udp",
"rtsp://" + ownDockerIp + ":8554/test/stream",
"rtsp://" + ownDockerIP + ":8554/test/stream",
})
require.NoError(t, err)
defer cnt1.close()
@@ -526,7 +526,7 @@ func TestAuth(t *testing.T) {
if soft == "ffmpeg" {
cnt2, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", "udp",
"-i", "rtsp://testuser:test!$()*+.;<=>[]^_-{}@" + ownDockerIp + ":8554/test/stream",
"-i", "rtsp://testuser:test!$()*+.;<=>[]^_-{}@" + ownDockerIP + ":8554/test/stream",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -539,7 +539,7 @@ func TestAuth(t *testing.T) {
} else {
cnt2, err := newContainer("vlc", "dest", []string{
"rtsp://testuser:test!$()*+.;<=>[]^_-{}@" + ownDockerIp + ":8554/test/stream",
"rtsp://testuser:test!$()*+.;<=>[]^_-{}@" + ownDockerIP + ":8554/test/stream",
})
require.NoError(t, err)
defer cnt2.close()
@@ -573,7 +573,7 @@ func TestSourceRtsp(t *testing.T) {
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", "udp",
"rtsp://" + ownDockerIp + ":8554/teststream",
"rtsp://" + ownDockerIP + ":8554/teststream",
})
require.NoError(t, err)
defer cnt1.close()
@@ -596,7 +596,7 @@ func TestSourceRtsp(t *testing.T) {
cnt2, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", "udp",
"-i", "rtsp://" + ownDockerIp + ":8555/proxied",
"-i", "rtsp://" + ownDockerIP + ":8555/proxied",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -641,7 +641,7 @@ func TestSourceRtmp(t *testing.T) {
cnt3, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", "udp",
"-i", "rtsp://" + ownDockerIp + ":8554/proxied",
"-i", "rtsp://" + ownDockerIP + ":8554/proxied",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -657,7 +657,7 @@ func TestRedirect(t *testing.T) {
p1, err := testProgram("paths:\n" +
" path1:\n" +
" source: redirect\n" +
" sourceRedirect: rtsp://" + ownDockerIp + ":8554/path2\n" +
" sourceRedirect: rtsp://" + ownDockerIP + ":8554/path2\n" +
" path2:\n")
require.NoError(t, err)
defer p1.close()
@@ -671,7 +671,7 @@ func TestRedirect(t *testing.T) {
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", "udp",
"rtsp://" + ownDockerIp + ":8554/path2",
"rtsp://" + ownDockerIP + ":8554/path2",
})
require.NoError(t, err)
defer cnt1.close()
@@ -680,7 +680,7 @@ func TestRedirect(t *testing.T) {
cnt2, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", "udp",
"-i", "rtsp://" + ownDockerIp + ":8554/path1",
"-i", "rtsp://" + ownDockerIP + ":8554/path1",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -695,7 +695,7 @@ func TestRedirect(t *testing.T) {
func TestFallback(t *testing.T) {
p1, err := testProgram("paths:\n" +
" path1:\n" +
" fallback: rtsp://" + ownDockerIp + ":8554/path2\n" +
" fallback: rtsp://" + ownDockerIP + ":8554/path2\n" +
" path2:\n")
require.NoError(t, err)
defer p1.close()
@@ -709,7 +709,7 @@ func TestFallback(t *testing.T) {
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", "udp",
"rtsp://" + ownDockerIp + ":8554/path2",
"rtsp://" + ownDockerIP + ":8554/path2",
})
require.NoError(t, err)
defer cnt1.close()
@@ -718,7 +718,7 @@ func TestFallback(t *testing.T) {
cnt2, err := newContainer("ffmpeg", "dest", []string{
"-rtsp_transport", "udp",
"-i", "rtsp://" + ownDockerIp + ":8554/path1",
"-i", "rtsp://" + ownDockerIP + ":8554/path1",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -740,7 +740,7 @@ func TestRunOnDemand(t *testing.T) {
time.Sleep(1 * time.Second)
cnt1, err := newContainer("ffmpeg", "dest", []string{
"-i", "rtsp://" + ownDockerIp + ":8554/ondemand",
"-i", "rtsp://" + ownDockerIP + ":8554/ondemand",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -774,7 +774,7 @@ func TestHotReloading(t *testing.T) {
func() {
cnt1, err := newContainer("ffmpeg", "dest", []string{
"-i", "rtsp://" + ownDockerIp + ":8554/test1",
"-i", "rtsp://" + ownDockerIP + ":8554/test1",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -799,7 +799,7 @@ func TestHotReloading(t *testing.T) {
func() {
cnt1, err := newContainer("ffmpeg", "dest", []string{
"-i", "rtsp://" + ownDockerIp + ":8554/test1",
"-i", "rtsp://" + ownDockerIP + ":8554/test1",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",
@@ -813,7 +813,7 @@ func TestHotReloading(t *testing.T) {
func() {
cnt1, err := newContainer("ffmpeg", "dest", []string{
"-i", "rtsp://" + ownDockerIp + ":8554/test2",
"-i", "rtsp://" + ownDockerIP + ":8554/test2",
"-vframes", "1",
"-f", "image2",
"-y", "/dev/null",