apps/core
1
0
Fork 0
mirror of https://github.com/datarhei/core.git synced 2025-09-27 04:16:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

Use path without app as session reference

Browse Source
This commit is contained in:
Ingo Oppermann
2022-08-19 11:24:44 +03:00
parent 0febae3242
commit 9cd132650e
1 changed files with 8 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
rtmp/rtmp.go
View File

@@ -101,15 +101,17 @@ type channel struct {
collector session.Collector collector session.Collector
path string path string
reference string
publisher *client publisher *client
subscriber map[string]*client subscriber map[string]*client
lock sync.RWMutex lock sync.RWMutex
} }
func newChannel(conn *rtmp.Conn, collector session.Collector) *channel { func newChannel(conn *rtmp.Conn, reference string, collector session.Collector) *channel {
ch := &channel{ ch := &channel{
path: conn.URL.Path, path: conn.URL.Path,
reference: reference,
publisher: newClient(conn, conn.URL.Path, collector), publisher: newClient(conn, conn.URL.Path, collector),
subscriber: make(map[string]*client), subscriber: make(map[string]*client),
collector: collector, collector: collector,
@@ -119,8 +121,7 @@ func newChannel(conn *rtmp.Conn, collector session.Collector) *channel {
ip, _, _ := net.SplitHostPort(addr) ip, _, _ := net.SplitHostPort(addr)
if collector.IsCollectableIP(ip) { if collector.IsCollectableIP(ip) {
reference := strings.TrimSuffix(filepath.Base(conn.URL.Path), filepath.Ext(conn.URL.Path)) collector.RegisterAndActivate(ch.path, ch.reference, "publish:"+ch.path, addr)
collector.RegisterAndActivate(conn.URL.Path, reference, "publish:"+conn.URL.Path, addr)
} }
return ch return ch
@@ -144,8 +145,7 @@ func (ch *channel) AddSubscriber(conn *rtmp.Conn) string {
client := newClient(conn, addr, ch.collector) client := newClient(conn, addr, ch.collector)
if ch.collector.IsCollectableIP(ip) { if ch.collector.IsCollectableIP(ip) {
reference := strings.TrimSuffix(filepath.Base(conn.URL.Path), filepath.Ext(conn.URL.Path)) ch.collector.RegisterAndActivate(addr, ch.reference, "play:"+ch.path, addr)
ch.collector.RegisterAndActivate(addr, reference, "play:"+conn.URL.Path, addr)
} }
ch.lock.Lock() ch.lock.Lock()
@@ -437,8 +437,10 @@ func (s *server) handlePublish(conn *rtmp.Conn) {
ch := s.channels[conn.URL.Path] ch := s.channels[conn.URL.Path]
if ch == nil { if ch == nil {
reference := strings.TrimPrefix(strings.TrimSuffix(conn.URL.Path, filepath.Ext(conn.URL.Path)), s.app+"/")
// Create a new channel // Create a new channel
ch = newChannel(conn, s.collector) ch = newChannel(conn, reference, s.collector)
ch.metadata = conn.GetMetaData() ch.metadata = conn.GetMetaData()
ch.queue = pubsub.NewQueue() ch.queue = pubsub.NewQueue()
ch.queue.WriteHeader(streams) ch.queue.WriteHeader(streams)