forked from Monibuca/plugin-rtsp
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.go
102 lines (93 loc) · 3.06 KB
/
main.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
package rtsp
import (
"net/http"
"strconv"
"sync"
"github.com/bluenviron/gortsplib/v4"
"go.uber.org/zap"
. "m7s.live/engine/v4"
"m7s.live/engine/v4/config"
"m7s.live/engine/v4/util"
)
type RTSPConfig struct {
config.Publish
config.Subscribe
config.Pull
config.Push
ListenAddr string `default:":554" desc:"rtsp监听地址"`
UDPAddr string `default:":8000" desc:"udp rtp监听地址"`
RTCPAddr string `default:":8001" desc:"udp rtcp监听地址"`
WriteBufferCount int `default:"2048" desc:"rtsp写缓冲区大小"`
SendOptions bool `default:"true" desc:"是否发送options请求"`
sync.Map
server *gortsplib.Server
}
func (conf *RTSPConfig) OnEvent(event any) {
switch v := event.(type) {
case FirstConfig:
conf.server = &gortsplib.Server{
Handler: conf,
RTSPAddress: conf.ListenAddr,
UDPRTPAddress: conf.UDPAddr,
UDPRTCPAddress: conf.RTCPAddr,
MulticastIPRange: "224.1.0.0/16",
MulticastRTPPort: 8002,
MulticastRTCPPort: 8003,
WriteQueueSize: conf.WriteBufferCount,
}
if err := conf.server.Start(); err != nil {
RTSPPlugin.Error("server start", zap.Error(err))
RTSPPlugin.Disabled = true
}
for streamPath, url := range conf.PullOnStart {
if err := RTSPPlugin.Pull(streamPath, url, new(RTSPPuller), 0); err != nil {
RTSPPlugin.Error("pull", zap.String("streamPath", streamPath), zap.String("url", url), zap.Error(err))
}
}
case SEpublish:
if remoteURL := conf.CheckPush(v.Target.Path); remoteURL != "" {
if err := RTSPPlugin.Push(v.Target.Path, remoteURL, new(RTSPPusher), false); err != nil {
RTSPPlugin.Error("push", zap.String("streamPath", v.Target.Path), zap.String("url", remoteURL), zap.Error(err))
}
}
case InvitePublish: //按需拉流
if remoteURL := conf.CheckPullOnSub(v.Target); remoteURL != "" {
if err := RTSPPlugin.Pull(v.Target, remoteURL, new(RTSPPuller), 0); err != nil {
RTSPPlugin.Error("pull", zap.String("streamPath", v.Target), zap.String("url", remoteURL), zap.Error(err))
}
}
}
}
var rtspConfig = &RTSPConfig{}
var RTSPPlugin = InstallPlugin(rtspConfig)
func filterStreams() (ss []*Stream) {
Streams.Range(func(key string, s *Stream) {
switch s.Publisher.(type) {
case *RTSPPublisher, *RTSPPuller:
ss = append(ss, s)
}
})
return
}
func (*RTSPConfig) API_list(w http.ResponseWriter, r *http.Request) {
util.ReturnFetchValue(filterStreams, w, r)
}
func (*RTSPConfig) API_Pull(rw http.ResponseWriter, r *http.Request) {
query := r.URL.Query()
save, _ := strconv.Atoi(query.Get("save"))
err := RTSPPlugin.Pull(query.Get("streamPath"), query.Get("target"), new(RTSPPuller), save)
if err != nil {
util.ReturnError(util.APIErrorQueryParse, err.Error(), rw, r)
} else {
util.ReturnOK(rw, r)
}
}
func (*RTSPConfig) API_Push(rw http.ResponseWriter, r *http.Request) {
query := r.URL.Query()
err := RTSPPlugin.Push(query.Get("streamPath"), query.Get("target"), new(RTSPPusher), query.Has("save"))
if err != nil {
util.ReturnError(util.APIErrorQueryParse, err.Error(), rw, r)
} else {
util.ReturnOK(rw, r)
}
}