pimhwang
/
mediamtx
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

update gortsplib

pull/52/head
aler9 5 years ago
parent
2396832d52
commit
ce5e11f6c9
6 changed files with 57 additions and 49 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      go.mod
  2. 4
      go.sum
  3. 2
      main_test.go
  4. 51
      server-client.go
  5. 34
      source.go
  6. 13
      utils.go

2
go.mod
Unescape View File

@ -5,7 +5,7 @@ go 1.13
require ( require (
github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect
github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d // indirect github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d // indirect
github.com/aler9/gortsplib v0.0.0-20200713155346-67368c8ec5f6 github.com/aler9/gortsplib v0.0.0-20200718113358-4754822be147
github.com/pion/sdp v1.3.0 github.com/pion/sdp v1.3.0
github.com/stretchr/testify v1.5.1 github.com/stretchr/testify v1.5.1
gopkg.in/alecthomas/kingpin.v2 v2.2.6 gopkg.in/alecthomas/kingpin.v2 v2.2.6

4
go.sum
Unescape 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/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 h1:UQZhZ2O0vMHr2cI+DC1Mbh0TJxzA3RcLoMsFw+aXw7E=
github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho= github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
github.com/aler9/gortsplib v0.0.0-20200713155346-67368c8ec5f6 h1:ndRQ+7v10ok+unOVo6f7PIffg65yUukq2qKgR44xI0I= github.com/aler9/gortsplib v0.0.0-20200718113358-4754822be147 h1:1rPmAAdNf9bdxXWuY9PSko6eFMjwLiwlTXbOukG0cFg=
github.com/aler9/gortsplib v0.0.0-20200713155346-67368c8ec5f6/go.mod h1:17dcA4Qak5TLqgun8OR0wnSbFQIg4cvYVSf1nbCt+qU= github.com/aler9/gortsplib v0.0.0-20200718113358-4754822be147/go.mod h1:17dcA4Qak5TLqgun8OR0wnSbFQIg4cvYVSf1nbCt+qU=
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= 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.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/pion/rtcp v1.2.3 h1:2wrhKnqgSz91Q5nzYTO07mQXztYPtxL8a0XOss4rJqA= github.com/pion/rtcp v1.2.3 h1:2wrhKnqgSz91Q5nzYTO07mQXztYPtxL8a0XOss4rJqA=

2
main_test.go
Unescape View File

@ -92,7 +92,7 @@ func (c *container) wait() int {
return int(code) return int(code)
} }
func TestProtocols(t *testing.T) { func TestPublishRead(t *testing.T) {
for _, conf := range [][3]string{ for _, conf := range [][3]string{
{"udp", "udp", "ffmpeg"}, {"udp", "udp", "ffmpeg"},
{"udp", "tcp", "ffmpeg"}, {"udp", "tcp", "ffmpeg"},

51
server-client.go
Unescape View File

@ -313,7 +313,7 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
StatusCode: gortsplib.StatusOK, StatusCode: gortsplib.StatusOK,
Header: gortsplib.Header{ Header: gortsplib.Header{
"CSeq": cseq, "CSeq": cseq,
"Public": []string{strings.Join([]string{ "Public": gortsplib.HeaderValue{strings.Join([]string{
string(gortsplib.DESCRIBE), string(gortsplib.DESCRIBE),
string(gortsplib.ANNOUNCE), string(gortsplib.ANNOUNCE),
string(gortsplib.SETUP), string(gortsplib.SETUP),
@ -359,8 +359,8 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
StatusCode: gortsplib.StatusOK, StatusCode: gortsplib.StatusOK,
Header: gortsplib.Header{ Header: gortsplib.Header{
"CSeq": cseq, "CSeq": cseq,
"Content-Base": []string{req.Url.String() + "/"}, "Content-Base": gortsplib.HeaderValue{req.Url.String() + "/"},
"Content-Type": []string{"application/sdp"}, "Content-Type": gortsplib.HeaderValue{"application/sdp"},
}, },
Content: sdp, Content: sdp,
}) })
@ -410,13 +410,21 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("invalid SDP: %s", err)) c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("invalid SDP: %s", err))
return false return false
} }
sdpParsed, req.Content = sdpForServer(sdpParsed)
if len(sdpParsed.MediaDescriptions) == 0 { if len(sdpParsed.MediaDescriptions) == 0 {
c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("no tracks defined")) c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("no tracks defined"))
return false return false
} }
var tracks []*gortsplib.Track
for i, media := range sdpParsed.MediaDescriptions {
tracks = append(tracks, &gortsplib.Track{
Id: i,
Media: media,
})
}
sdpParsed, req.Content = sdpForServer(tracks)
res := make(chan error) res := make(chan error)
c.p.events <- programEventClientAnnounce{res, c, path} c.p.events <- programEventClientAnnounce{res, c, path}
err = <-res err = <-res
@ -437,13 +445,12 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
return true return true
case gortsplib.SETUP: case gortsplib.SETUP:
tsRaw, ok := req.Header["Transport"] th, err := gortsplib.ReadHeaderTransport(req.Header["Transport"])
if !ok || len(tsRaw) != 1 { if err != nil {
c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("transport header missing")) c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("transport header: %s", err))
return false return false
} }
th := gortsplib.ReadHeaderTransport(tsRaw[0])
if _, ok := th["multicast"]; ok { if _, ok := th["multicast"]; ok {
c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("multicast is not supported")) c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("multicast is not supported"))
return false return false
@ -486,7 +493,7 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
rtpPort, rtcpPort := th.GetPorts("client_port") rtpPort, rtcpPort := th.GetPorts("client_port")
if rtpPort == 0 || rtcpPort == 0 { if rtpPort == 0 || rtcpPort == 0 {
c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("transport header does not have valid client ports (%s)", tsRaw[0])) c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("transport header does not have valid client ports (%v)", req.Header["Transport"]))
return false return false
} }
@ -512,13 +519,13 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
StatusCode: gortsplib.StatusOK, StatusCode: gortsplib.StatusOK,
Header: gortsplib.Header{ Header: gortsplib.Header{
"CSeq": cseq, "CSeq": cseq,
"Transport": []string{strings.Join([]string{ "Transport": gortsplib.HeaderValue{strings.Join([]string{
"RTP/AVP/UDP", "RTP/AVP/UDP",
"unicast", "unicast",
fmt.Sprintf("client_port=%d-%d", rtpPort, rtcpPort), fmt.Sprintf("client_port=%d-%d", rtpPort, rtcpPort),
fmt.Sprintf("server_port=%d-%d", c.p.conf.RtpPort, c.p.conf.RtcpPort), fmt.Sprintf("server_port=%d-%d", c.p.conf.RtpPort, c.p.conf.RtcpPort),
}, ";")}, }, ";")},
"Session": []string{"12345678"}, "Session": gortsplib.HeaderValue{"12345678"},
}, },
}) })
return true return true
@ -554,18 +561,18 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
StatusCode: gortsplib.StatusOK, StatusCode: gortsplib.StatusOK,
Header: gortsplib.Header{ Header: gortsplib.Header{
"CSeq": cseq, "CSeq": cseq,
"Transport": []string{strings.Join([]string{ "Transport": gortsplib.HeaderValue{strings.Join([]string{
"RTP/AVP/TCP", "RTP/AVP/TCP",
"unicast", "unicast",
fmt.Sprintf("interleaved=%s", interleaved), fmt.Sprintf("interleaved=%s", interleaved),
}, ";")}, }, ";")},
"Session": []string{"12345678"}, "Session": gortsplib.HeaderValue{"12345678"},
}, },
}) })
return true return true
} else { } else {
c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("transport header does not contain a valid protocol (RTP/AVP, RTP/AVP/UDP or RTP/AVP/TCP) (%s)", tsRaw[0])) c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("transport header does not contain a valid protocol (RTP/AVP, RTP/AVP/UDP or RTP/AVP/TCP) (%s)", req.Header["Transport"]))
return false return false
} }
@ -601,7 +608,7 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
rtpPort, rtcpPort := th.GetPorts("client_port") rtpPort, rtcpPort := th.GetPorts("client_port")
if rtpPort == 0 || rtcpPort == 0 { if rtpPort == 0 || rtcpPort == 0 {
c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("transport header does not have valid client ports (%s)", tsRaw[0])) c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("transport header does not have valid client ports (%s)", req.Header["Transport"]))
return false return false
} }
@ -627,13 +634,13 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
StatusCode: gortsplib.StatusOK, StatusCode: gortsplib.StatusOK,
Header: gortsplib.Header{ Header: gortsplib.Header{
"CSeq": cseq, "CSeq": cseq,
"Transport": []string{strings.Join([]string{ "Transport": gortsplib.HeaderValue{strings.Join([]string{
"RTP/AVP/UDP", "RTP/AVP/UDP",
"unicast", "unicast",
fmt.Sprintf("client_port=%d-%d", rtpPort, rtcpPort), fmt.Sprintf("client_port=%d-%d", rtpPort, rtcpPort),
fmt.Sprintf("server_port=%d-%d", c.p.conf.RtpPort, c.p.conf.RtcpPort), fmt.Sprintf("server_port=%d-%d", c.p.conf.RtpPort, c.p.conf.RtcpPort),
}, ";")}, }, ";")},
"Session": []string{"12345678"}, "Session": gortsplib.HeaderValue{"12345678"},
}, },
}) })
return true return true
@ -679,18 +686,18 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
StatusCode: gortsplib.StatusOK, StatusCode: gortsplib.StatusOK,
Header: gortsplib.Header{ Header: gortsplib.Header{
"CSeq": cseq, "CSeq": cseq,
"Transport": []string{strings.Join([]string{ "Transport": gortsplib.HeaderValue{strings.Join([]string{
"RTP/AVP/TCP", "RTP/AVP/TCP",
"unicast", "unicast",
fmt.Sprintf("interleaved=%s", interleaved), fmt.Sprintf("interleaved=%s", interleaved),
}, ";")}, }, ";")},
"Session": []string{"12345678"}, "Session": gortsplib.HeaderValue{"12345678"},
}, },
}) })
return true return true
} else { } else {
c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("transport header does not contain a valid protocol (RTP/AVP, RTP/AVP/UDP or RTP/AVP/TCP) (%s)", tsRaw[0])) c.writeResError(req, gortsplib.StatusBadRequest, fmt.Errorf("transport header does not contain a valid protocol (RTP/AVP, RTP/AVP/UDP or RTP/AVP/TCP) (%s)", req.Header["Transport"]))
return false return false
} }
@ -727,7 +734,7 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
StatusCode: gortsplib.StatusOK, StatusCode: gortsplib.StatusOK,
Header: gortsplib.Header{ Header: gortsplib.Header{
"CSeq": cseq, "CSeq": cseq,
"Session": []string{"12345678"}, "Session": gortsplib.HeaderValue{"12345678"},
}, },
}) })
@ -755,7 +762,7 @@ func (c *serverClient) handleRequest(req *gortsplib.Request) bool {
StatusCode: gortsplib.StatusOK, StatusCode: gortsplib.StatusOK,
Header: gortsplib.Header{ Header: gortsplib.Header{
"CSeq": cseq, "CSeq": cseq,
"Session": []string{"12345678"}, "Session": gortsplib.HeaderValue{"12345678"},
}, },
}) })

34
source.go
Unescape View File

@ -29,7 +29,7 @@ type source struct {
u *url.URL u *url.URL
proto streamProtocol proto streamProtocol
ready bool ready bool
clientSdpParsed *sdp.SessionDescription clientTracks []*gortsplib.Track
serverSdpText []byte serverSdpText []byte
serverSdpParsed *sdp.SessionDescription serverSdpParsed *sdp.SessionDescription
rtcpReceivers []*gortsplib.RtcpReceiver rtcpReceivers []*gortsplib.RtcpReceiver
@ -155,16 +155,16 @@ func (s *source) do() bool {
return true return true
} }
clientSdpParsed, _, err := conn.Describe(s.u) clientTracks, _, err := conn.Describe(s.u)
if err != nil { if err != nil {
s.log("ERR: %s", err) s.log("ERR: %s", err)
return true return true
} }
// create a filtered SDP that is used by the server (not by the client) // create a filtered SDP that is used by the server (not by the client)
serverSdpParsed, serverSdpText := sdpForServer(clientSdpParsed) serverSdpParsed, serverSdpText := sdpForServer(clientTracks)
s.clientSdpParsed = clientSdpParsed s.clientTracks = clientTracks
s.serverSdpText = serverSdpText s.serverSdpText = serverSdpText
s.serverSdpParsed = serverSdpParsed s.serverSdpParsed = serverSdpParsed
@ -187,7 +187,7 @@ func (s *source) runUdp(conn *gortsplib.ConnClient) bool {
} }
}() }()
for i, media := range s.clientSdpParsed.MediaDescriptions { for i, track := range s.clientTracks {
var rtpPort int var rtpPort int
var rtcpPort int var rtcpPort int
var rtpl *sourceUdpListener var rtpl *sourceUdpListener
@ -217,7 +217,7 @@ func (s *source) runUdp(conn *gortsplib.ConnClient) bool {
} }
}() }()
rtpServerPort, rtcpServerPort, _, err := conn.SetupUdp(s.u, media, rtpPort, rtcpPort) rtpServerPort, rtcpServerPort, _, err := conn.SetupUdp(s.u, track, rtpPort, rtcpPort)
if err != nil { if err != nil {
s.log("ERR: %s", err) s.log("ERR: %s", err)
rtpl.close() rtpl.close()
@ -240,8 +240,8 @@ func (s *source) runUdp(conn *gortsplib.ConnClient) bool {
return true return true
} }
s.rtcpReceivers = make([]*gortsplib.RtcpReceiver, len(s.clientSdpParsed.MediaDescriptions)) s.rtcpReceivers = make([]*gortsplib.RtcpReceiver, len(s.clientTracks))
for trackId := range s.clientSdpParsed.MediaDescriptions { for trackId := range s.clientTracks {
s.rtcpReceivers[trackId] = gortsplib.NewRtcpReceiver() s.rtcpReceivers[trackId] = gortsplib.NewRtcpReceiver()
} }
@ -274,7 +274,7 @@ outer:
} }
case <-checkStreamTicker.C: case <-checkStreamTicker.C:
for trackId := range s.clientSdpParsed.MediaDescriptions { for trackId := range s.clientTracks {
if time.Since(s.rtcpReceivers[trackId].LastFrameTime()) >= s.p.conf.StreamDeadAfter { if time.Since(s.rtcpReceivers[trackId].LastFrameTime()) >= s.p.conf.StreamDeadAfter {
s.log("ERR: stream is dead") s.log("ERR: stream is dead")
ret = true ret = true
@ -283,7 +283,7 @@ outer:
} }
case <-receiverReportTicker.C: case <-receiverReportTicker.C:
for trackId := range s.clientSdpParsed.MediaDescriptions { for trackId := range s.clientTracks {
frame := s.rtcpReceivers[trackId].Report() frame := s.rtcpReceivers[trackId].Report()
sourceUdpListenerPairs[trackId].rtcpl.writeChan <- &udpAddrBufPair{ sourceUdpListenerPairs[trackId].rtcpl.writeChan <- &udpAddrBufPair{
addr: &net.UDPAddr{ addr: &net.UDPAddr{
@ -308,7 +308,7 @@ outer:
pair.rtcpl.stop() pair.rtcpl.stop()
} }
for trackId := range s.clientSdpParsed.MediaDescriptions { for trackId := range s.clientTracks {
s.rtcpReceivers[trackId].Close() s.rtcpReceivers[trackId].Close()
} }
@ -316,8 +316,8 @@ outer:
} }
func (s *source) runTcp(conn *gortsplib.ConnClient) bool { func (s *source) runTcp(conn *gortsplib.ConnClient) bool {
for i, media := range s.clientSdpParsed.MediaDescriptions { for _, track := range s.clientTracks {
_, err := conn.SetupTcp(s.u, media, i) _, err := conn.SetupTcp(s.u, track)
if err != nil { if err != nil {
s.log("ERR: %s", err) s.log("ERR: %s", err)
return true return true
@ -330,8 +330,8 @@ func (s *source) runTcp(conn *gortsplib.ConnClient) bool {
return true return true
} }
s.rtcpReceivers = make([]*gortsplib.RtcpReceiver, len(s.clientSdpParsed.MediaDescriptions)) s.rtcpReceivers = make([]*gortsplib.RtcpReceiver, len(s.clientTracks))
for trackId := range s.clientSdpParsed.MediaDescriptions { for trackId := range s.clientTracks {
s.rtcpReceivers[trackId] = gortsplib.NewRtcpReceiver() s.rtcpReceivers[trackId] = gortsplib.NewRtcpReceiver()
} }
@ -375,7 +375,7 @@ outer:
break outer break outer
case <-receiverReportTicker.C: case <-receiverReportTicker.C:
for trackId := range s.clientSdpParsed.MediaDescriptions { for trackId := range s.clientTracks {
frame := s.rtcpReceivers[trackId].Report() frame := s.rtcpReceivers[trackId].Report()
conn.WriteFrame(&gortsplib.InterleavedFrame{ conn.WriteFrame(&gortsplib.InterleavedFrame{
@ -391,7 +391,7 @@ outer:
s.p.events <- programEventStreamerNotReady{s} s.p.events <- programEventStreamerNotReady{s}
for trackId := range s.clientSdpParsed.MediaDescriptions { for trackId := range s.clientTracks {
s.rtcpReceivers[trackId].Close() s.rtcpReceivers[trackId].Close()
} }

13
utils.go
Unescape View File

@ -5,6 +5,7 @@ import (
"net" "net"
"strconv" "strconv"
"github.com/aler9/gortsplib"
"github.com/pion/sdp" "github.com/pion/sdp"
) )
@ -73,7 +74,7 @@ func (db *doubleBuffer) swap() []byte {
return ret return ret
} }
func sdpForServer(sin *sdp.SessionDescription) (*sdp.SessionDescription, []byte) { func sdpForServer(tracks []*gortsplib.Track) (*sdp.SessionDescription, []byte) {
sout := &sdp.SessionDescription{ sout := &sdp.SessionDescription{
SessionName: "Stream", SessionName: "Stream",
Origin: sdp.Origin{ Origin: sdp.Origin{
@ -87,18 +88,18 @@ func sdpForServer(sin *sdp.SessionDescription) (*sdp.SessionDescription, []byte)
}, },
} }
for i, min := range sin.MediaDescriptions { for i, track := range tracks {
mout := &sdp.MediaDescription{ mout := &sdp.MediaDescription{
MediaName: sdp.MediaName{ MediaName: sdp.MediaName{
Media: min.MediaName.Media, Media: track.Media.MediaName.Media,
Protos: []string{"RTP", "AVP"}, // override protocol Protos: []string{"RTP", "AVP"}, // override protocol
Formats: min.MediaName.Formats, Formats: track.Media.MediaName.Formats,
}, },
Bandwidth: min.Bandwidth, Bandwidth: track.Media.Bandwidth,
Attributes: func() []sdp.Attribute { Attributes: func() []sdp.Attribute {
var ret []sdp.Attribute var ret []sdp.Attribute
for _, attr := range min.Attributes { for _, attr := range track.Media.Attributes {
if attr.Key == "rtpmap" || attr.Key == "fmtp" { if attr.Key == "rtpmap" || attr.Key == "fmtp" {
ret = append(ret, attr) ret = append(ret, attr)
} }

Write Preview
Loading…
Cancel
Save