apps/plugin-rtsp
1
0
Fork 0
mirror of https://github.com/Monibuca/plugin-rtsp.git synced 2025-10-31 19:03:05 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

跟随升级gotsplib

Browse Source
This commit is contained in:
dexter
2021-11-18 19:05:06 +08:00
parent 0ac9513920
commit ef106e42f8
6 changed files with 65 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
client.go
View File

@@ -7,11 +7,12 @@ import (
. "github.com/Monibuca/engine/v3"
. "github.com/Monibuca/utils/v3"
"github.com/aler9/gortsplib"
"github.com/aler9/gortsplib/pkg/base"
)
type RTSPClient struct {
RTSPublisher
Conn *gortsplib.ClientConn
gortsplib.Client `json:"-"`
}
// PullStream 从外部拉流
@@ -21,6 +22,9 @@ func (rtsp *RTSPClient) PullStream(streamPath string, rtspUrl string) (err error
Type: "RTSP Pull",
ExtraProp: rtsp,
}
rtsp.OnPacketRTP = func(trackID int, payload []byte) {
rtsp.processFunc[trackID](payload)
}
if result := rtsp.Publish(); result {
rtsp.URL = rtspUrl
if config.Reconnect {
@@ -29,7 +33,7 @@ func (rtsp *RTSPClient) PullStream(streamPath string, rtspUrl string) (err error
Printf("reconnecting:%s in 5 seconds", rtspUrl)
time.Sleep(time.Second * 5)
}
rtsp.Conn.Close()
rtsp.Client.Close()
if rtsp.IsTimeout {
go rtsp.PullStream(streamPath, rtspUrl)
}
@@ -37,7 +41,7 @@ func (rtsp *RTSPClient) PullStream(streamPath string, rtspUrl string) (err error
} else {
go func() {
rtsp.startStream()
rtsp.Conn.Close()
rtsp.Client.Close()
}()
}
return
@@ -49,19 +53,45 @@ func (client *RTSPClient) startStream() {
if client.Err() != nil {
return
}
// startTime := time.Now()
//loggerTime := time.Now().Add(-10 * time.Second)
conn, err := gortsplib.DialRead(client.URL)
// parse URL
u, err := base.ParseURL(client.URL)
if err != nil {
Printf("ParseURL:%s error:%v", client.URL, err)
return
}
// connect to the server
if err = client.Start(u.Scheme, u.Host); err != nil {
Printf("connect:%s error:%v", client.URL, err)
return
}
client.Conn = conn
tracks := conn.Tracks()
client.setTracks(tracks)
err = conn.ReadFrames(func(trackID int, streamType gortsplib.StreamType, payload []byte) {
if streamType == gortsplib.StreamTypeRTP {
client.processFunc[trackID](payload)
var res *base.Response
if res, err = client.Options(u); err != nil {
Printf("option:%s error:%v", client.URL, err)
return
}
})
Println(res)
// find published tracks
tracks, baseURL, res, err := client.Describe(u)
if err != nil {
Printf("Describe:%s error:%v", baseURL.String(), err)
return
}
Println(res)
client.setTracks(tracks)
for _, track := range tracks {
if res, err = client.Setup(true, baseURL, track, 0, 0); err != nil {
Printf("Setup:%s error:%v", baseURL.String(), err)
return
}
Println(res)
}
// start reading tracks
if res, err = client.Play(nil); err != nil {
Printf("Play:%s error:%v", baseURL.String(), err)
return
}
Println(res)
// wait until a fatal error
err = client.Wait()
Printf("Wait:%s error:%v", baseURL.String(), err)
}

8
go.mod
View File

@@ -3,10 +3,10 @@ module github.com/Monibuca/plugin-rtsp/v3
go 1.16
require (
github.com/Monibuca/engine/v3 v3.3.10
github.com/Monibuca/engine/v3 v3.3.11
github.com/Monibuca/utils/v3 v3.0.4
github.com/aler9/gortsplib v0.0.0-20211106122816-6e38851a096e
github.com/aler9/gortsplib v0.0.0-20211115164017-1411cb33f558
github.com/pion/rtp v1.7.4
github.com/pion/sdp/v3 v3.0.2
golang.org/x/sys v0.0.0-20211113001501-0c823b97ae02 // indirect
github.com/pion/sdp/v3 v3.0.4
golang.org/x/sys v0.0.0-20211116061358-0a5406a5449c // indirect
)

8
go.sum
View File

@@ -3,10 +3,14 @@ github.com/BurntSushi/toml v0.4.1 h1:GaI7EiDXDRfa8VshkTj7Fym7ha+y8/XxIgD2okUIjLw
github.com/BurntSushi/toml v0.4.1/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
github.com/Monibuca/engine/v3 v3.3.10 h1:zRw9aGEmB6K6ee0figdRh2HZFGQSn2nvpptMT3Xm0HY=
github.com/Monibuca/engine/v3 v3.3.10/go.mod h1:LowMZ/iw4t6tfTZkSYZHIA0Z1HE8b7xfTDLO4WhX3Hg=
github.com/Monibuca/engine/v3 v3.3.11 h1:4zuuW34UgBxNT66wg+SbYm0C/kp2wghnoqWjM06eIIs=
github.com/Monibuca/engine/v3 v3.3.11/go.mod h1:LowMZ/iw4t6tfTZkSYZHIA0Z1HE8b7xfTDLO4WhX3Hg=
github.com/Monibuca/utils/v3 v3.0.4 h1:PssGhww+qePzw4qpB3g2DCG5Buru0Cu64UiqtAPuHjc=
github.com/Monibuca/utils/v3 v3.0.4/go.mod h1:RpNS95gapWs6gimwh8Xn2x72FN5tO7Powabj7dTFyvE=
github.com/aler9/gortsplib v0.0.0-20211106122816-6e38851a096e h1:qSjVAaIvJukmEuLxV0agmQ5KmBabBK+jzb+eNqG3Z+w=
github.com/aler9/gortsplib v0.0.0-20211106122816-6e38851a096e/go.mod h1:fyQrQyHo8QvdR/h357tkv1g36VesZlzEPsdAu2VrHHc=
github.com/aler9/gortsplib v0.0.0-20211115164017-1411cb33f558 h1:uyW1alIzoJCAmNZ2xuo5EOTbSbf9W7tVYOzb0UrzZ0U=
github.com/aler9/gortsplib v0.0.0-20211115164017-1411cb33f558/go.mod h1:fyQrQyHo8QvdR/h357tkv1g36VesZlzEPsdAu2VrHHc=
github.com/asticode/go-astikit v0.20.0/go.mod h1:h4ly7idim1tNhaVkdVBeXQZEE3L0xblP7fCWbgwipF0=
github.com/asticode/go-astits v1.10.0/go.mod h1:DkOWmBNQpnr9mv24KfZjq4JawCFX1FCqjLVGvO0DygQ=
github.com/cnotch/apirouter v0.0.0-20200731232942-89e243a791f3/go.mod h1:5deJPLON/x/s2dLOQfuKS0lenhOIT4xX0pvtN/OEIuY=
@@ -54,6 +58,8 @@ github.com/pion/rtp v1.7.4 h1:4dMbjb1SuynU5OpA3kz1zHK+u+eOCQjW3MAeVHf1ODA=
github.com/pion/rtp v1.7.4/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
github.com/pion/sdp/v3 v3.0.2 h1:UNnSPVaMM+Pdu/mR9UvAyyo6zkdYbKeuOooCwZvTl/g=
github.com/pion/sdp/v3 v3.0.2/go.mod h1:bNiSknmJE0HYBprTHXKPQ3+JjacTv5uap92ueJZKsRk=
github.com/pion/sdp/v3 v3.0.4 h1:2Kf+dgrzJflNCSw3TV5v2VLeI0s/qkzy2r5jlR0wzf8=
github.com/pion/sdp/v3 v3.0.4/go.mod h1:bNiSknmJE0HYBprTHXKPQ3+JjacTv5uap92ueJZKsRk=
github.com/pixelbender/go-sdp v1.1.0/go.mod h1:6IBlz9+BrUHoFTea7gcp4S54khtOhjCW/nVDLhmZBAs=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -86,6 +92,8 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20211113001501-0c823b97ae02 h1:7NCfEGl0sfUojmX78nK9pBJuUlSZWEJA/TwASvfiPLo=
golang.org/x/sys v0.0.0-20211113001501-0c823b97ae02/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20211116061358-0a5406a5449c h1:DHcbWVXeY+0Y8HHKR+rbLwnoh2F4tNCY7rTiHJ30RmA=
golang.org/x/sys v0.0.0-20211116061358-0a5406a5449c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

3
main.go
View File

@@ -79,11 +79,12 @@ func ListenRtsp(addr string) error {
defer log.Println("rtsp server start!")
s := &gortsplib.Server{
Handler: &RTSPServer{},
RTSPAddress: addr,
UDPRTPAddress: config.UDPAddr,
UDPRTCPAddress: config.RTCPAddr,
MulticastIPRange: "224.1.0.0/16",
MulticastRTPPort: 8002,
MulticastRTCPPort: 8003,
}
return s.StartAndWait(addr)
return s.StartAndWait()
}

4
publisher.go
View File

@@ -19,11 +19,15 @@ type RTSPublisher struct {
}
func (p *RTSPublisher) setTracks(tracks gortsplib.Tracks) {
if p.processFunc != nil {
p.processFunc = p.processFunc[:0]
}
for i, track := range tracks {
v, ok := track.Media.Attribute("rtpmap")
if !ok {
continue
}
fmtp := make(map[string]string)
if v, ok := track.Media.Attribute("fmtp"); ok {
if tmp := strings.SplitN(v, " ", 2); len(tmp) == 2 {

10
server.go
View File

@@ -100,7 +100,7 @@ func (sh *RTSPServer) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (*ba
for _, nalu := range pack.NALUs {
for _, pack := range vpacketer.Packetize(nalu, (ts-st)*90) {
rtp, _ := pack.Marshal()
stream.WriteFrame(trackId, gortsplib.StreamTypeRTP, rtp)
stream.WritePacketRTP(trackId, rtp)
}
}
st = ts
@@ -124,7 +124,7 @@ func (sh *RTSPServer) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (*ba
sub.OnAudio = func(ts uint32, pack *engine.AudioPack) {
for _, pack := range apacketizer.Packetize(pack.Raw, (ts-st)*8) {
buf, _ := pack.Marshal()
stream.WriteFrame(trackId, gortsplib.StreamTypeRTP, buf)
stream.WritePacketRTP(trackId, buf)
}
st = ts
}
@@ -143,7 +143,7 @@ func (sh *RTSPServer) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (*ba
sub.OnAudio = func(ts uint32, pack *engine.AudioPack) {
for _, pack := range apacketizer.Packetize(pack.Raw, (ts-st)*uint32(mpegConf.SampleRate)/1000) {
buf, _ := pack.Marshal()
stream.WriteFrame(trackId, gortsplib.StreamTypeRTP, buf)
stream.WritePacketRTP(trackId, buf)
}
st = ts
}
@@ -241,10 +241,8 @@ func (sh *RTSPServer) OnRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
}
// called after receiving a frame.
func (sh *RTSPServer) OnFrame(ctx *gortsplib.ServerHandlerOnFrameCtx) {
func (sh *RTSPServer) OnPacketRTP(ctx *gortsplib.ServerHandlerOnPacketRTPCtx) {
if p, ok := sh.Load(ctx.Session); ok {
if ctx.StreamType == gortsplib.StreamTypeRTP {
p.(*RTSPublisher).processFunc[ctx.TrackID](ctx.Payload)
}
}
}