From 19440f248689963cdd838655940565c1d3e6c520 Mon Sep 17 00:00:00 2001 From: aler9 <46489434+aler9@users.noreply.github.com> Date: Sat, 5 Dec 2020 20:42:59 +0100 Subject: [PATCH] switch to golangci-lint --- .github/workflows/lint.yml | 25 ++ Makefile | 12 +- README.md | 2 +- go.mod | 2 +- go.sum | 4 +- internal/client/client.go | 409 ++++++++++++++++---------------- internal/clientman/clientman.go | 31 ++- internal/conf/pathconf.go | 10 +- internal/conf/utils.go | 4 +- internal/path/path.go | 14 +- internal/path/readersmap.go | 4 +- internal/pathman/pathman.go | 28 +-- internal/serverudp/server.go | 10 +- internal/sourcertmp/source.go | 11 +- internal/sourcertsp/source.go | 7 +- main.go | 76 +++--- main_test.go | 62 ++--- 17 files changed, 367 insertions(+), 344 deletions(-) create mode 100644 .github/workflows/lint.yml diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml new file mode 100644 index 00000000..02511d81 --- /dev/null +++ b/.github/workflows/lint.yml @@ -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 diff --git a/Makefile b/Makefile index a8af647f..81c33e58 100644 --- a/Makefile +++ b/Makefile @@ -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 diff --git a/README.md b/README.md index 954e542f..6e461433 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/go.mod b/go.mod index 3f5eb037..356ac7f8 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-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 diff --git a/go.sum b/go.sum index 4ca65832..09516ad8 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-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= diff --git a/internal/client/client.go b/internal/client/client.go index b45f33fe..ad5e287c 100644 --- a/internal/client/client.go +++ b/internal/client/client.go @@ -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() } diff --git a/internal/clientman/clientman.go b/internal/clientman/clientman.go index 178feeca..dbdff97e 100644 --- a/internal/clientman/clientman.go +++ b/internal/clientman/clientman.go @@ -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) } diff --git a/internal/conf/pathconf.go b/internal/conf/pathconf.go index 4b15544b..dc189661 100644 --- a/internal/conf/pathconf.go +++ b/internal/conf/pathconf.go @@ -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 } diff --git a/internal/conf/utils.go b/internal/conf/utils.go index f3953fb9..ce174b0a 100644 --- a/internal/conf/utils.go +++ b/internal/conf/utils.go @@ -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 } diff --git a/internal/path/path.go b/internal/path/path.go index 02982b13..92dc7411 100644 --- a/internal/path/path.go +++ b/internal/path/path.go @@ -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) } diff --git a/internal/path/readersmap.go b/internal/path/readersmap.go index 63177292..1998cd21 100644 --- a/internal/path/readersmap.go +++ b/internal/path/readersmap.go @@ -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) } } diff --git a/internal/pathman/pathman.go b/internal/pathman/pathman.go index 0177c591..efde5063 100644 --- a/internal/pathman/pathman.go +++ b/internal/pathman/pathman.go @@ -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 } diff --git a/internal/serverudp/server.go b/internal/serverudp/server.go index b666109a..3dc3b252 100644 --- a/internal/serverudp/server.go +++ b/internal/serverudp/server.go @@ -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, } } diff --git a/internal/sourcertmp/source.go b/internal/sourcertmp/source.go index ca0efd79..6ed413d1 100644 --- a/internal/sourcertmp/source.go +++ b/internal/sourcertmp/source.go @@ -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: diff --git a/internal/sourcertsp/source.go b/internal/sourcertsp/source.go index be702559..c5e8ef1f 100644 --- a/internal/sourcertsp/source.go +++ b/internal/sourcertsp/source.go @@ -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 { diff --git a/main.go b/main.go index 60602739..4972d186 100644 --- a/main.go +++ b/main.go @@ -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 { diff --git a/main_test.go b/main_test.go index d60e3541..2e94b11e 100644 --- a/main_test.go +++ b/main_test.go @@ -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¶m2=val", + "rtsp://" + ownDockerIP + ":8554/test?param1=val¶m2=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",