diff --git a/.gitignore b/.gitignore deleted file mode 100644 index 7f32e4b..0000000 --- a/.gitignore +++ /dev/null @@ -1,2 +0,0 @@ -.vscode -node_modules \ No newline at end of file diff --git a/README.md b/README.md index a50a5c8..b826cef 100644 --- a/README.md +++ b/README.md @@ -7,26 +7,19 @@ github.com/Monibuca/plugin-rtsp ## 插件引入 ```go import ( - _ "github.com/Monibuca/plugin-rtsp" + _ "m7s.live/plugin/rtsp/v4" ) ``` ## 默认插件配置 -```toml -[RTSP] +```yaml +rtsp: # 端口接收推流 -ListenAddr = ":554" -Reconnect = true -[RTSP.AutoPullList] -"live/rtsp1" = "rtsp://admin:admin@192.168.1.212:554/cam/realmonitor?channel=1&subtype=1" -"live/rtsp2" = "rtsp://admin:admin@192.168.1.212:554/cam/realmonitor?channel=2&subtype=1" + listenaddr: :554 + udpaddr: :8000 + readbuffersize: 2048 ``` - -- `ListenAddr`是监听的地址 -- `Reconnect` 是否自动重连 -- `RTSP.AutoPullList` 可以配置多项,用于自动拉流,key是streamPath,value是远程rtsp地址 - ### 特殊功能 当自动拉流列表中当的streamPath为sub/xxx 这种形式的话,在gb28181的分屏显示时会优先采用rtsp流,已实现分屏观看子码流效果 @@ -45,7 +38,7 @@ ffmpeg -i **** rtsp://localhost/live/test ### 从远程拉取rtsp到m7s中 可调用接口 -`/api/rtsp/pull?target=[RTSP地址]&streamPath=[流标识]` +`rtsp/api/pull?target=[RTSP地址]&streamPath=[流标识]` ## 使用编程方式拉流 ```go @@ -55,9 +48,8 @@ new(RTSPClient).PullStream("live/user1","rtsp://xxx.xxx.xxx.xxx/live/user1") ### 罗列所有的rtsp协议的流 可调用接口 -`/api/rtsp/list` +`rtsp/api/list` ### 从m7s中拉取rtsp协议流 直接通过协议rtsp://xxx.xxx.xxx.xxx/live/user1 即可播放 -> h265 编码拉流尚未实现,敬请期待 diff --git a/client.go b/client.go deleted file mode 100644 index 27010e7..0000000 --- a/client.go +++ /dev/null @@ -1,213 +0,0 @@ -package rtsp - -import ( - "errors" - "time" - "unsafe" - - . "github.com/Monibuca/engine/v3" - . "github.com/Monibuca/utils/v3" - "github.com/Monibuca/utils/v3/codec" - "github.com/aler9/gortsplib" - "github.com/aler9/gortsplib/pkg/aac" - "github.com/aler9/gortsplib/pkg/base" - "github.com/pion/rtp" - "github.com/pion/rtp/codecs" -) - -type RTSPClient struct { - RTSPublisher - Transport gortsplib.Transport - *gortsplib.Client `json:"-"` -} - -// PullStream 从外部拉流 -func (rtsp *RTSPClient) PullStream(streamPath string, rtspUrl string) (err error) { - rtsp.Stream = &Stream{ - StreamPath: streamPath, - Type: "RTSP Pull", - ExtraProp: rtsp, - } - if result := rtsp.Publish(); result { - rtsp.URL = rtspUrl - if config.Reconnect { - go func() { - for rtsp.pullStream(); rtsp.Err() == nil; rtsp.pullStream() { - Printf("reconnecting:%s in 5 seconds", rtspUrl) - if rtsp.Transport == gortsplib.TransportTCP { - rtsp.Transport = gortsplib.TransportUDP - } else { - rtsp.Transport = gortsplib.TransportTCP - } - time.Sleep(time.Second * 5) - } - if rtsp.IsTimeout { - go rtsp.PullStream(streamPath, rtspUrl) - } - }() - } else { - go rtsp.pullStream() - } - return - } - return errors.New("publish badname") -} -func (rtsp *RTSPClient) PushStream(streamPath string, rtspUrl string) (err error) { - if s := FindStream(streamPath); s != nil { - var tracks gortsplib.Tracks - var sub RTSPSubscriber - sub.Type = "RTSP push out" - sub.vt = s.WaitVideoTrack("h264", "h265") - sub.at = s.WaitAudioTrack("aac", "pcma", "pcmu") - ssrc := uintptr(unsafe.Pointer(&sub)) - var trackIds = 0 - if sub.vt != nil { - trackId := trackIds - var vtrack *gortsplib.Track - var vpacketer rtp.Packetizer - switch sub.vt.CodecID { - case codec.CodecID_H264: - if vtrack, err = gortsplib.NewTrackH264(96, &gortsplib.TrackConfigH264{ - SPS: sub.vt.ExtraData.NALUs[0], - PPS: sub.vt.ExtraData.NALUs[1], - }); err == nil { - vpacketer = rtp.NewPacketizer(1200, 96, uint32(ssrc), &codecs.H264Payloader{}, rtp.NewFixedSequencer(1), 90000) - } else { - return err - } - case codec.CodecID_H265: - vtrack = NewH265Track(96, sub.vt.ExtraData.NALUs) - vpacketer = rtp.NewPacketizer(1200, 96, uint32(ssrc), &H265Payloader{}, rtp.NewFixedSequencer(1), 90000) - } - var st uint32 - onVideo := func(ts uint32, pack *VideoPack) { - for _, nalu := range pack.NALUs { - for _, pack := range vpacketer.Packetize(nalu, (ts-st)*90) { - rtp, _ := pack.Marshal() - rtsp.WritePacketRTP(trackId, rtp) - } - } - st = ts - } - sub.OnVideo = func(ts uint32, pack *VideoPack) { - if st = ts; st != 0 { - sub.OnVideo = onVideo - } - onVideo(ts, pack) - } - tracks = append(tracks, vtrack) - trackIds++ - } - if sub.at != nil { - var st uint32 - trackId := trackIds - switch sub.at.CodecID { - case codec.CodecID_PCMA, codec.CodecID_PCMU: - atrack := NewG711Track(97, map[byte]string{7: "pcma", 8: "pcmu"}[sub.at.CodecID]) - apacketizer := rtp.NewPacketizer(1200, 97, uint32(ssrc), &codecs.G711Payloader{}, rtp.NewFixedSequencer(1), 8000) - sub.OnAudio = func(ts uint32, pack *AudioPack) { - for _, pack := range apacketizer.Packetize(pack.Raw, (ts-st)*8) { - buf, _ := pack.Marshal() - rtsp.WritePacketRTP(trackId, buf) - } - st = ts - } - tracks = append(tracks, atrack) - case codec.CodecID_AAC: - var mpegConf aac.MPEG4AudioConfig - mpegConf.Decode(sub.at.ExtraData[2:]) - conf := &gortsplib.TrackConfigAAC{ - Type: int(mpegConf.Type), - SampleRate: mpegConf.SampleRate, - ChannelCount: mpegConf.ChannelCount, - AOTSpecificConfig: mpegConf.AOTSpecificConfig, - } - if atrack, err := gortsplib.NewTrackAAC(97, conf); err == nil { - apacketizer := rtp.NewPacketizer(1200, 97, uint32(ssrc), &AACPayloader{}, rtp.NewFixedSequencer(1), uint32(mpegConf.SampleRate)) - sub.OnAudio = func(ts uint32, pack *AudioPack) { - for _, pack := range apacketizer.Packetize(pack.Raw, (ts-st)*uint32(mpegConf.SampleRate)/1000) { - buf, _ := pack.Marshal() - rtsp.WritePacketRTP(trackId, buf) - } - st = ts - } - tracks = append(tracks, atrack) - } - } - } - return rtsp.StartPublishing(rtspUrl, tracks) - } - return errors.New("stream not exist") -} -func (client *RTSPClient) pullStream() { - if client.Err() != nil { - return - } - client.Client = &gortsplib.Client{ - OnPacketRTP: func(trackID int, payload []byte) { - // Println("OnPacketRTP", trackID, len(payload)) - if f := client.processFunc[trackID]; f != nil { - var clone []byte - f(append(clone, payload...)) - } - }, - ReadBufferSize: config.ReadBufferSize, - Transport: &client.Transport, - } - // 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.OnClose = func() { - client.Client.Close() - } - //client.close should be after connected! - defer client.Client.Close() - 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", client.URL, err) - return - } - Println(res) - if client.processFunc == nil { - client.setTracks(tracks) - } - for _, track := range tracks { - if res, err = client.Setup(true, track, baseURL, 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 - var fatalChan = make(chan error) - go func() { - fatalChan <- client.Wait() - }() - select { - case err := <-fatalChan: - Printf("Wait:%s error:%v", baseURL.String(), err) - case <-client.Done(): - Printf("client:%s done", client.URL) - } -} diff --git a/go.mod b/go.mod index 07044dd..f79bce9 100644 --- a/go.mod +++ b/go.mod @@ -1,11 +1,40 @@ -module github.com/Monibuca/plugin-rtsp/v3 +module m7s.live/plugin/rtsp/v4 -go 1.16 +go 1.18 require ( - github.com/Monibuca/engine/v3 v3.4.1 - github.com/Monibuca/utils/v3 v3.0.5 - github.com/aler9/gortsplib v0.0.0-20211212220644-6f374e396529 - github.com/pion/rtp v1.7.4 - github.com/pion/sdp/v3 v3.0.4 + github.com/aler9/gortsplib v0.0.0-20220315115637-d1cd6357f958 + github.com/pion/sdp/v3 v3.0.2 + m7s.live/engine/v4 v4.0.0-alpha8 +) + +require ( + github.com/cnotch/ipchub v1.1.0 // indirect + github.com/go-ole/go-ole v1.2.6 // indirect + github.com/gobwas/httphead v0.1.0 // indirect + github.com/gobwas/pool v0.2.1 // indirect + github.com/gobwas/ws v1.1.0 // indirect + github.com/google/uuid v1.3.0 // indirect + github.com/icza/bitio v1.0.0 // indirect + github.com/kr/text v0.2.0 // indirect + github.com/logrusorgru/aurora v2.0.3+incompatible // indirect + github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect + github.com/pion/randutil v0.1.0 // indirect + github.com/pion/rtcp v1.2.9 // indirect + github.com/pion/rtp v1.7.4 // indirect + github.com/pion/rtp/v2 v2.0.0-20220302185659-b3d10fc096b0 // indirect + github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect + github.com/q191201771/naza v0.19.1 // indirect + github.com/rogpeppe/go-internal v1.8.1 // indirect + github.com/shirou/gopsutil/v3 v3.22.1 // indirect + github.com/tklauser/go-sysconf v0.3.9 // indirect + github.com/tklauser/numcpus v0.3.0 // indirect + github.com/yusufpapurcu/wmi v1.2.2 // indirect + go.uber.org/atomic v1.7.0 // indirect + go.uber.org/multierr v1.6.0 // indirect + go.uber.org/zap v1.21.0 // indirect + golang.org/x/net v0.0.0-20210610132358-84b48f89b13b // indirect + golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect + golang.org/x/sys v0.0.0-20220111092808-5a964db01320 // indirect + gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect ) diff --git a/go.sum b/go.sum index f23a8be..a478b38 100644 --- a/go.sum +++ b/go.sum @@ -1,14 +1,10 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= -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.4.1 h1:Ap2VbwTkMUkv80NPeUX2sNdV5Vz5nPVoU/6RU51PSAc= -github.com/Monibuca/engine/v3 v3.4.1/go.mod h1:rgAUey5ziRhlh6WugWyA5fYKyGOvcwhtTMDk4sukE7E= -github.com/Monibuca/utils/v3 v3.0.5 h1:w14x0HkWTbF4MmHbINLlOwe4VJNoSOeaQChMk5E/4es= -github.com/Monibuca/utils/v3 v3.0.5/go.mod h1:RpNS95gapWs6gimwh8Xn2x72FN5tO7Powabj7dTFyvE= -github.com/aler9/gortsplib v0.0.0-20211212220644-6f374e396529 h1:j2tfs+eUubyZnuwmYWzK+IS681IixfUyD8bivz4sqAw= -github.com/aler9/gortsplib v0.0.0-20211212220644-6f374e396529/go.mod h1:fyQrQyHo8QvdR/h357tkv1g36VesZlzEPsdAu2VrHHc= +github.com/aler9/gortsplib v0.0.0-20220315115637-d1cd6357f958 h1:e0pC+89s3vKoSQYuRTAN7Ti+N7XASi4XtWYz1fhF6yY= +github.com/aler9/gortsplib v0.0.0-20220315115637-d1cd6357f958/go.mod h1:4HE78w95Rqw1B2T90CHwtA4xBPPCRZ7/G8ds8ZdWcFk= 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/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8= +github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= github.com/cnotch/apirouter v0.0.0-20200731232942-89e243a791f3/go.mod h1:5deJPLON/x/s2dLOQfuKS0lenhOIT4xX0pvtN/OEIuY= github.com/cnotch/ipchub v1.1.0 h1:hH0lh2mU3AZXPiqMwA0pdtqrwo7PFIMRGush9OobMUs= github.com/cnotch/ipchub v1.1.0/go.mod h1:2PbeBs2q2VxxTVCn1eYCDwpAWuVXbq1+N0FU7GimOH4= @@ -17,15 +13,25 @@ github.com/cnotch/queue v0.0.0-20200326024423-6e88bdbf2ad4/go.mod h1:zOssjAlNusO github.com/cnotch/queue v0.0.0-20201224060551-4191569ce8f6/go.mod h1:zOssjAlNusOxvtaqT+EMA+Iyi8rrtKr4/XfzN1Fgoeg= github.com/cnotch/scheduler v0.0.0-20200522024700-1d2da93eefc5/go.mod h1:F4GE3SZkJZ8an1Y0ZCqvSM3jeozNuKzoC67erG1PhIo= github.com/cnotch/xlog v0.0.0-20201208005456-cfda439cd3a0/go.mod h1:RW9oHsR79ffl3sR3yMGgxYupMn2btzdtJUwoxFPUE5E= -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= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/emitter-io/address v1.0.0/go.mod h1:GfZb5+S/o8694B1GMGK2imUYQyn2skszMvGNA5D84Ug= github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= -github.com/funny/slab v0.0.0-20180511031532-b1fad5e5d478 h1:Db9StoJ6RZN3YttC0Pm0I4Y5izITRYch3RMbT59BYN0= -github.com/funny/slab v0.0.0-20180511031532-b1fad5e5d478/go.mod h1:0j1+svBH8ABEIPdUP0AIg4qedsybnXGJBakCEw8cfoo= -github.com/funny/utest v0.0.0-20161029064919-43870a374500 h1:Z0r1CZnoIWFB/Uiwh1BU5FYmuFe6L5NPi6XWQEmsTRg= -github.com/funny/utest v0.0.0-20161029064919-43870a374500/go.mod h1:mUn39tBov9jKnTWV1RlOYoNzxdBFHiSzXWdY1FoNGGg= +github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= +github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= +github.com/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU= +github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM= +github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og= +github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw= +github.com/gobwas/ws v1.1.0 h1:7RFti/xnNkMJnrK7D1yQ/iCIB5OrrY/54/H930kIbHA= +github.com/gobwas/ws v1.1.0/go.mod h1:nzvNcVha5eUziGrbxFCo6qFIojQHjJV5cLYIbezhfL0= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o= +github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE= +github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= +github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= github.com/icza/bitio v1.0.0 h1:squ/m1SHyFeCA6+6Gyol1AxV9nmPPlJFT8c2vKdj3U8= @@ -35,67 +41,124 @@ github.com/icza/mighty v0.0.0-20180919140131-cfd07d671de6/go.mod h1:xQig96I1VNBD github.com/kelindar/process v0.0.0-20170730150328-69a29e249ec3/go.mod h1:+lTCLnZFXOkqwD8sLPl6u4erAc0cP8wFegQHfipz7KE= github.com/kelindar/rate v1.0.0/go.mod h1:AjT4G+hTItNwt30lucEGZIz8y7Uk5zPho6vurIZ+1Es= github.com/kelindar/tcp v1.0.0/go.mod h1:JB5hj1cshLU60XrLij2BBxW3JQ4hOye8vqbyvuKb52k= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/logrusorgru/aurora v2.0.3+incompatible h1:tOpm7WcpBTn4fjmVfgpQq0EfczGlG91VSDkswnjF5A8= github.com/logrusorgru/aurora v2.0.3+incompatible/go.mod h1:7rIyQOR62GCctdiQpZ/zOJlFyk6y+94wXzv6RNZgaR4= -github.com/mattn/go-colorable v0.1.8 h1:c1ghPdyEDarC70ftn0y+A/Ee++9zz8ljHG1b13eJ0s8= -github.com/mattn/go-colorable v0.1.8/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc= -github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY= -github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY= github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= -github.com/pion/rtcp v1.2.4 h1:NT3H5LkUGgaEapvp0HGik+a+CpflRF7KTD7H+o7OWIM= -github.com/pion/rtcp v1.2.4/go.mod h1:52rMNPWFsjr39z9B9MhnkqhPLoeHTv1aN63o/42bWE0= -github.com/pion/rtp v1.6.1/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko= +github.com/pion/rtcp v1.2.9 h1:1ujStwg++IOLIEoOiIQ2s+qBuJ1VN81KW+9pMPsif+U= +github.com/pion/rtcp v1.2.9/go.mod h1:qVPhiCzAm4D/rxb6XzKeyZiQK69yJpbUDJSF7TgrqNo= github.com/pion/rtp v1.6.2/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko= 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/rtp/v2 v2.0.0-20220302185659-b3d10fc096b0 h1:zyOGxHutZ6IhksQSMtwf3OFXB29W5R18yFQWOQJYWjU= +github.com/pion/rtp/v2 v2.0.0-20220302185659-b3d10fc096b0/go.mod h1:Vj+rrFbJCT3yxqE/VSwaOo9DQ2pMKGPxuE7hplGOlOs= +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= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= +github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= +github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/profile v1.4.0/go.mod h1:NWz/XGvpEW1FyYQ7fCx4dqYBLlfTcE+A9FLAkNKqjFE= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw= +github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= github.com/q191201771/naza v0.19.1 h1:4KLcxT2CHztO+7miPRtBG3FFgadSQYQw1gPPPKN7rnY= github.com/q191201771/naza v0.19.1/go.mod h1:5LeGupZZFtYP1g/S203n9vXoUNVdlRnPIfM6rExjqt0= +github.com/rogpeppe/go-internal v1.8.1 h1:geMPLpDpQOgVyCg5z5GoRwLHepNdb71NXb67XFkP+Eg= +github.com/rogpeppe/go-internal v1.8.1/go.mod h1:JeRgkft04UBgHMgCIwADu4Pn6Mtm5d4nPKWu0nJ5d+o= +github.com/shirou/gopsutil/v3 v3.22.1 h1:33y31Q8J32+KstqPfscvFwBlNJ6xLaBy4xqBXzlYV5w= +github.com/shirou/gopsutil/v3 v3.22.1/go.mod h1:WapW1AOOPlHyXr+yOyw3uYx36enocrtSoSBy0L5vUHY= github.com/sqs/goreturns v0.0.0-20181028201513-538ac6014518/go.mod h1:CKI4AZ4XmGV240rTHfO0hfE83S6/a3/Q1siZJ/vXf7A= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= -github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0= github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/tklauser/go-sysconf v0.3.9 h1:JeUVdAOWhhxVcU6Eqr/ATFHgXk/mmiItdKeJPev3vTo= +github.com/tklauser/go-sysconf v0.3.9/go.mod h1:11DU/5sG7UexIrp/O6g35hrWzu0JxlwQ3LSFUzyeuhs= +github.com/tklauser/numcpus v0.3.0 h1:ILuRUQBtssgnxw0XXIjKUC56fgnOrFoQQ/4+DeU2biQ= +github.com/tklauser/numcpus v0.3.0/go.mod h1:yFGUr7TUHQRAhyqBcEg0Ge34zDBAsIvJJcyE6boqnA8= +github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= +github.com/yusufpapurcu/wmi v1.2.2 h1:KBNDSne4vP5mbSWnJbO+51IMOXJB67QiYCSBrubbPRg= +github.com/yusufpapurcu/wmi v1.2.2/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw= +go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= +go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI= +go.uber.org/goleak v1.1.11/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ= +go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4= +go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= +go.uber.org/zap v1.21.0 h1:WefMeulhovoZ2sYXz7st6K0sLj7bBhpiFaud4r4zST8= +go.uber.org/zap v1.21.0/go.mod h1:wjWOCqI0f2ZZrJF/UufIOkiC8ii6tm1iqIsLo76RfJw= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I= +golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= +golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20210610132358-84b48f89b13b h1:k+E048sYJHyVnsr1GDrRZWQ32D2C7lWs9JRc0bel53A= golang.org/x/net v0.0.0-20210610132358-84b48f89b13b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20201207232520-09787c993a3a h1:DcqTD9SDLc+1P/r1EmRBwnVsrOwW+kk2vWf9n+1sGhs= -golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ= +golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20210423082822-04245dca01da h1:b3NXsE2LusjYGGjL5bxEVZZORm/YEFFrWFjR8eFrw/c= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201207223542-d4d67f95c62d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20210816074244-15123e1e1f71/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220111092808-5a964db01320 h1:0jf+tOCoZ3LyutmCOWpVni1chK4VfFLhRsDK7MhqGRY= +golang.org/x/sys v0.0.0-20220111092808-5a964db01320/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= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= +gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI= gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys= gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k= gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw= gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= -gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo= +gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo= +gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +m7s.live/engine/v4 v4.0.0-alpha8 h1:reb4Agbrmc22cO6o6Momeew1D/O93OnhbvbJrwEEWM8= +m7s.live/engine/v4 v4.0.0-alpha8/go.mod h1:+3MSMe8XFiriAB9dtfJ1gn+mkAIcopF+xrAf2WYPp08= diff --git a/main.go b/main.go index 5bc7b52..db7f2df 100644 --- a/main.go +++ b/main.go @@ -1,113 +1,44 @@ package rtsp import ( - "encoding/json" - "fmt" - "log" - "net/http" - "time" + "sync" - . "github.com/Monibuca/engine/v3" - . "github.com/Monibuca/utils/v3" "github.com/aler9/gortsplib" + . "m7s.live/engine/v4" + "m7s.live/engine/v4/config" ) -var config = struct { +type RTSPConfig struct { + config.Publish + config.Subscribe + config.Pull + config.Push ListenAddr string UDPAddr string RTCPAddr string - Timeout int - Reconnect bool - AutoPullList map[string]string - AutoPushList map[string]string ReadBufferSize int -}{":554", ":8000", ":8001", 0, false, nil, nil, 2048} - -var pconfig = PluginConfig{ - Name: "RTSP", - Config: &config, + sync.Map } -func init() { - pconfig.Install(runPlugin) -} - -func getRtspList() (info []*Stream) { - for _, s := range Streams.ToList() { - switch s.ExtraProp.(type) { - case *RTSPublisher: - info = append(info, s) - case *RTSPClient: - info = append(info, s) +func (conf *RTSPConfig) OnEvent(event any) { + switch event.(type) { + case FirstConfig: + s := &gortsplib.Server{ + Handler: conf, + RTSPAddress: conf.ListenAddr, + UDPRTPAddress: conf.UDPAddr, + UDPRTCPAddress: conf.RTCPAddr, + MulticastIPRange: "224.1.0.0/16", + MulticastRTPPort: 8002, + MulticastRTCPPort: 8003, } - } - return -} -func runPlugin() { - http.HandleFunc("/api/rtsp/list", func(w http.ResponseWriter, r *http.Request) { - CORS(w, r) - if r.URL.Query().Get("json") != "" { - if jsonData, err := json.Marshal(getRtspList()); err == nil { - w.Write(jsonData) - } else { - w.WriteHeader(500) - } - return - } - sse := NewSSE(w, r.Context()) - var err error - for tick := time.NewTicker(time.Second); err == nil; <-tick.C { - err = sse.WriteJSON(getRtspList()) - } - }) - http.HandleFunc("/api/rtsp/pull", func(w http.ResponseWriter, r *http.Request) { - CORS(w, r) - targetURL := r.URL.Query().Get("target") - streamPath := r.URL.Query().Get("streamPath") - save := r.URL.Query().Get("save") - if err := (&RTSPClient{Transport: gortsplib.TransportTCP}).PullStream(streamPath, targetURL); err == nil { - if save == "1" { - if config.AutoPullList == nil { - config.AutoPullList = make(map[string]string) - } - config.AutoPullList[streamPath] = targetURL - if err := pconfig.Save(); err != nil { - Println(err) - } - } - w.Write([]byte(`{"code":0}`)) - } else { - w.Write([]byte(fmt.Sprintf(`{"code":1,"msg":"%s"}`, err.Error()))) - } - }) - for streamPath, url := range config.AutoPullList { - if err := (&RTSPClient{Transport: gortsplib.TransportTCP}).PullStream(streamPath, url); err != nil { - Println(err) - } - } - - go AddHook(HOOK_PUBLISH, func(s *Stream) { - for streamPath, url := range config.AutoPushList { - if s.StreamPath == streamPath { - (&RTSPClient{}).PushStream(streamPath, url) - } - } - }) - if config.ListenAddr != "" { - go log.Fatal(ListenRtsp(config.ListenAddr)) + s.Start() } } -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() -} +var plugin = InstallPlugin(&RTSPConfig{ + ListenAddr: ":554", + UDPAddr: ":8000", + RTCPAddr: ":8001", + ReadBufferSize: 2048, +}) diff --git a/payloader.go b/payloader.go deleted file mode 100644 index d408dcb..0000000 --- a/payloader.go +++ /dev/null @@ -1,24 +0,0 @@ -package rtsp - -// AACayloader payloads AAC packets -type AACPayloader struct{} - -// Payload fragments an AAC packet across one or more byte arrays -func (p *AACPayloader) Payload(mtu uint16, payload []byte) [][]byte { - var out [][]byte - o := make([]byte, len(payload)+4) - //AU_HEADER_LENGTH,因为单位是bit, 除以8就是auHeader的字节长度;又因为单个auheader字节长度2字节,所以再除以2就是auheader的个数。 - o[0] = 0x00 //高位 - o[1] = 0x10 //低位 - //AU_HEADER - o[2] = (byte)((len(payload) & 0x1fe0) >> 5) //高位 - o[3] = (byte)((len(payload) & 0x1f) << 3) //低位 - copy(o[4:], payload) - return append(out, o) -} - -type H265Payloader struct{} - -func (p *H265Payloader) Payload(mtu uint16, payload []byte) [][]byte { - return nil -} diff --git a/publisher.go b/publisher.go deleted file mode 100644 index c2fef6d..0000000 --- a/publisher.go +++ /dev/null @@ -1,121 +0,0 @@ -package rtsp - -import ( - "encoding/base64" - "encoding/hex" - "strconv" - "strings" - - . "github.com/Monibuca/engine/v3" - . "github.com/Monibuca/utils/v3" - "github.com/aler9/gortsplib" -) - -type RTSPublisher struct { - *Stream `json:"-"` - stream *gortsplib.ServerStream - processFunc []func([]byte) -} - -func (p *RTSPublisher) setTracks(tracks gortsplib.Tracks) { - if p.processFunc != nil { - p.processFunc = p.processFunc[:len(tracks)] - return - } else { - p.processFunc = make([]func([]byte), len(tracks)) - } - 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 { - for _, kv := range strings.Split(tmp[1], ";") { - kv = strings.Trim(kv, " ") - - if len(kv) == 0 { - continue - } - tmp := strings.SplitN(kv, "=", 2) - if len(tmp) == 2 { - fmtp[strings.TrimSpace(tmp[0])] = tmp[1] - } - } - } - } - - v = strings.TrimSpace(v) - vals := strings.Split(v, " ") - if len(vals) != 2 { - continue - } - timeScale := 0 - keyval := strings.Split(vals[1], "/") - if i, err := strconv.Atoi(keyval[1]); err == nil { - timeScale = i - } - if len(keyval) >= 2 { - Printf("track %d is %s", i, keyval[0]) - switch strings.ToLower(keyval[0]) { - case "h264": - vt := p.NewRTPVideo(7) - if conf, err := track.ExtractConfigH264(); err == nil { - vt.PushNalu(0, 0, conf.SPS, conf.PPS) - } - p.processFunc[i] = vt.Push - case "h265", "hevc": - vt := p.NewRTPVideo(12) - if v, ok := fmtp["sprop-vps"]; ok { - vps, _ := base64.StdEncoding.DecodeString(v) - vt.PushNalu(0, 0, vps) - } - if v, ok := fmtp["sprop-sps"]; ok { - sps, _ := base64.StdEncoding.DecodeString(v) - vt.PushNalu(0, 0, sps) - } - if v, ok := fmtp["sprop-pps"]; ok { - pps, _ := base64.StdEncoding.DecodeString(v) - vt.PushNalu(0, 0, pps) - } - p.processFunc[i] = vt.Push - case "pcma": - at := p.NewRTPAudio(7) - at.SoundRate = timeScale - at.SoundSize = 16 - if len(keyval) >= 3 { - x, _ := strconv.Atoi(keyval[2]) - at.Channels = byte(x) - } else { - at.Channels = 1 - } - at.ExtraData = []byte{(at.CodecID << 4) | (1 << 1)} - p.processFunc[i] = at.Push - case "pcmu": - at := p.NewRTPAudio(8) - at.SoundRate = timeScale - at.SoundSize = 16 - if len(keyval) >= 3 { - x, _ := strconv.Atoi(keyval[2]) - at.Channels = byte(x) - } else { - at.Channels = 1 - } - at.ExtraData = []byte{(at.CodecID << 4) | (1 << 1)} - p.processFunc[i] = at.Push - case "mpeg4-generic": - at := p.NewRTPAudio(0) - if config, ok := fmtp["config"]; ok { - asc, _ := hex.DecodeString(config) - at.SetASC(asc) - } else { - Println("aac no config") - } - at.SoundSize = 16 - p.processFunc[i] = at.Push - } - } - } -} diff --git a/server.go b/server.go index 857c1e9..c9b967c 100644 --- a/server.go +++ b/server.go @@ -1,264 +1,72 @@ package rtsp import ( - "fmt" - "sync" - "unsafe" - - "github.com/Monibuca/engine/v3" - . "github.com/Monibuca/utils/v3" - "github.com/Monibuca/utils/v3/codec" "github.com/aler9/gortsplib" - "github.com/aler9/gortsplib/pkg/aac" "github.com/aler9/gortsplib/pkg/base" - "github.com/pion/rtp" - "github.com/pion/rtp/codecs" + . "m7s.live/engine/v4" ) -// 接收RTSP推流:OnConnOpen->OnAnnounce->OnSetup->OnSessionOpen -// 接收RTSP拉流:OnConnOpen->OnDescribe->OnSetup->OnSessionOpen -type RTSPServer struct { - sync.Map -} -type RTSPSubscriber struct { - stream *gortsplib.ServerStream - engine.Subscriber - vt *engine.VideoTrack - at *engine.AudioTrack +func (conf *RTSPConfig) OnConnOpen(ctx *gortsplib.ServerHandlerOnConnOpenCtx) { + plugin.Debug("conn opened") } -// called after a connection is opened. -func (sh *RTSPServer) OnConnOpen(ctx *gortsplib.ServerHandlerOnConnOpenCtx) { - Printf("rtsp conn opened") -} - -// called after a connection is closed. -func (sh *RTSPServer) OnConnClose(ctx *gortsplib.ServerHandlerOnConnCloseCtx) { - Printf("rtsp conn closed (%v)", ctx.Error) - if p, ok := sh.Load(ctx.Conn); ok { - switch v := p.(type) { - case *RTSPublisher: - v.Close() - case *RTSPSubscriber: - v.Close() - } - sh.Delete(ctx.Conn) +func (conf *RTSPConfig) OnConnClose(ctx *gortsplib.ServerHandlerOnConnCloseCtx) { + plugin.Debug("conn closed") + if p, ok := conf.LoadAndDelete(ctx.Conn); ok { + p.(IIO).Stop() } } -// called after a session is opened. -func (sh *RTSPServer) OnSessionOpen(ctx *gortsplib.ServerHandlerOnSessionOpenCtx) { - Printf("rtsp session opened") +func (conf *RTSPConfig) OnSessionOpen(ctx *gortsplib.ServerHandlerOnSessionOpenCtx) { + plugin.Debug("session opened") } -// called after a session is closed. -func (sh *RTSPServer) OnSessionClose(ctx *gortsplib.ServerHandlerOnSessionCloseCtx) { - Printf("rtsp session closed") - if v, ok := sh.LoadAndDelete(ctx.Session); ok { - switch v := v.(type) { - case *RTSPublisher: - v.Close() - case *RTSPSubscriber: - v.Close() - } +func (conf *RTSPConfig) OnSessionClose(ctx *gortsplib.ServerHandlerOnSessionCloseCtx) { + plugin.Debug("session closed") + if p, ok := conf.LoadAndDelete(ctx.Session); ok { + p.(IIO).Stop() } } // called after receiving a DESCRIBE request. -func (sh *RTSPServer) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (*base.Response, *gortsplib.ServerStream, error) { - Printf("describe request") - var err error - if s := engine.FindStream(ctx.Path); s != nil { - var tracks gortsplib.Tracks - var stream *gortsplib.ServerStream - var sub RTSPSubscriber - sub.Type = "RTSP pull" - sub.vt = s.WaitVideoTrack("h264", "h265") - sub.at = s.WaitAudioTrack("aac", "pcma", "pcmu") - ssrc := uintptr(unsafe.Pointer(&stream)) - var trackIds = 0 - if sub.vt != nil { - trackId := trackIds - var vtrack *gortsplib.Track - var vpacketer rtp.Packetizer - switch sub.vt.CodecID { - case codec.CodecID_H264: - if vtrack, err = gortsplib.NewTrackH264(96, &gortsplib.TrackConfigH264{ - SPS: sub.vt.ExtraData.NALUs[0], - PPS: sub.vt.ExtraData.NALUs[1], - }); err == nil { - vpacketer = rtp.NewPacketizer(1200, 96, uint32(ssrc), &codecs.H264Payloader{}, rtp.NewFixedSequencer(1), 90000) - } else { - return nil, nil, err - } - case codec.CodecID_H265: - vtrack = NewH265Track(96, sub.vt.ExtraData.NALUs) - vpacketer = rtp.NewPacketizer(1200, 96, uint32(ssrc), &H265Payloader{}, rtp.NewFixedSequencer(1), 90000) - } - var st uint32 - onVideo := func(ts uint32, pack *engine.VideoPack) { - for i, nalu := range pack.NALUs { - var samples uint32 - if i == len(pack.NALUs)-1 { - samples = (ts - st) * 90 - } else { - samples = 0 - } - packs := vpacketer.Packetize(nalu, samples) - for j, rtpack := range packs { - rtpack.Marker = i == len(pack.NALUs)-1 && j == len(packs)-1 - rtp, _ := rtpack.Marshal() - stream.WritePacketRTP(trackId, rtp) - } - } - st = ts - } - sub.OnVideo = func(ts uint32, pack *engine.VideoPack) { - if st = ts; st != 0 { - sub.OnVideo = onVideo - } - onVideo(ts, pack) - } - tracks = append(tracks, vtrack) - trackIds++ - } - if sub.at != nil { - var st uint32 - trackId := trackIds - switch sub.at.CodecID { - case codec.CodecID_PCMA, codec.CodecID_PCMU: - atrack := NewG711Track(97, map[byte]string{7: "pcma", 8: "pcmu"}[sub.at.CodecID]) - apacketizer := rtp.NewPacketizer(1200, 97, uint32(ssrc), &codecs.G711Payloader{}, rtp.NewFixedSequencer(1), 8000) - sub.OnAudio = func(ts uint32, pack *engine.AudioPack) { - for _, pack := range apacketizer.Packetize(pack.Raw, (ts-st)*8) { - buf, _ := pack.Marshal() - stream.WritePacketRTP(trackId, buf) - } - st = ts - } - tracks = append(tracks, atrack) - case codec.CodecID_AAC: - var mpegConf aac.MPEG4AudioConfig - mpegConf.Decode(sub.at.ExtraData[2:]) - conf := &gortsplib.TrackConfigAAC{ - Type: int(mpegConf.Type), - SampleRate: mpegConf.SampleRate, - ChannelCount: mpegConf.ChannelCount, - AOTSpecificConfig: mpegConf.AOTSpecificConfig, - } - if atrack, err := gortsplib.NewTrackAAC(97, conf); err == nil { - apacketizer := rtp.NewPacketizer(1200, 97, uint32(ssrc), &AACPayloader{}, rtp.NewFixedSequencer(1), uint32(mpegConf.SampleRate)) - 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.WritePacketRTP(trackId, buf) - } - st = ts - } - tracks = append(tracks, atrack) - } - } - } - stream = gortsplib.NewServerStream(tracks) - sub.stream = stream - sh.Store(ctx.Conn, &sub) +func (conf *RTSPConfig) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (*base.Response, *gortsplib.ServerStream, error) { + plugin.Debug("describe request") + var suber RTSPSubscriber + if err := plugin.Subscribe(ctx.Path, &suber); err == nil { + conf.Store(ctx.Conn, &suber) return &base.Response{ StatusCode: base.StatusOK, - }, stream, nil - // if stream, ok := s.ExtraProp.(*gortsplib.ServerStream); ok { - // return &base.Response{ - // StatusCode: base.StatusOK, - // }, stream, nil - // } - } - return &base.Response{ - StatusCode: base.StatusNotFound, - }, nil, nil -} - -// called after receiving an ANNOUNCE request. -func (sh *RTSPServer) OnAnnounce(ctx *gortsplib.ServerHandlerOnAnnounceCtx) (*base.Response, error) { - Printf("announce request") - p := &RTSPublisher{ - Stream: &engine.Stream{ - StreamPath: ctx.Path, - Type: "RTSP push", - }, - } - p.ExtraProp = p - p.URL = ctx.Req.URL.String() - if p.Publish() { - p.setTracks(ctx.Tracks) - p.stream = gortsplib.NewServerStream(ctx.Tracks) - sh.Store(ctx.Conn, p) - sh.Store(ctx.Session, p) + }, suber.stream, nil } else { - return &base.Response{ - StatusCode: base.StatusBadRequest, - }, fmt.Errorf("streamPath is already exist") + return nil, nil, err } - return &base.Response{ - StatusCode: base.StatusOK, - }, nil } -// called after receiving a SETUP request. -func (sh *RTSPServer) OnSetup(ctx *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error) { - Printf("setup request") - if p, ok := sh.Load(ctx.Conn); ok { +func (conf *RTSPConfig) OnSetup(ctx *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error) { + var resp base.Response + resp.StatusCode = base.StatusOK + if p, ok := conf.Load(ctx.Conn); ok { switch v := p.(type) { - case *RTSPublisher: - return &base.Response{ - StatusCode: base.StatusOK, - }, v.stream, nil case *RTSPSubscriber: - return &base.Response{ - StatusCode: base.StatusOK, - }, v.stream, nil + return &resp, v.stream, nil } } - return &base.Response{ - StatusCode: base.StatusNotFound, - }, nil, nil + resp.StatusCode = base.StatusNotFound + return &resp, nil, nil } -// called after receiving a PLAY request. -func (sh *RTSPServer) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) { - Printf("play request") - if p, ok := sh.Load(ctx.Conn); ok { - if sub := p.(*RTSPSubscriber); sub.Subscribe(ctx.Path) == nil { +func (conf *RTSPConfig) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) { + var resp base.Response + resp.StatusCode = base.StatusNotFound + if p, ok := conf.Load(ctx.Conn); ok { + switch v := p.(type) { + case *RTSPSubscriber: + resp.StatusCode = base.StatusOK go func() { - sub.Play(sub.at, sub.vt) + v.PlayBlock(v) ctx.Conn.Close() }() - return &base.Response{ - StatusCode: base.StatusOK, - }, nil - } - } - return &base.Response{ - StatusCode: base.StatusNotFound, - }, nil -} - -// called after receiving a RECORD request. -func (sh *RTSPServer) OnRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.Response, error) { - Printf("record request") - return &base.Response{ - StatusCode: base.StatusOK, - }, nil -} - -// called after receiving a frame. -func (sh *RTSPServer) OnPacketRTP(ctx *gortsplib.ServerHandlerOnPacketRTPCtx) { - if p, ok := sh.Load(ctx.Session); ok { - rtsp := p.(*RTSPublisher) - if rtsp.Err() != nil { - ctx.Session.Close() - return - } - if f := rtsp.processFunc[ctx.TrackID]; f != nil { - f(ctx.Payload) } } + return &resp, nil } diff --git a/subscriber.go b/subscriber.go new file mode 100644 index 0000000..8a6e305 --- /dev/null +++ b/subscriber.go @@ -0,0 +1,63 @@ +package rtsp + +import ( + "github.com/aler9/gortsplib" + "github.com/aler9/gortsplib/pkg/aac" + . "m7s.live/engine/v4" + "m7s.live/engine/v4/codec" + "m7s.live/engine/v4/track" +) + +type RTSPSubscriber struct { + tracks gortsplib.Tracks + stream *gortsplib.ServerStream + Subscriber + audioTrackId int + videoTrackId int +} + +func (s *RTSPSubscriber) OnEvent(event any) { + switch v := event.(type) { + case *track.Video: + switch v.CodecID { + case codec.CodecID_H264: + extra := v.GetDecoderConfiguration().Raw + if vtrack, err := gortsplib.NewTrackH264(96, extra[0], extra[1], nil); err == nil { + s.videoTrackId = len(s.tracks) + s.tracks = append(s.tracks, vtrack) + } + case codec.CodecID_H265: + if vtrack, err := NewH265Track(96, v.GetDecoderConfiguration().Raw); err == nil { + s.videoTrackId = len(s.tracks) + s.tracks = append(s.tracks, vtrack) + } + } + case *track.Audio: + switch v.CodecID { + case codec.CodecID_AAC: + var mpegConf aac.MPEG4AudioConfig + mpegConf.Decode(v.GetDecoderConfiguration().Raw) + if atrack, err := gortsplib.NewTrackAAC(97, int(mpegConf.Type), mpegConf.SampleRate, mpegConf.ChannelCount, mpegConf.AOTSpecificConfig); err == nil { + s.audioTrackId = len(s.tracks) + s.tracks = append(s.tracks, atrack) + } + case codec.CodecID_PCMA: + s.audioTrackId = len(s.tracks) + s.tracks = append(s.tracks, NewPCMATrack()) + case codec.CodecID_PCMU: + s.audioTrackId = len(s.tracks) + s.tracks = append(s.tracks, gortsplib.NewTrackPCMU()) + } + case ISubscriber: + s.stream = gortsplib.NewServerStream(s.tracks) + case *AudioFrame: + for _, pack := range v.RTP { + s.stream.WritePacketRTP(s.audioTrackId, &pack.Packet) + } + case *VideoFrame: + for _, pack := range v.RTP { + s.stream.WritePacketRTP(s.videoTrackId, &pack.Packet) + } + } + s.Subscriber.OnEvent(event) +} diff --git a/track.go b/track.go index e41aee8..7a69bd7 100644 --- a/track.go +++ b/track.go @@ -3,85 +3,26 @@ package rtsp import ( "encoding/base64" "fmt" - "strconv" "github.com/aler9/gortsplib" psdp "github.com/pion/sdp/v3" ) -// func NewTrackAAC(payloadType uint8, conf *gortsplib.TrackConfigAAC) (*gortsplib.Track, error) { -// mpegConf, err := aac.MPEG4AudioConfig{ -// Type: aac.MPEG4AudioType(conf.Type), -// SampleRate: conf.SampleRate, -// ChannelCount: conf.ChannelCount, -// AOTSpecificConfig: conf.AOTSpecificConfig, -// }.Encode() -// if err != nil { -// return nil, err -// } +type TrackPCMA struct { + *gortsplib.TrackPCMU +} -// typ := strconv.FormatInt(int64(payloadType), 10) - -// return &gortsplib.Track{ -// Media: &psdp.MediaDescription{ -// MediaName: psdp.MediaName{ -// Media: "audio", -// Protos: []string{"RTP", "AVP"}, -// Formats: []string{typ}, -// }, -// Attributes: []psdp.Attribute{ -// { -// Key: "rtpmap", -// Value: typ + " mpeg4-generic/" + strconv.FormatInt(int64(conf.SampleRate), 10) + -// "/" + strconv.FormatInt(int64(conf.ChannelCount), 10), -// }, -// { -// Key: "fmtp", -// Value: typ + " profile-level-id=1; " + -// "mode=AAC-hbr; " + -// "sizelength=6; " + -// "indexlength=2; " + -// "indexdeltalength=2; " + -// "config=" + hex.EncodeToString(mpegConf), -// }, -// }, -// }, -// }, nil -// } -func NewG711Track(payloadType uint8, law string) *gortsplib.Track { - return &gortsplib.Track{ - Media: &psdp.MediaDescription{ - MediaName: psdp.MediaName{ - Media: "audio", - Protos: []string{"RTP", "AVP"}, - Formats: []string{strconv.FormatInt(int64(payloadType), 10)}}, - Attributes: []psdp.Attribute{ - { - Key: "rtpmap", - Value: fmt.Sprintf("%d %s/8000/1", payloadType, law), - }, - }, - }, +func NewPCMATrack() *TrackPCMA { + return &TrackPCMA{ + gortsplib.NewTrackPCMU(), } } -func NewH265Track(payloadType uint8, sprop [][]byte) *gortsplib.Track { - return &gortsplib.Track{ - Media: &psdp.MediaDescription{ - MediaName: psdp.MediaName{ - Media: "video", - Protos: []string{"RTP", "AVP"}, - Formats: []string{fmt.Sprintf("%d", payloadType)}, - }, - Attributes: []psdp.Attribute{ - { - Key: "rtpmap", - Value: fmt.Sprintf("%d H265/90000", payloadType), - }, - { - Key: "fmtp", - Value: fmt.Sprintf("%d packetization-mode=1;sprop-vps=%s;sprop-sps=%s;sprop-pps=%s;", payloadType, base64.StdEncoding.EncodeToString(sprop[0]), base64.StdEncoding.EncodeToString(sprop[1]), base64.StdEncoding.EncodeToString(sprop[2])), - }, - }, - }, - } +func (t *TrackPCMA) MediaDescription() *psdp.MediaDescription { + md := t.TrackPCMU.MediaDescription() + md.Attributes[0].Value = "0 PCMA/8000" + return md +} + +func NewH265Track(payloadType uint8, sprop [][]byte) (gortsplib.Track, error) { + return gortsplib.NewTrackGeneric("video", []string{fmt.Sprintf("%d", payloadType)}, fmt.Sprintf("%d H265/90000", payloadType), fmt.Sprintf("%d packetization-mode=1;sprop-vps=%s;sprop-sps=%s;sprop-pps=%s;", payloadType, base64.StdEncoding.EncodeToString(sprop[0]), base64.StdEncoding.EncodeToString(sprop[1]), base64.StdEncoding.EncodeToString(sprop[2]))) }