Alessandro Ros
1 year ago
committed by
GitHub
58 changed files with 2856 additions and 3115 deletions
@ -1,309 +0,0 @@
@@ -1,309 +0,0 @@
|
||||
package core |
||||
|
||||
import ( |
||||
"context" |
||||
"fmt" |
||||
"sort" |
||||
"sync" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/defs" |
||||
"github.com/bluenviron/mediamtx/internal/logger" |
||||
) |
||||
|
||||
type hlsManagerAPIMuxersListRes struct { |
||||
data *defs.APIHLSMuxerList |
||||
err error |
||||
} |
||||
|
||||
type hlsManagerAPIMuxersListReq struct { |
||||
res chan hlsManagerAPIMuxersListRes |
||||
} |
||||
|
||||
type hlsManagerAPIMuxersGetRes struct { |
||||
data *defs.APIHLSMuxer |
||||
err error |
||||
} |
||||
|
||||
type hlsManagerAPIMuxersGetReq struct { |
||||
name string |
||||
res chan hlsManagerAPIMuxersGetRes |
||||
} |
||||
|
||||
type hlsManagerParent interface { |
||||
logger.Writer |
||||
} |
||||
|
||||
type hlsManager struct { |
||||
externalAuthenticationURL string |
||||
alwaysRemux bool |
||||
variant conf.HLSVariant |
||||
segmentCount int |
||||
segmentDuration conf.StringDuration |
||||
partDuration conf.StringDuration |
||||
segmentMaxSize conf.StringSize |
||||
directory string |
||||
writeQueueSize int |
||||
pathManager *pathManager |
||||
parent hlsManagerParent |
||||
|
||||
ctx context.Context |
||||
ctxCancel func() |
||||
wg sync.WaitGroup |
||||
httpServer *hlsHTTPServer |
||||
muxers map[string]*hlsMuxer |
||||
|
||||
// in
|
||||
chPathReady chan *path |
||||
chPathNotReady chan *path |
||||
chHandleRequest chan hlsMuxerHandleRequestReq |
||||
chCloseMuxer chan *hlsMuxer |
||||
chAPIMuxerList chan hlsManagerAPIMuxersListReq |
||||
chAPIMuxerGet chan hlsManagerAPIMuxersGetReq |
||||
} |
||||
|
||||
func newHLSManager( |
||||
address string, |
||||
encryption bool, |
||||
serverKey string, |
||||
serverCert string, |
||||
externalAuthenticationURL string, |
||||
alwaysRemux bool, |
||||
variant conf.HLSVariant, |
||||
segmentCount int, |
||||
segmentDuration conf.StringDuration, |
||||
partDuration conf.StringDuration, |
||||
segmentMaxSize conf.StringSize, |
||||
allowOrigin string, |
||||
trustedProxies conf.IPsOrCIDRs, |
||||
directory string, |
||||
readTimeout conf.StringDuration, |
||||
writeQueueSize int, |
||||
pathManager *pathManager, |
||||
parent hlsManagerParent, |
||||
) (*hlsManager, error) { |
||||
ctx, ctxCancel := context.WithCancel(context.Background()) |
||||
|
||||
m := &hlsManager{ |
||||
externalAuthenticationURL: externalAuthenticationURL, |
||||
alwaysRemux: alwaysRemux, |
||||
variant: variant, |
||||
segmentCount: segmentCount, |
||||
segmentDuration: segmentDuration, |
||||
partDuration: partDuration, |
||||
segmentMaxSize: segmentMaxSize, |
||||
directory: directory, |
||||
writeQueueSize: writeQueueSize, |
||||
pathManager: pathManager, |
||||
parent: parent, |
||||
ctx: ctx, |
||||
ctxCancel: ctxCancel, |
||||
muxers: make(map[string]*hlsMuxer), |
||||
chPathReady: make(chan *path), |
||||
chPathNotReady: make(chan *path), |
||||
chHandleRequest: make(chan hlsMuxerHandleRequestReq), |
||||
chCloseMuxer: make(chan *hlsMuxer), |
||||
chAPIMuxerList: make(chan hlsManagerAPIMuxersListReq), |
||||
chAPIMuxerGet: make(chan hlsManagerAPIMuxersGetReq), |
||||
} |
||||
|
||||
var err error |
||||
m.httpServer, err = newHLSHTTPServer( |
||||
address, |
||||
encryption, |
||||
serverKey, |
||||
serverCert, |
||||
allowOrigin, |
||||
trustedProxies, |
||||
readTimeout, |
||||
m.pathManager, |
||||
m, |
||||
) |
||||
if err != nil { |
||||
ctxCancel() |
||||
return nil, err |
||||
} |
||||
|
||||
m.Log(logger.Info, "listener opened on "+address) |
||||
|
||||
m.wg.Add(1) |
||||
go m.run() |
||||
|
||||
return m, nil |
||||
} |
||||
|
||||
// Log is the main logging function.
|
||||
func (m *hlsManager) Log(level logger.Level, format string, args ...interface{}) { |
||||
m.parent.Log(level, "[HLS] "+format, args...) |
||||
} |
||||
|
||||
func (m *hlsManager) close() { |
||||
m.Log(logger.Info, "listener is closing") |
||||
m.ctxCancel() |
||||
m.wg.Wait() |
||||
} |
||||
|
||||
func (m *hlsManager) run() { |
||||
defer m.wg.Done() |
||||
|
||||
outer: |
||||
for { |
||||
select { |
||||
case pa := <-m.chPathReady: |
||||
if m.alwaysRemux && !pa.conf.SourceOnDemand { |
||||
if _, ok := m.muxers[pa.name]; !ok { |
||||
m.createMuxer(pa.name, "") |
||||
} |
||||
} |
||||
|
||||
case pa := <-m.chPathNotReady: |
||||
c, ok := m.muxers[pa.name] |
||||
if ok && c.remoteAddr == "" { // created with "always remux"
|
||||
c.close() |
||||
delete(m.muxers, pa.name) |
||||
} |
||||
|
||||
case req := <-m.chHandleRequest: |
||||
r, ok := m.muxers[req.path] |
||||
switch { |
||||
case ok: |
||||
r.processRequest(&req) |
||||
|
||||
default: |
||||
r := m.createMuxer(req.path, req.ctx.ClientIP()) |
||||
r.processRequest(&req) |
||||
} |
||||
|
||||
case c := <-m.chCloseMuxer: |
||||
if c2, ok := m.muxers[c.PathName()]; !ok || c2 != c { |
||||
continue |
||||
} |
||||
delete(m.muxers, c.PathName()) |
||||
|
||||
case req := <-m.chAPIMuxerList: |
||||
data := &defs.APIHLSMuxerList{ |
||||
Items: []*defs.APIHLSMuxer{}, |
||||
} |
||||
|
||||
for _, muxer := range m.muxers { |
||||
data.Items = append(data.Items, muxer.apiItem()) |
||||
} |
||||
|
||||
sort.Slice(data.Items, func(i, j int) bool { |
||||
return data.Items[i].Created.Before(data.Items[j].Created) |
||||
}) |
||||
|
||||
req.res <- hlsManagerAPIMuxersListRes{ |
||||
data: data, |
||||
} |
||||
|
||||
case req := <-m.chAPIMuxerGet: |
||||
muxer, ok := m.muxers[req.name] |
||||
if !ok { |
||||
req.res <- hlsManagerAPIMuxersGetRes{err: fmt.Errorf("muxer not found")} |
||||
continue |
||||
} |
||||
|
||||
req.res <- hlsManagerAPIMuxersGetRes{data: muxer.apiItem()} |
||||
|
||||
case <-m.ctx.Done(): |
||||
break outer |
||||
} |
||||
} |
||||
|
||||
m.ctxCancel() |
||||
|
||||
m.httpServer.close() |
||||
} |
||||
|
||||
func (m *hlsManager) createMuxer(pathName string, remoteAddr string) *hlsMuxer { |
||||
r := newHLSMuxer( |
||||
m.ctx, |
||||
remoteAddr, |
||||
m.externalAuthenticationURL, |
||||
m.variant, |
||||
m.segmentCount, |
||||
m.segmentDuration, |
||||
m.partDuration, |
||||
m.segmentMaxSize, |
||||
m.directory, |
||||
m.writeQueueSize, |
||||
&m.wg, |
||||
pathName, |
||||
m.pathManager, |
||||
m) |
||||
m.muxers[pathName] = r |
||||
return r |
||||
} |
||||
|
||||
// closeMuxer is called by hlsMuxer.
|
||||
func (m *hlsManager) closeMuxer(c *hlsMuxer) { |
||||
select { |
||||
case m.chCloseMuxer <- c: |
||||
case <-m.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// pathReady is called by pathManager.
|
||||
func (m *hlsManager) pathReady(pa *path) { |
||||
select { |
||||
case m.chPathReady <- pa: |
||||
case <-m.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// pathNotReady is called by pathManager.
|
||||
func (m *hlsManager) pathNotReady(pa *path) { |
||||
select { |
||||
case m.chPathNotReady <- pa: |
||||
case <-m.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// apiMuxersList is called by api.
|
||||
func (m *hlsManager) apiMuxersList() (*defs.APIHLSMuxerList, error) { |
||||
req := hlsManagerAPIMuxersListReq{ |
||||
res: make(chan hlsManagerAPIMuxersListRes), |
||||
} |
||||
|
||||
select { |
||||
case m.chAPIMuxerList <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-m.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
// apiMuxersGet is called by api.
|
||||
func (m *hlsManager) apiMuxersGet(name string) (*defs.APIHLSMuxer, error) { |
||||
req := hlsManagerAPIMuxersGetReq{ |
||||
name: name, |
||||
res: make(chan hlsManagerAPIMuxersGetRes), |
||||
} |
||||
|
||||
select { |
||||
case m.chAPIMuxerGet <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-m.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
func (m *hlsManager) handleRequest(req hlsMuxerHandleRequestReq) { |
||||
req.res = make(chan *hlsMuxer) |
||||
|
||||
select { |
||||
case m.chHandleRequest <- req: |
||||
muxer := <-req.res |
||||
if muxer != nil { |
||||
req.ctx.Request.URL.Path = req.file |
||||
muxer.handleRequest(req.ctx) |
||||
} |
||||
|
||||
case <-m.ctx.Done(): |
||||
} |
||||
} |
||||
@ -1,7 +0,0 @@
@@ -1,7 +0,0 @@
|
||||
package core |
||||
|
||||
// publisher is an entity that can publish a stream.
|
||||
type publisher interface { |
||||
source |
||||
close() |
||||
} |
||||
@ -1,17 +0,0 @@
@@ -1,17 +0,0 @@
|
||||
package core |
||||
|
||||
import ( |
||||
"github.com/bluenviron/mediamtx/internal/asyncwriter" |
||||
"github.com/bluenviron/mediamtx/internal/defs" |
||||
"github.com/bluenviron/mediamtx/internal/stream" |
||||
) |
||||
|
||||
// reader is an entity that can read a stream.
|
||||
type reader interface { |
||||
close() |
||||
apiReaderDescribe() defs.APIPathSourceOrReader |
||||
} |
||||
|
||||
func readerMediaInfo(r *asyncwriter.Writer, stream *stream.Stream) string { |
||||
return mediaInfo(stream.MediasForReader(r)) |
||||
} |
||||
@ -1,48 +0,0 @@
@@ -1,48 +0,0 @@
|
||||
package core |
||||
|
||||
import ( |
||||
"net" |
||||
"sync" |
||||
) |
||||
|
||||
type rtmpListener struct { |
||||
ln net.Listener |
||||
wg *sync.WaitGroup |
||||
parent *rtmpServer |
||||
} |
||||
|
||||
func newRTMPListener( |
||||
ln net.Listener, |
||||
wg *sync.WaitGroup, |
||||
parent *rtmpServer, |
||||
) *rtmpListener { |
||||
l := &rtmpListener{ |
||||
ln: ln, |
||||
wg: wg, |
||||
parent: parent, |
||||
} |
||||
|
||||
l.wg.Add(1) |
||||
go l.run() |
||||
|
||||
return l |
||||
} |
||||
|
||||
func (l *rtmpListener) run() { |
||||
defer l.wg.Done() |
||||
|
||||
err := l.runInner() |
||||
|
||||
l.parent.acceptError(err) |
||||
} |
||||
|
||||
func (l *rtmpListener) runInner() error { |
||||
for { |
||||
conn, err := l.ln.Accept() |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
l.parent.newConn(conn) |
||||
} |
||||
} |
||||
@ -1,327 +0,0 @@
@@ -1,327 +0,0 @@
|
||||
package core |
||||
|
||||
import ( |
||||
"context" |
||||
"crypto/tls" |
||||
"fmt" |
||||
"net" |
||||
"sort" |
||||
"sync" |
||||
|
||||
"github.com/google/uuid" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/defs" |
||||
"github.com/bluenviron/mediamtx/internal/externalcmd" |
||||
"github.com/bluenviron/mediamtx/internal/logger" |
||||
"github.com/bluenviron/mediamtx/internal/restrictnetwork" |
||||
) |
||||
|
||||
type rtmpServerAPIConnsListRes struct { |
||||
data *defs.APIRTMPConnList |
||||
err error |
||||
} |
||||
|
||||
type rtmpServerAPIConnsListReq struct { |
||||
res chan rtmpServerAPIConnsListRes |
||||
} |
||||
|
||||
type rtmpServerAPIConnsGetRes struct { |
||||
data *defs.APIRTMPConn |
||||
err error |
||||
} |
||||
|
||||
type rtmpServerAPIConnsGetReq struct { |
||||
uuid uuid.UUID |
||||
res chan rtmpServerAPIConnsGetRes |
||||
} |
||||
|
||||
type rtmpServerAPIConnsKickRes struct { |
||||
err error |
||||
} |
||||
|
||||
type rtmpServerAPIConnsKickReq struct { |
||||
uuid uuid.UUID |
||||
res chan rtmpServerAPIConnsKickRes |
||||
} |
||||
|
||||
type rtmpServerParent interface { |
||||
logger.Writer |
||||
} |
||||
|
||||
type rtmpServer struct { |
||||
readTimeout conf.StringDuration |
||||
writeTimeout conf.StringDuration |
||||
writeQueueSize int |
||||
isTLS bool |
||||
rtspAddress string |
||||
runOnConnect string |
||||
runOnConnectRestart bool |
||||
runOnDisconnect string |
||||
externalCmdPool *externalcmd.Pool |
||||
pathManager *pathManager |
||||
parent rtmpServerParent |
||||
|
||||
ctx context.Context |
||||
ctxCancel func() |
||||
wg sync.WaitGroup |
||||
ln net.Listener |
||||
conns map[*rtmpConn]struct{} |
||||
|
||||
// in
|
||||
chNewConn chan net.Conn |
||||
chAcceptErr chan error |
||||
chCloseConn chan *rtmpConn |
||||
chAPIConnsList chan rtmpServerAPIConnsListReq |
||||
chAPIConnsGet chan rtmpServerAPIConnsGetReq |
||||
chAPIConnsKick chan rtmpServerAPIConnsKickReq |
||||
} |
||||
|
||||
func newRTMPServer( |
||||
address string, |
||||
readTimeout conf.StringDuration, |
||||
writeTimeout conf.StringDuration, |
||||
writeQueueSize int, |
||||
isTLS bool, |
||||
serverCert string, |
||||
serverKey string, |
||||
rtspAddress string, |
||||
runOnConnect string, |
||||
runOnConnectRestart bool, |
||||
runOnDisconnect string, |
||||
externalCmdPool *externalcmd.Pool, |
||||
pathManager *pathManager, |
||||
parent rtmpServerParent, |
||||
) (*rtmpServer, error) { |
||||
ln, err := func() (net.Listener, error) { |
||||
if !isTLS { |
||||
return net.Listen(restrictnetwork.Restrict("tcp", address)) |
||||
} |
||||
|
||||
cert, err := tls.LoadX509KeyPair(serverCert, serverKey) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
network, address := restrictnetwork.Restrict("tcp", address) |
||||
return tls.Listen(network, address, &tls.Config{Certificates: []tls.Certificate{cert}}) |
||||
}() |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
ctx, ctxCancel := context.WithCancel(context.Background()) |
||||
|
||||
s := &rtmpServer{ |
||||
readTimeout: readTimeout, |
||||
writeTimeout: writeTimeout, |
||||
writeQueueSize: writeQueueSize, |
||||
rtspAddress: rtspAddress, |
||||
runOnConnect: runOnConnect, |
||||
runOnConnectRestart: runOnConnectRestart, |
||||
runOnDisconnect: runOnDisconnect, |
||||
isTLS: isTLS, |
||||
externalCmdPool: externalCmdPool, |
||||
pathManager: pathManager, |
||||
parent: parent, |
||||
ctx: ctx, |
||||
ctxCancel: ctxCancel, |
||||
ln: ln, |
||||
conns: make(map[*rtmpConn]struct{}), |
||||
chNewConn: make(chan net.Conn), |
||||
chAcceptErr: make(chan error), |
||||
chCloseConn: make(chan *rtmpConn), |
||||
chAPIConnsList: make(chan rtmpServerAPIConnsListReq), |
||||
chAPIConnsGet: make(chan rtmpServerAPIConnsGetReq), |
||||
chAPIConnsKick: make(chan rtmpServerAPIConnsKickReq), |
||||
} |
||||
|
||||
s.Log(logger.Info, "listener opened on %s", address) |
||||
|
||||
newRTMPListener( |
||||
s.ln, |
||||
&s.wg, |
||||
s, |
||||
) |
||||
|
||||
s.wg.Add(1) |
||||
go s.run() |
||||
|
||||
return s, nil |
||||
} |
||||
|
||||
func (s *rtmpServer) Log(level logger.Level, format string, args ...interface{}) { |
||||
label := func() string { |
||||
if s.isTLS { |
||||
return "RTMPS" |
||||
} |
||||
return "RTMP" |
||||
}() |
||||
s.parent.Log(level, "[%s] "+format, append([]interface{}{label}, args...)...) |
||||
} |
||||
|
||||
func (s *rtmpServer) close() { |
||||
s.Log(logger.Info, "listener is closing") |
||||
s.ctxCancel() |
||||
s.wg.Wait() |
||||
} |
||||
|
||||
func (s *rtmpServer) run() { |
||||
defer s.wg.Done() |
||||
|
||||
outer: |
||||
for { |
||||
select { |
||||
case err := <-s.chAcceptErr: |
||||
s.Log(logger.Error, "%s", err) |
||||
break outer |
||||
|
||||
case nconn := <-s.chNewConn: |
||||
c := newRTMPConn( |
||||
s.ctx, |
||||
s.isTLS, |
||||
s.rtspAddress, |
||||
s.readTimeout, |
||||
s.writeTimeout, |
||||
s.writeQueueSize, |
||||
s.runOnConnect, |
||||
s.runOnConnectRestart, |
||||
s.runOnDisconnect, |
||||
&s.wg, |
||||
nconn, |
||||
s.externalCmdPool, |
||||
s.pathManager, |
||||
s) |
||||
s.conns[c] = struct{}{} |
||||
|
||||
case c := <-s.chCloseConn: |
||||
delete(s.conns, c) |
||||
|
||||
case req := <-s.chAPIConnsList: |
||||
data := &defs.APIRTMPConnList{ |
||||
Items: []*defs.APIRTMPConn{}, |
||||
} |
||||
|
||||
for c := range s.conns { |
||||
data.Items = append(data.Items, c.apiItem()) |
||||
} |
||||
|
||||
sort.Slice(data.Items, func(i, j int) bool { |
||||
return data.Items[i].Created.Before(data.Items[j].Created) |
||||
}) |
||||
|
||||
req.res <- rtmpServerAPIConnsListRes{data: data} |
||||
|
||||
case req := <-s.chAPIConnsGet: |
||||
c := s.findConnByUUID(req.uuid) |
||||
if c == nil { |
||||
req.res <- rtmpServerAPIConnsGetRes{err: fmt.Errorf("connection not found")} |
||||
continue |
||||
} |
||||
|
||||
req.res <- rtmpServerAPIConnsGetRes{data: c.apiItem()} |
||||
|
||||
case req := <-s.chAPIConnsKick: |
||||
c := s.findConnByUUID(req.uuid) |
||||
if c == nil { |
||||
req.res <- rtmpServerAPIConnsKickRes{err: fmt.Errorf("connection not found")} |
||||
continue |
||||
} |
||||
|
||||
delete(s.conns, c) |
||||
c.close() |
||||
req.res <- rtmpServerAPIConnsKickRes{} |
||||
|
||||
case <-s.ctx.Done(): |
||||
break outer |
||||
} |
||||
} |
||||
|
||||
s.ctxCancel() |
||||
|
||||
s.ln.Close() |
||||
} |
||||
|
||||
func (s *rtmpServer) findConnByUUID(uuid uuid.UUID) *rtmpConn { |
||||
for c := range s.conns { |
||||
if c.uuid == uuid { |
||||
return c |
||||
} |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
// newConn is called by rtmpListener.
|
||||
func (s *rtmpServer) newConn(conn net.Conn) { |
||||
select { |
||||
case s.chNewConn <- conn: |
||||
case <-s.ctx.Done(): |
||||
conn.Close() |
||||
} |
||||
} |
||||
|
||||
// acceptError is called by rtmpListener.
|
||||
func (s *rtmpServer) acceptError(err error) { |
||||
select { |
||||
case s.chAcceptErr <- err: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// closeConn is called by rtmpConn.
|
||||
func (s *rtmpServer) closeConn(c *rtmpConn) { |
||||
select { |
||||
case s.chCloseConn <- c: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// apiConnsList is called by api.
|
||||
func (s *rtmpServer) apiConnsList() (*defs.APIRTMPConnList, error) { |
||||
req := rtmpServerAPIConnsListReq{ |
||||
res: make(chan rtmpServerAPIConnsListRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsList <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
// apiConnsGet is called by api.
|
||||
func (s *rtmpServer) apiConnsGet(uuid uuid.UUID) (*defs.APIRTMPConn, error) { |
||||
req := rtmpServerAPIConnsGetReq{ |
||||
uuid: uuid, |
||||
res: make(chan rtmpServerAPIConnsGetRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsGet <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
// apiConnsKick is called by api.
|
||||
func (s *rtmpServer) apiConnsKick(uuid uuid.UUID) error { |
||||
req := rtmpServerAPIConnsKickReq{ |
||||
uuid: uuid, |
||||
res: make(chan rtmpServerAPIConnsKickRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsKick <- req: |
||||
res := <-req.res |
||||
return res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
@ -1,454 +0,0 @@
@@ -1,454 +0,0 @@
|
||||
package core |
||||
|
||||
import ( |
||||
"context" |
||||
"crypto/tls" |
||||
"fmt" |
||||
"sort" |
||||
"strings" |
||||
"sync" |
||||
"time" |
||||
|
||||
"github.com/bluenviron/gortsplib/v4" |
||||
"github.com/bluenviron/gortsplib/v4/pkg/base" |
||||
"github.com/bluenviron/gortsplib/v4/pkg/headers" |
||||
"github.com/bluenviron/gortsplib/v4/pkg/liberrors" |
||||
"github.com/google/uuid" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/defs" |
||||
"github.com/bluenviron/mediamtx/internal/externalcmd" |
||||
"github.com/bluenviron/mediamtx/internal/logger" |
||||
) |
||||
|
||||
type rtspServerParent interface { |
||||
logger.Writer |
||||
} |
||||
|
||||
func printAddresses(srv *gortsplib.Server) string { |
||||
var ret []string |
||||
|
||||
ret = append(ret, fmt.Sprintf("%s (TCP)", srv.RTSPAddress)) |
||||
|
||||
if srv.UDPRTPAddress != "" { |
||||
ret = append(ret, fmt.Sprintf("%s (UDP/RTP)", srv.UDPRTPAddress)) |
||||
} |
||||
|
||||
if srv.UDPRTCPAddress != "" { |
||||
ret = append(ret, fmt.Sprintf("%s (UDP/RTCP)", srv.UDPRTCPAddress)) |
||||
} |
||||
|
||||
return strings.Join(ret, ", ") |
||||
} |
||||
|
||||
type rtspServer struct { |
||||
authMethods []headers.AuthMethod |
||||
readTimeout conf.StringDuration |
||||
isTLS bool |
||||
rtspAddress string |
||||
protocols map[conf.Protocol]struct{} |
||||
runOnConnect string |
||||
runOnConnectRestart bool |
||||
runOnDisconnect string |
||||
externalCmdPool *externalcmd.Pool |
||||
pathManager *pathManager |
||||
parent rtspServerParent |
||||
|
||||
ctx context.Context |
||||
ctxCancel func() |
||||
wg sync.WaitGroup |
||||
srv *gortsplib.Server |
||||
mutex sync.RWMutex |
||||
conns map[*gortsplib.ServerConn]*rtspConn |
||||
sessions map[*gortsplib.ServerSession]*rtspSession |
||||
} |
||||
|
||||
func newRTSPServer( |
||||
address string, |
||||
authMethods []headers.AuthMethod, |
||||
readTimeout conf.StringDuration, |
||||
writeTimeout conf.StringDuration, |
||||
writeQueueSize int, |
||||
useUDP bool, |
||||
useMulticast bool, |
||||
rtpAddress string, |
||||
rtcpAddress string, |
||||
multicastIPRange string, |
||||
multicastRTPPort int, |
||||
multicastRTCPPort int, |
||||
isTLS bool, |
||||
serverCert string, |
||||
serverKey string, |
||||
rtspAddress string, |
||||
protocols map[conf.Protocol]struct{}, |
||||
runOnConnect string, |
||||
runOnConnectRestart bool, |
||||
runOnDisconnect string, |
||||
externalCmdPool *externalcmd.Pool, |
||||
pathManager *pathManager, |
||||
parent rtspServerParent, |
||||
) (*rtspServer, error) { |
||||
ctx, ctxCancel := context.WithCancel(context.Background()) |
||||
|
||||
s := &rtspServer{ |
||||
authMethods: authMethods, |
||||
readTimeout: readTimeout, |
||||
isTLS: isTLS, |
||||
rtspAddress: rtspAddress, |
||||
protocols: protocols, |
||||
runOnConnect: runOnConnect, |
||||
runOnConnectRestart: runOnConnectRestart, |
||||
runOnDisconnect: runOnDisconnect, |
||||
externalCmdPool: externalCmdPool, |
||||
pathManager: pathManager, |
||||
parent: parent, |
||||
ctx: ctx, |
||||
ctxCancel: ctxCancel, |
||||
conns: make(map[*gortsplib.ServerConn]*rtspConn), |
||||
sessions: make(map[*gortsplib.ServerSession]*rtspSession), |
||||
} |
||||
|
||||
s.srv = &gortsplib.Server{ |
||||
Handler: s, |
||||
ReadTimeout: time.Duration(readTimeout), |
||||
WriteTimeout: time.Duration(writeTimeout), |
||||
WriteQueueSize: writeQueueSize, |
||||
RTSPAddress: address, |
||||
} |
||||
|
||||
if useUDP { |
||||
s.srv.UDPRTPAddress = rtpAddress |
||||
s.srv.UDPRTCPAddress = rtcpAddress |
||||
} |
||||
|
||||
if useMulticast { |
||||
s.srv.MulticastIPRange = multicastIPRange |
||||
s.srv.MulticastRTPPort = multicastRTPPort |
||||
s.srv.MulticastRTCPPort = multicastRTCPPort |
||||
} |
||||
|
||||
if isTLS { |
||||
cert, err := tls.LoadX509KeyPair(serverCert, serverKey) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
s.srv.TLSConfig = &tls.Config{Certificates: []tls.Certificate{cert}} |
||||
} |
||||
|
||||
err := s.srv.Start() |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
s.Log(logger.Info, "listener opened on %s", printAddresses(s.srv)) |
||||
|
||||
s.wg.Add(1) |
||||
go s.run() |
||||
|
||||
return s, nil |
||||
} |
||||
|
||||
func (s *rtspServer) Log(level logger.Level, format string, args ...interface{}) { |
||||
label := func() string { |
||||
if s.isTLS { |
||||
return "RTSPS" |
||||
} |
||||
return "RTSP" |
||||
}() |
||||
s.parent.Log(level, "[%s] "+format, append([]interface{}{label}, args...)...) |
||||
} |
||||
|
||||
func (s *rtspServer) getISTLS() bool { |
||||
return s.isTLS |
||||
} |
||||
|
||||
func (s *rtspServer) getServer() *gortsplib.Server { |
||||
return s.srv |
||||
} |
||||
|
||||
func (s *rtspServer) close() { |
||||
s.Log(logger.Info, "listener is closing") |
||||
s.ctxCancel() |
||||
s.wg.Wait() |
||||
} |
||||
|
||||
func (s *rtspServer) run() { |
||||
defer s.wg.Done() |
||||
|
||||
serverErr := make(chan error) |
||||
go func() { |
||||
serverErr <- s.srv.Wait() |
||||
}() |
||||
|
||||
outer: |
||||
select { |
||||
case err := <-serverErr: |
||||
s.Log(logger.Error, "%s", err) |
||||
break outer |
||||
|
||||
case <-s.ctx.Done(): |
||||
s.srv.Close() |
||||
<-serverErr |
||||
break outer |
||||
} |
||||
|
||||
s.ctxCancel() |
||||
} |
||||
|
||||
// OnConnOpen implements gortsplib.ServerHandlerOnConnOpen.
|
||||
func (s *rtspServer) OnConnOpen(ctx *gortsplib.ServerHandlerOnConnOpenCtx) { |
||||
c := newRTSPConn( |
||||
s.isTLS, |
||||
s.rtspAddress, |
||||
s.authMethods, |
||||
s.readTimeout, |
||||
s.runOnConnect, |
||||
s.runOnConnectRestart, |
||||
s.runOnDisconnect, |
||||
s.externalCmdPool, |
||||
s.pathManager, |
||||
ctx.Conn, |
||||
s) |
||||
s.mutex.Lock() |
||||
s.conns[ctx.Conn] = c |
||||
s.mutex.Unlock() |
||||
|
||||
ctx.Conn.SetUserData(c) |
||||
} |
||||
|
||||
// OnConnClose implements gortsplib.ServerHandlerOnConnClose.
|
||||
func (s *rtspServer) OnConnClose(ctx *gortsplib.ServerHandlerOnConnCloseCtx) { |
||||
s.mutex.Lock() |
||||
c := s.conns[ctx.Conn] |
||||
delete(s.conns, ctx.Conn) |
||||
s.mutex.Unlock() |
||||
c.onClose(ctx.Error) |
||||
} |
||||
|
||||
// OnRequest implements gortsplib.ServerHandlerOnRequest.
|
||||
func (s *rtspServer) OnRequest(sc *gortsplib.ServerConn, req *base.Request) { |
||||
c := sc.UserData().(*rtspConn) |
||||
c.onRequest(req) |
||||
} |
||||
|
||||
// OnResponse implements gortsplib.ServerHandlerOnResponse.
|
||||
func (s *rtspServer) OnResponse(sc *gortsplib.ServerConn, res *base.Response) { |
||||
c := sc.UserData().(*rtspConn) |
||||
c.OnResponse(res) |
||||
} |
||||
|
||||
// OnSessionOpen implements gortsplib.ServerHandlerOnSessionOpen.
|
||||
func (s *rtspServer) OnSessionOpen(ctx *gortsplib.ServerHandlerOnSessionOpenCtx) { |
||||
se := newRTSPSession( |
||||
s.isTLS, |
||||
s.protocols, |
||||
ctx.Session, |
||||
ctx.Conn, |
||||
s.externalCmdPool, |
||||
s.pathManager, |
||||
s) |
||||
s.mutex.Lock() |
||||
s.sessions[ctx.Session] = se |
||||
s.mutex.Unlock() |
||||
ctx.Session.SetUserData(se) |
||||
} |
||||
|
||||
// OnSessionClose implements gortsplib.ServerHandlerOnSessionClose.
|
||||
func (s *rtspServer) OnSessionClose(ctx *gortsplib.ServerHandlerOnSessionCloseCtx) { |
||||
s.mutex.Lock() |
||||
se := s.sessions[ctx.Session] |
||||
delete(s.sessions, ctx.Session) |
||||
s.mutex.Unlock() |
||||
|
||||
if se != nil { |
||||
se.onClose(ctx.Error) |
||||
} |
||||
} |
||||
|
||||
// OnDescribe implements gortsplib.ServerHandlerOnDescribe.
|
||||
func (s *rtspServer) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx, |
||||
) (*base.Response, *gortsplib.ServerStream, error) { |
||||
c := ctx.Conn.UserData().(*rtspConn) |
||||
return c.onDescribe(ctx) |
||||
} |
||||
|
||||
// OnAnnounce implements gortsplib.ServerHandlerOnAnnounce.
|
||||
func (s *rtspServer) OnAnnounce(ctx *gortsplib.ServerHandlerOnAnnounceCtx) (*base.Response, error) { |
||||
c := ctx.Conn.UserData().(*rtspConn) |
||||
se := ctx.Session.UserData().(*rtspSession) |
||||
return se.onAnnounce(c, ctx) |
||||
} |
||||
|
||||
// OnSetup implements gortsplib.ServerHandlerOnSetup.
|
||||
func (s *rtspServer) OnSetup(ctx *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error) { |
||||
c := ctx.Conn.UserData().(*rtspConn) |
||||
se := ctx.Session.UserData().(*rtspSession) |
||||
return se.onSetup(c, ctx) |
||||
} |
||||
|
||||
// OnPlay implements gortsplib.ServerHandlerOnPlay.
|
||||
func (s *rtspServer) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) { |
||||
se := ctx.Session.UserData().(*rtspSession) |
||||
return se.onPlay(ctx) |
||||
} |
||||
|
||||
// OnRecord implements gortsplib.ServerHandlerOnRecord.
|
||||
func (s *rtspServer) OnRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.Response, error) { |
||||
se := ctx.Session.UserData().(*rtspSession) |
||||
return se.onRecord(ctx) |
||||
} |
||||
|
||||
// OnPause implements gortsplib.ServerHandlerOnPause.
|
||||
func (s *rtspServer) OnPause(ctx *gortsplib.ServerHandlerOnPauseCtx) (*base.Response, error) { |
||||
se := ctx.Session.UserData().(*rtspSession) |
||||
return se.onPause(ctx) |
||||
} |
||||
|
||||
// OnPacketLost implements gortsplib.ServerHandlerOnDecodeError.
|
||||
func (s *rtspServer) OnPacketLost(ctx *gortsplib.ServerHandlerOnPacketLostCtx) { |
||||
se := ctx.Session.UserData().(*rtspSession) |
||||
se.onPacketLost(ctx) |
||||
} |
||||
|
||||
// OnDecodeError implements gortsplib.ServerHandlerOnDecodeError.
|
||||
func (s *rtspServer) OnDecodeError(ctx *gortsplib.ServerHandlerOnDecodeErrorCtx) { |
||||
se := ctx.Session.UserData().(*rtspSession) |
||||
se.onDecodeError(ctx) |
||||
} |
||||
|
||||
// OnDecodeError implements gortsplib.ServerHandlerOnStreamWriteError.
|
||||
func (s *rtspServer) OnStreamWriteError(ctx *gortsplib.ServerHandlerOnStreamWriteErrorCtx) { |
||||
se := ctx.Session.UserData().(*rtspSession) |
||||
se.onStreamWriteError(ctx) |
||||
} |
||||
|
||||
func (s *rtspServer) findConnByUUID(uuid uuid.UUID) *rtspConn { |
||||
for _, c := range s.conns { |
||||
if c.uuid == uuid { |
||||
return c |
||||
} |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
func (s *rtspServer) findSessionByUUID(uuid uuid.UUID) (*gortsplib.ServerSession, *rtspSession) { |
||||
for key, sx := range s.sessions { |
||||
if sx.uuid == uuid { |
||||
return key, sx |
||||
} |
||||
} |
||||
return nil, nil |
||||
} |
||||
|
||||
// apiConnsList is called by api and metrics.
|
||||
func (s *rtspServer) apiConnsList() (*defs.APIRTSPConnsList, error) { |
||||
select { |
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
default: |
||||
} |
||||
|
||||
s.mutex.RLock() |
||||
defer s.mutex.RUnlock() |
||||
|
||||
data := &defs.APIRTSPConnsList{ |
||||
Items: []*defs.APIRTSPConn{}, |
||||
} |
||||
|
||||
for _, c := range s.conns { |
||||
data.Items = append(data.Items, c.apiItem()) |
||||
} |
||||
|
||||
sort.Slice(data.Items, func(i, j int) bool { |
||||
return data.Items[i].Created.Before(data.Items[j].Created) |
||||
}) |
||||
|
||||
return data, nil |
||||
} |
||||
|
||||
// apiConnsGet is called by api.
|
||||
func (s *rtspServer) apiConnsGet(uuid uuid.UUID) (*defs.APIRTSPConn, error) { |
||||
select { |
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
default: |
||||
} |
||||
|
||||
s.mutex.RLock() |
||||
defer s.mutex.RUnlock() |
||||
|
||||
conn := s.findConnByUUID(uuid) |
||||
if conn == nil { |
||||
return nil, fmt.Errorf("connection not found") |
||||
} |
||||
|
||||
return conn.apiItem(), nil |
||||
} |
||||
|
||||
// apiSessionsList is called by api and metrics.
|
||||
func (s *rtspServer) apiSessionsList() (*defs.APIRTSPSessionList, error) { |
||||
select { |
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
default: |
||||
} |
||||
|
||||
s.mutex.RLock() |
||||
defer s.mutex.RUnlock() |
||||
|
||||
data := &defs.APIRTSPSessionList{ |
||||
Items: []*defs.APIRTSPSession{}, |
||||
} |
||||
|
||||
for _, s := range s.sessions { |
||||
data.Items = append(data.Items, s.apiItem()) |
||||
} |
||||
|
||||
sort.Slice(data.Items, func(i, j int) bool { |
||||
return data.Items[i].Created.Before(data.Items[j].Created) |
||||
}) |
||||
|
||||
return data, nil |
||||
} |
||||
|
||||
// apiSessionsGet is called by api.
|
||||
func (s *rtspServer) apiSessionsGet(uuid uuid.UUID) (*defs.APIRTSPSession, error) { |
||||
select { |
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
default: |
||||
} |
||||
|
||||
s.mutex.RLock() |
||||
defer s.mutex.RUnlock() |
||||
|
||||
_, sx := s.findSessionByUUID(uuid) |
||||
if sx == nil { |
||||
return nil, fmt.Errorf("session not found") |
||||
} |
||||
|
||||
return sx.apiItem(), nil |
||||
} |
||||
|
||||
// apiSessionsKick is called by api.
|
||||
func (s *rtspServer) apiSessionsKick(uuid uuid.UUID) error { |
||||
select { |
||||
case <-s.ctx.Done(): |
||||
return fmt.Errorf("terminated") |
||||
default: |
||||
} |
||||
|
||||
s.mutex.RLock() |
||||
defer s.mutex.RUnlock() |
||||
|
||||
key, sx := s.findSessionByUUID(uuid) |
||||
if sx == nil { |
||||
return fmt.Errorf("session not found") |
||||
} |
||||
|
||||
sx.close() |
||||
delete(s.sessions, key) |
||||
sx.onClose(liberrors.ErrServerTerminated{}) |
||||
return nil |
||||
} |
||||
@ -1,330 +0,0 @@
@@ -1,330 +0,0 @@
|
||||
package core |
||||
|
||||
import ( |
||||
"context" |
||||
"fmt" |
||||
"sort" |
||||
"sync" |
||||
"time" |
||||
|
||||
srt "github.com/datarhei/gosrt" |
||||
"github.com/google/uuid" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/defs" |
||||
"github.com/bluenviron/mediamtx/internal/externalcmd" |
||||
"github.com/bluenviron/mediamtx/internal/logger" |
||||
) |
||||
|
||||
func srtMaxPayloadSize(u int) int { |
||||
return ((u - 16) / 188) * 188 // 16 = SRT header, 188 = MPEG-TS packet
|
||||
} |
||||
|
||||
type srtNewConnReq struct { |
||||
connReq srt.ConnRequest |
||||
res chan *srtConn |
||||
} |
||||
|
||||
type srtServerAPIConnsListRes struct { |
||||
data *defs.APISRTConnList |
||||
err error |
||||
} |
||||
|
||||
type srtServerAPIConnsListReq struct { |
||||
res chan srtServerAPIConnsListRes |
||||
} |
||||
|
||||
type srtServerAPIConnsGetRes struct { |
||||
data *defs.APISRTConn |
||||
err error |
||||
} |
||||
|
||||
type srtServerAPIConnsGetReq struct { |
||||
uuid uuid.UUID |
||||
res chan srtServerAPIConnsGetRes |
||||
} |
||||
|
||||
type srtServerAPIConnsKickRes struct { |
||||
err error |
||||
} |
||||
|
||||
type srtServerAPIConnsKickReq struct { |
||||
uuid uuid.UUID |
||||
res chan srtServerAPIConnsKickRes |
||||
} |
||||
|
||||
type srtServerParent interface { |
||||
logger.Writer |
||||
} |
||||
|
||||
type srtServer struct { |
||||
rtspAddress string |
||||
readTimeout conf.StringDuration |
||||
writeTimeout conf.StringDuration |
||||
writeQueueSize int |
||||
udpMaxPayloadSize int |
||||
runOnConnect string |
||||
runOnConnectRestart bool |
||||
runOnDisconnect string |
||||
externalCmdPool *externalcmd.Pool |
||||
pathManager *pathManager |
||||
parent srtServerParent |
||||
|
||||
ctx context.Context |
||||
ctxCancel func() |
||||
wg sync.WaitGroup |
||||
ln srt.Listener |
||||
conns map[*srtConn]struct{} |
||||
|
||||
// in
|
||||
chNewConnRequest chan srtNewConnReq |
||||
chAcceptErr chan error |
||||
chCloseConn chan *srtConn |
||||
chAPIConnsList chan srtServerAPIConnsListReq |
||||
chAPIConnsGet chan srtServerAPIConnsGetReq |
||||
chAPIConnsKick chan srtServerAPIConnsKickReq |
||||
} |
||||
|
||||
func newSRTServer( |
||||
address string, |
||||
rtspAddress string, |
||||
readTimeout conf.StringDuration, |
||||
writeTimeout conf.StringDuration, |
||||
writeQueueSize int, |
||||
udpMaxPayloadSize int, |
||||
runOnConnect string, |
||||
runOnConnectRestart bool, |
||||
runOnDisconnect string, |
||||
externalCmdPool *externalcmd.Pool, |
||||
pathManager *pathManager, |
||||
parent srtServerParent, |
||||
) (*srtServer, error) { |
||||
conf := srt.DefaultConfig() |
||||
conf.ConnectionTimeout = time.Duration(readTimeout) |
||||
conf.PayloadSize = uint32(srtMaxPayloadSize(udpMaxPayloadSize)) |
||||
|
||||
ln, err := srt.Listen("srt", address, conf) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
ctx, ctxCancel := context.WithCancel(context.Background()) |
||||
|
||||
s := &srtServer{ |
||||
rtspAddress: rtspAddress, |
||||
readTimeout: readTimeout, |
||||
writeTimeout: writeTimeout, |
||||
writeQueueSize: writeQueueSize, |
||||
udpMaxPayloadSize: udpMaxPayloadSize, |
||||
runOnConnect: runOnConnect, |
||||
runOnConnectRestart: runOnConnectRestart, |
||||
runOnDisconnect: runOnDisconnect, |
||||
externalCmdPool: externalCmdPool, |
||||
pathManager: pathManager, |
||||
parent: parent, |
||||
ctx: ctx, |
||||
ctxCancel: ctxCancel, |
||||
ln: ln, |
||||
conns: make(map[*srtConn]struct{}), |
||||
chNewConnRequest: make(chan srtNewConnReq), |
||||
chAcceptErr: make(chan error), |
||||
chCloseConn: make(chan *srtConn), |
||||
chAPIConnsList: make(chan srtServerAPIConnsListReq), |
||||
chAPIConnsGet: make(chan srtServerAPIConnsGetReq), |
||||
chAPIConnsKick: make(chan srtServerAPIConnsKickReq), |
||||
} |
||||
|
||||
s.Log(logger.Info, "listener opened on "+address+" (UDP)") |
||||
|
||||
newSRTListener( |
||||
s.ln, |
||||
&s.wg, |
||||
s, |
||||
) |
||||
|
||||
s.wg.Add(1) |
||||
go s.run() |
||||
|
||||
return s, nil |
||||
} |
||||
|
||||
// Log is the main logging function.
|
||||
func (s *srtServer) Log(level logger.Level, format string, args ...interface{}) { |
||||
s.parent.Log(level, "[SRT] "+format, args...) |
||||
} |
||||
|
||||
func (s *srtServer) close() { |
||||
s.Log(logger.Info, "listener is closing") |
||||
s.ctxCancel() |
||||
s.wg.Wait() |
||||
} |
||||
|
||||
func (s *srtServer) run() { |
||||
defer s.wg.Done() |
||||
|
||||
outer: |
||||
for { |
||||
select { |
||||
case err := <-s.chAcceptErr: |
||||
s.Log(logger.Error, "%s", err) |
||||
break outer |
||||
|
||||
case req := <-s.chNewConnRequest: |
||||
c := newSRTConn( |
||||
s.ctx, |
||||
s.rtspAddress, |
||||
s.readTimeout, |
||||
s.writeTimeout, |
||||
s.writeQueueSize, |
||||
s.udpMaxPayloadSize, |
||||
req.connReq, |
||||
s.runOnConnect, |
||||
s.runOnConnectRestart, |
||||
s.runOnDisconnect, |
||||
&s.wg, |
||||
s.externalCmdPool, |
||||
s.pathManager, |
||||
s) |
||||
s.conns[c] = struct{}{} |
||||
req.res <- c |
||||
|
||||
case c := <-s.chCloseConn: |
||||
delete(s.conns, c) |
||||
|
||||
case req := <-s.chAPIConnsList: |
||||
data := &defs.APISRTConnList{ |
||||
Items: []*defs.APISRTConn{}, |
||||
} |
||||
|
||||
for c := range s.conns { |
||||
data.Items = append(data.Items, c.apiItem()) |
||||
} |
||||
|
||||
sort.Slice(data.Items, func(i, j int) bool { |
||||
return data.Items[i].Created.Before(data.Items[j].Created) |
||||
}) |
||||
|
||||
req.res <- srtServerAPIConnsListRes{data: data} |
||||
|
||||
case req := <-s.chAPIConnsGet: |
||||
c := s.findConnByUUID(req.uuid) |
||||
if c == nil { |
||||
req.res <- srtServerAPIConnsGetRes{err: fmt.Errorf("connection not found")} |
||||
continue |
||||
} |
||||
|
||||
req.res <- srtServerAPIConnsGetRes{data: c.apiItem()} |
||||
|
||||
case req := <-s.chAPIConnsKick: |
||||
c := s.findConnByUUID(req.uuid) |
||||
if c == nil { |
||||
req.res <- srtServerAPIConnsKickRes{err: fmt.Errorf("connection not found")} |
||||
continue |
||||
} |
||||
|
||||
delete(s.conns, c) |
||||
c.close() |
||||
req.res <- srtServerAPIConnsKickRes{} |
||||
|
||||
case <-s.ctx.Done(): |
||||
break outer |
||||
} |
||||
} |
||||
|
||||
s.ctxCancel() |
||||
|
||||
s.ln.Close() |
||||
} |
||||
|
||||
func (s *srtServer) findConnByUUID(uuid uuid.UUID) *srtConn { |
||||
for sx := range s.conns { |
||||
if sx.uuid == uuid { |
||||
return sx |
||||
} |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
// newConnRequest is called by srtListener.
|
||||
func (s *srtServer) newConnRequest(connReq srt.ConnRequest) *srtConn { |
||||
req := srtNewConnReq{ |
||||
connReq: connReq, |
||||
res: make(chan *srtConn), |
||||
} |
||||
|
||||
select { |
||||
case s.chNewConnRequest <- req: |
||||
c := <-req.res |
||||
|
||||
return c.new(req) |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil |
||||
} |
||||
} |
||||
|
||||
// acceptError is called by srtListener.
|
||||
func (s *srtServer) acceptError(err error) { |
||||
select { |
||||
case s.chAcceptErr <- err: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// closeConn is called by srtConn.
|
||||
func (s *srtServer) closeConn(c *srtConn) { |
||||
select { |
||||
case s.chCloseConn <- c: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// apiConnsList is called by api.
|
||||
func (s *srtServer) apiConnsList() (*defs.APISRTConnList, error) { |
||||
req := srtServerAPIConnsListReq{ |
||||
res: make(chan srtServerAPIConnsListRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsList <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
// apiConnsGet is called by api.
|
||||
func (s *srtServer) apiConnsGet(uuid uuid.UUID) (*defs.APISRTConn, error) { |
||||
req := srtServerAPIConnsGetReq{ |
||||
uuid: uuid, |
||||
res: make(chan srtServerAPIConnsGetRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsGet <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
// apiConnsKick is called by api.
|
||||
func (s *srtServer) apiConnsKick(uuid uuid.UUID) error { |
||||
req := srtServerAPIConnsKickReq{ |
||||
uuid: uuid, |
||||
res: make(chan srtServerAPIConnsKickRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsKick <- req: |
||||
res := <-req.res |
||||
return res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
@ -0,0 +1,23 @@
@@ -0,0 +1,23 @@
|
||||
package defs |
||||
|
||||
// AuthProtocol is a authentication protocol.
|
||||
type AuthProtocol string |
||||
|
||||
// authentication protocols.
|
||||
const ( |
||||
AuthProtocolRTSP AuthProtocol = "rtsp" |
||||
AuthProtocolRTMP AuthProtocol = "rtmp" |
||||
AuthProtocolHLS AuthProtocol = "hls" |
||||
AuthProtocolWebRTC AuthProtocol = "webrtc" |
||||
AuthProtocolSRT AuthProtocol = "srt" |
||||
) |
||||
|
||||
// ErrAuthentication is a authentication error.
|
||||
type ErrAuthentication struct { |
||||
Message string |
||||
} |
||||
|
||||
// Error implements the error interface.
|
||||
func (e *ErrAuthentication) Error() string { |
||||
return "authentication failed: " + e.Message |
||||
} |
||||
@ -1,25 +1,156 @@
@@ -1,25 +1,156 @@
|
||||
package defs |
||||
|
||||
import ( |
||||
"fmt" |
||||
"net" |
||||
|
||||
"github.com/bluenviron/gortsplib/v4/pkg/base" |
||||
"github.com/bluenviron/gortsplib/v4/pkg/description" |
||||
"github.com/google/uuid" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/externalcmd" |
||||
"github.com/bluenviron/mediamtx/internal/stream" |
||||
) |
||||
|
||||
// PathSourceStaticSetReadyRes is a set ready response to a static source.
|
||||
// ErrPathNoOnePublishing is returned when no one is publishing.
|
||||
type ErrPathNoOnePublishing struct { |
||||
PathName string |
||||
} |
||||
|
||||
// Error implements the error interface.
|
||||
func (e ErrPathNoOnePublishing) Error() string { |
||||
return fmt.Sprintf("no one is publishing to path '%s'", e.PathName) |
||||
} |
||||
|
||||
// Path is a path.
|
||||
type Path interface { |
||||
Name() string |
||||
SafeConf() *conf.Path |
||||
ExternalCmdEnv() externalcmd.Environment |
||||
StartPublisher(req PathStartPublisherReq) PathStartPublisherRes |
||||
StopPublisher(req PathStopPublisherReq) |
||||
RemovePublisher(req PathRemovePublisherReq) |
||||
RemoveReader(req PathRemoveReaderReq) |
||||
} |
||||
|
||||
// PathAccessRequest is an access request.
|
||||
type PathAccessRequest struct { |
||||
Name string |
||||
Query string |
||||
Publish bool |
||||
SkipAuth bool |
||||
|
||||
// only if skipAuth = false
|
||||
IP net.IP |
||||
User string |
||||
Pass string |
||||
Proto AuthProtocol |
||||
ID *uuid.UUID |
||||
RTSPRequest *base.Request |
||||
RTSPBaseURL *base.URL |
||||
RTSPNonce string |
||||
} |
||||
|
||||
// PathGetConfForPathRes contains the response of GetConfForPath().
|
||||
type PathGetConfForPathRes struct { |
||||
Conf *conf.Path |
||||
Err error |
||||
} |
||||
|
||||
// PathGetConfForPathReq contains arguments of GetConfForPath().
|
||||
type PathGetConfForPathReq struct { |
||||
AccessRequest PathAccessRequest |
||||
Res chan PathGetConfForPathRes |
||||
} |
||||
|
||||
// PathDescribeRes contains the response of Describe().
|
||||
type PathDescribeRes struct { |
||||
Path Path |
||||
Stream *stream.Stream |
||||
Redirect string |
||||
Err error |
||||
} |
||||
|
||||
// PathDescribeReq contains arguments of Describe().
|
||||
type PathDescribeReq struct { |
||||
AccessRequest PathAccessRequest |
||||
Res chan PathDescribeRes |
||||
} |
||||
|
||||
// PathAddPublisherRes contains the response of AddPublisher().
|
||||
type PathAddPublisherRes struct { |
||||
Path Path |
||||
Err error |
||||
} |
||||
|
||||
// PathAddPublisherReq contains arguments of AddPublisher().
|
||||
type PathAddPublisherReq struct { |
||||
Author Publisher |
||||
AccessRequest PathAccessRequest |
||||
Res chan PathAddPublisherRes |
||||
} |
||||
|
||||
// PathRemovePublisherReq contains arguments of RemovePublisher().
|
||||
type PathRemovePublisherReq struct { |
||||
Author Publisher |
||||
Res chan struct{} |
||||
} |
||||
|
||||
// PathStartPublisherRes contains the response of StartPublisher().
|
||||
type PathStartPublisherRes struct { |
||||
Stream *stream.Stream |
||||
Err error |
||||
} |
||||
|
||||
// PathStartPublisherReq contains arguments of StartPublisher().
|
||||
type PathStartPublisherReq struct { |
||||
Author Publisher |
||||
Desc *description.Session |
||||
GenerateRTPPackets bool |
||||
Res chan PathStartPublisherRes |
||||
} |
||||
|
||||
// PathStopPublisherReq contains arguments of StopPublisher().
|
||||
type PathStopPublisherReq struct { |
||||
Author Publisher |
||||
Res chan struct{} |
||||
} |
||||
|
||||
// PathAddReaderRes contains the response of AddReader().
|
||||
type PathAddReaderRes struct { |
||||
Path Path |
||||
Stream *stream.Stream |
||||
Err error |
||||
} |
||||
|
||||
// PathAddReaderReq contains arguments of AddReader().
|
||||
type PathAddReaderReq struct { |
||||
Author Reader |
||||
AccessRequest PathAccessRequest |
||||
Res chan PathAddReaderRes |
||||
} |
||||
|
||||
// PathRemoveReaderReq contains arguments of RemoveReader().
|
||||
type PathRemoveReaderReq struct { |
||||
Author Reader |
||||
Res chan struct{} |
||||
} |
||||
|
||||
// PathSourceStaticSetReadyRes contains the response of SetReadu().
|
||||
type PathSourceStaticSetReadyRes struct { |
||||
Stream *stream.Stream |
||||
Err error |
||||
} |
||||
|
||||
// PathSourceStaticSetReadyReq is a set ready request from a static source.
|
||||
// PathSourceStaticSetReadyReq contains arguments of SetReady().
|
||||
type PathSourceStaticSetReadyReq struct { |
||||
Desc *description.Session |
||||
GenerateRTPPackets bool |
||||
Res chan PathSourceStaticSetReadyRes |
||||
} |
||||
|
||||
// PathSourceStaticSetNotReadyReq is a set not ready request from a static source.
|
||||
// PathSourceStaticSetNotReadyReq contains arguments of SetNotReady().
|
||||
type PathSourceStaticSetNotReadyReq struct { |
||||
Res chan struct{} |
||||
} |
||||
|
||||
@ -0,0 +1,9 @@
@@ -0,0 +1,9 @@
|
||||
package defs |
||||
|
||||
// PathManager is a path manager.
|
||||
type PathManager interface { |
||||
GetConfForPath(req PathGetConfForPathReq) PathGetConfForPathRes |
||||
Describe(req PathDescribeReq) PathDescribeRes |
||||
AddPublisher(req PathAddPublisherReq) PathAddPublisherRes |
||||
AddReader(req PathAddReaderReq) PathAddReaderRes |
||||
} |
||||
@ -0,0 +1,7 @@
@@ -0,0 +1,7 @@
|
||||
package defs |
||||
|
||||
// Publisher is an entity that can publish a stream.
|
||||
type Publisher interface { |
||||
Source |
||||
Close() |
||||
} |
||||
@ -0,0 +1,7 @@
@@ -0,0 +1,7 @@
|
||||
package defs |
||||
|
||||
// Reader is an entity that can read a stream.
|
||||
type Reader interface { |
||||
Close() |
||||
APIReaderDescribe() APIPathSourceOrReader |
||||
} |
||||
@ -0,0 +1,290 @@
@@ -0,0 +1,290 @@
|
||||
// Package hls contains a HLS server.
|
||||
package hls |
||||
|
||||
import ( |
||||
"context" |
||||
"fmt" |
||||
"sort" |
||||
"sync" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/defs" |
||||
"github.com/bluenviron/mediamtx/internal/logger" |
||||
) |
||||
|
||||
type serverAPIMuxersListRes struct { |
||||
data *defs.APIHLSMuxerList |
||||
err error |
||||
} |
||||
|
||||
type serverAPIMuxersListReq struct { |
||||
res chan serverAPIMuxersListRes |
||||
} |
||||
|
||||
type serverAPIMuxersGetRes struct { |
||||
data *defs.APIHLSMuxer |
||||
err error |
||||
} |
||||
|
||||
type serverAPIMuxersGetReq struct { |
||||
name string |
||||
res chan serverAPIMuxersGetRes |
||||
} |
||||
|
||||
type serverParent interface { |
||||
logger.Writer |
||||
} |
||||
|
||||
// Server is a HLS server.
|
||||
type Server struct { |
||||
Address string |
||||
Encryption bool |
||||
ServerKey string |
||||
ServerCert string |
||||
ExternalAuthenticationURL string |
||||
AlwaysRemux bool |
||||
Variant conf.HLSVariant |
||||
SegmentCount int |
||||
SegmentDuration conf.StringDuration |
||||
PartDuration conf.StringDuration |
||||
SegmentMaxSize conf.StringSize |
||||
AllowOrigin string |
||||
TrustedProxies conf.IPsOrCIDRs |
||||
Directory string |
||||
ReadTimeout conf.StringDuration |
||||
WriteQueueSize int |
||||
PathManager defs.PathManager |
||||
Parent serverParent |
||||
|
||||
ctx context.Context |
||||
ctxCancel func() |
||||
wg sync.WaitGroup |
||||
httpServer *httpServer |
||||
muxers map[string]*muxer |
||||
|
||||
// in
|
||||
chPathReady chan defs.Path |
||||
chPathNotReady chan defs.Path |
||||
chHandleRequest chan muxerHandleRequestReq |
||||
chCloseMuxer chan *muxer |
||||
chAPIMuxerList chan serverAPIMuxersListReq |
||||
chAPIMuxerGet chan serverAPIMuxersGetReq |
||||
} |
||||
|
||||
// Initialize initializes the server.
|
||||
func (s *Server) Initialize() error { |
||||
ctx, ctxCancel := context.WithCancel(context.Background()) |
||||
|
||||
s.ctx = ctx |
||||
s.ctxCancel = ctxCancel |
||||
s.muxers = make(map[string]*muxer) |
||||
s.chPathReady = make(chan defs.Path) |
||||
s.chPathNotReady = make(chan defs.Path) |
||||
s.chHandleRequest = make(chan muxerHandleRequestReq) |
||||
s.chCloseMuxer = make(chan *muxer) |
||||
s.chAPIMuxerList = make(chan serverAPIMuxersListReq) |
||||
s.chAPIMuxerGet = make(chan serverAPIMuxersGetReq) |
||||
|
||||
s.httpServer = &httpServer{ |
||||
address: s.Address, |
||||
encryption: s.Encryption, |
||||
serverKey: s.ServerKey, |
||||
serverCert: s.ServerCert, |
||||
allowOrigin: s.AllowOrigin, |
||||
trustedProxies: s.TrustedProxies, |
||||
readTimeout: s.ReadTimeout, |
||||
pathManager: s.PathManager, |
||||
parent: s, |
||||
} |
||||
err := s.httpServer.initialize() |
||||
if err != nil { |
||||
ctxCancel() |
||||
return err |
||||
} |
||||
|
||||
s.Log(logger.Info, "listener opened on "+s.Address) |
||||
|
||||
s.wg.Add(1) |
||||
go s.run() |
||||
|
||||
return nil |
||||
} |
||||
|
||||
// Log implements logger.Writer.
|
||||
func (s *Server) Log(level logger.Level, format string, args ...interface{}) { |
||||
s.Parent.Log(level, "[HLS] "+format, args...) |
||||
} |
||||
|
||||
// Close closes the server.
|
||||
func (s *Server) Close() { |
||||
s.Log(logger.Info, "listener is closing") |
||||
s.ctxCancel() |
||||
s.wg.Wait() |
||||
} |
||||
|
||||
func (s *Server) run() { |
||||
defer s.wg.Done() |
||||
|
||||
outer: |
||||
for { |
||||
select { |
||||
case pa := <-s.chPathReady: |
||||
if s.AlwaysRemux && !pa.SafeConf().SourceOnDemand { |
||||
if _, ok := s.muxers[pa.Name()]; !ok { |
||||
s.createMuxer(pa.Name(), "") |
||||
} |
||||
} |
||||
|
||||
case pa := <-s.chPathNotReady: |
||||
c, ok := s.muxers[pa.Name()] |
||||
if ok && c.remoteAddr == "" { // created with "always remux"
|
||||
c.Close() |
||||
delete(s.muxers, pa.Name()) |
||||
} |
||||
|
||||
case req := <-s.chHandleRequest: |
||||
r, ok := s.muxers[req.path] |
||||
switch { |
||||
case ok: |
||||
r.processRequest(&req) |
||||
|
||||
default: |
||||
r := s.createMuxer(req.path, req.ctx.ClientIP()) |
||||
r.processRequest(&req) |
||||
} |
||||
|
||||
case c := <-s.chCloseMuxer: |
||||
if c2, ok := s.muxers[c.PathName()]; !ok || c2 != c { |
||||
continue |
||||
} |
||||
delete(s.muxers, c.PathName()) |
||||
|
||||
case req := <-s.chAPIMuxerList: |
||||
data := &defs.APIHLSMuxerList{ |
||||
Items: []*defs.APIHLSMuxer{}, |
||||
} |
||||
|
||||
for _, muxer := range s.muxers { |
||||
data.Items = append(data.Items, muxer.apiItem()) |
||||
} |
||||
|
||||
sort.Slice(data.Items, func(i, j int) bool { |
||||
return data.Items[i].Created.Before(data.Items[j].Created) |
||||
}) |
||||
|
||||
req.res <- serverAPIMuxersListRes{ |
||||
data: data, |
||||
} |
||||
|
||||
case req := <-s.chAPIMuxerGet: |
||||
muxer, ok := s.muxers[req.name] |
||||
if !ok { |
||||
req.res <- serverAPIMuxersGetRes{err: fmt.Errorf("muxer not found")} |
||||
continue |
||||
} |
||||
|
||||
req.res <- serverAPIMuxersGetRes{data: muxer.apiItem()} |
||||
|
||||
case <-s.ctx.Done(): |
||||
break outer |
||||
} |
||||
} |
||||
|
||||
s.ctxCancel() |
||||
|
||||
s.httpServer.close() |
||||
} |
||||
|
||||
func (s *Server) createMuxer(pathName string, remoteAddr string) *muxer { |
||||
r := &muxer{ |
||||
parentCtx: s.ctx, |
||||
remoteAddr: remoteAddr, |
||||
externalAuthenticationURL: s.ExternalAuthenticationURL, |
||||
variant: s.Variant, |
||||
segmentCount: s.SegmentCount, |
||||
segmentDuration: s.SegmentDuration, |
||||
partDuration: s.PartDuration, |
||||
segmentMaxSize: s.SegmentMaxSize, |
||||
directory: s.Directory, |
||||
writeQueueSize: s.WriteQueueSize, |
||||
wg: &s.wg, |
||||
pathName: pathName, |
||||
pathManager: s.PathManager, |
||||
parent: s, |
||||
} |
||||
r.initialize() |
||||
s.muxers[pathName] = r |
||||
return r |
||||
} |
||||
|
||||
// closeMuxer is called by muxer.
|
||||
func (s *Server) closeMuxer(c *muxer) { |
||||
select { |
||||
case s.chCloseMuxer <- c: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
func (s *Server) handleRequest(req muxerHandleRequestReq) { |
||||
req.res = make(chan *muxer) |
||||
|
||||
select { |
||||
case s.chHandleRequest <- req: |
||||
muxer := <-req.res |
||||
if muxer != nil { |
||||
req.ctx.Request.URL.Path = req.file |
||||
muxer.handleRequest(req.ctx) |
||||
} |
||||
|
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// PathReady is called by pathManager.
|
||||
func (s *Server) PathReady(pa defs.Path) { |
||||
select { |
||||
case s.chPathReady <- pa: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// PathNotReady is called by pathManager.
|
||||
func (s *Server) PathNotReady(pa defs.Path) { |
||||
select { |
||||
case s.chPathNotReady <- pa: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// APIMuxersList is called by api.
|
||||
func (s *Server) APIMuxersList() (*defs.APIHLSMuxerList, error) { |
||||
req := serverAPIMuxersListReq{ |
||||
res: make(chan serverAPIMuxersListRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIMuxerList <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
// APIMuxersGet is called by api.
|
||||
func (s *Server) APIMuxersGet(name string) (*defs.APIHLSMuxer, error) { |
||||
req := serverAPIMuxersGetReq{ |
||||
name: name, |
||||
res: make(chan serverAPIMuxersGetRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIMuxerGet <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
@ -0,0 +1,36 @@
@@ -0,0 +1,36 @@
|
||||
package rtmp |
||||
|
||||
import ( |
||||
"net" |
||||
"sync" |
||||
) |
||||
|
||||
type listener struct { |
||||
ln net.Listener |
||||
wg *sync.WaitGroup |
||||
parent *Server |
||||
} |
||||
|
||||
func (l *listener) initialize() { |
||||
l.wg.Add(1) |
||||
go l.run() |
||||
} |
||||
|
||||
func (l *listener) run() { |
||||
defer l.wg.Done() |
||||
|
||||
err := l.runInner() |
||||
|
||||
l.parent.acceptError(err) |
||||
} |
||||
|
||||
func (l *listener) runInner() error { |
||||
for { |
||||
conn, err := l.ln.Accept() |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
l.parent.newConn(conn) |
||||
} |
||||
} |
||||
@ -0,0 +1,308 @@
@@ -0,0 +1,308 @@
|
||||
// Package rtmp contains a RTMP server.
|
||||
package rtmp |
||||
|
||||
import ( |
||||
"context" |
||||
"crypto/tls" |
||||
"fmt" |
||||
"net" |
||||
"sort" |
||||
"sync" |
||||
|
||||
"github.com/google/uuid" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/defs" |
||||
"github.com/bluenviron/mediamtx/internal/externalcmd" |
||||
"github.com/bluenviron/mediamtx/internal/logger" |
||||
"github.com/bluenviron/mediamtx/internal/restrictnetwork" |
||||
) |
||||
|
||||
type serverAPIConnsListRes struct { |
||||
data *defs.APIRTMPConnList |
||||
err error |
||||
} |
||||
|
||||
type serverAPIConnsListReq struct { |
||||
res chan serverAPIConnsListRes |
||||
} |
||||
|
||||
type serverAPIConnsGetRes struct { |
||||
data *defs.APIRTMPConn |
||||
err error |
||||
} |
||||
|
||||
type serverAPIConnsGetReq struct { |
||||
uuid uuid.UUID |
||||
res chan serverAPIConnsGetRes |
||||
} |
||||
|
||||
type serverAPIConnsKickRes struct { |
||||
err error |
||||
} |
||||
|
||||
type serverAPIConnsKickReq struct { |
||||
uuid uuid.UUID |
||||
res chan serverAPIConnsKickRes |
||||
} |
||||
|
||||
type serverParent interface { |
||||
logger.Writer |
||||
} |
||||
|
||||
// Server is a RTMP server.
|
||||
type Server struct { |
||||
Address string |
||||
ReadTimeout conf.StringDuration |
||||
WriteTimeout conf.StringDuration |
||||
WriteQueueSize int |
||||
IsTLS bool |
||||
ServerCert string |
||||
ServerKey string |
||||
RTSPAddress string |
||||
RunOnConnect string |
||||
RunOnConnectRestart bool |
||||
RunOnDisconnect string |
||||
ExternalCmdPool *externalcmd.Pool |
||||
PathManager defs.PathManager |
||||
Parent serverParent |
||||
|
||||
ctx context.Context |
||||
ctxCancel func() |
||||
wg sync.WaitGroup |
||||
ln net.Listener |
||||
conns map[*conn]struct{} |
||||
|
||||
// in
|
||||
chNewConn chan net.Conn |
||||
chAcceptErr chan error |
||||
chCloseConn chan *conn |
||||
chAPIConnsList chan serverAPIConnsListReq |
||||
chAPIConnsGet chan serverAPIConnsGetReq |
||||
chAPIConnsKick chan serverAPIConnsKickReq |
||||
} |
||||
|
||||
// Initialize initializes the server.
|
||||
func (s *Server) Initialize() error { |
||||
ln, err := func() (net.Listener, error) { |
||||
if !s.IsTLS { |
||||
return net.Listen(restrictnetwork.Restrict("tcp", s.Address)) |
||||
} |
||||
|
||||
cert, err := tls.LoadX509KeyPair(s.ServerCert, s.ServerKey) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
network, address := restrictnetwork.Restrict("tcp", s.Address) |
||||
return tls.Listen(network, address, &tls.Config{Certificates: []tls.Certificate{cert}}) |
||||
}() |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
s.ctx, s.ctxCancel = context.WithCancel(context.Background()) |
||||
|
||||
s.ln = ln |
||||
s.conns = make(map[*conn]struct{}) |
||||
s.chNewConn = make(chan net.Conn) |
||||
s.chAcceptErr = make(chan error) |
||||
s.chCloseConn = make(chan *conn) |
||||
s.chAPIConnsList = make(chan serverAPIConnsListReq) |
||||
s.chAPIConnsGet = make(chan serverAPIConnsGetReq) |
||||
s.chAPIConnsKick = make(chan serverAPIConnsKickReq) |
||||
|
||||
s.Log(logger.Info, "listener opened on %s", s.Address) |
||||
|
||||
l := &listener{ |
||||
ln: s.ln, |
||||
wg: &s.wg, |
||||
parent: s, |
||||
} |
||||
l.initialize() |
||||
|
||||
s.wg.Add(1) |
||||
go s.run() |
||||
|
||||
return nil |
||||
} |
||||
|
||||
// Log implements logger.Writer.
|
||||
func (s *Server) Log(level logger.Level, format string, args ...interface{}) { |
||||
label := func() string { |
||||
if s.IsTLS { |
||||
return "RTMPS" |
||||
} |
||||
return "RTMP" |
||||
}() |
||||
s.Parent.Log(level, "[%s] "+format, append([]interface{}{label}, args...)...) |
||||
} |
||||
|
||||
// Close closes the server.
|
||||
func (s *Server) Close() { |
||||
s.Log(logger.Info, "listener is closing") |
||||
s.ctxCancel() |
||||
s.wg.Wait() |
||||
} |
||||
|
||||
func (s *Server) run() { |
||||
defer s.wg.Done() |
||||
|
||||
outer: |
||||
for { |
||||
select { |
||||
case err := <-s.chAcceptErr: |
||||
s.Log(logger.Error, "%s", err) |
||||
break outer |
||||
|
||||
case nconn := <-s.chNewConn: |
||||
c := &conn{ |
||||
parentCtx: s.ctx, |
||||
isTLS: s.IsTLS, |
||||
rtspAddress: s.RTSPAddress, |
||||
readTimeout: s.ReadTimeout, |
||||
writeTimeout: s.WriteTimeout, |
||||
writeQueueSize: s.WriteQueueSize, |
||||
runOnConnect: s.RunOnConnect, |
||||
runOnConnectRestart: s.RunOnConnectRestart, |
||||
runOnDisconnect: s.RunOnDisconnect, |
||||
wg: &s.wg, |
||||
nconn: nconn, |
||||
externalCmdPool: s.ExternalCmdPool, |
||||
pathManager: s.PathManager, |
||||
parent: s, |
||||
} |
||||
c.initialize() |
||||
s.conns[c] = struct{}{} |
||||
|
||||
case c := <-s.chCloseConn: |
||||
delete(s.conns, c) |
||||
|
||||
case req := <-s.chAPIConnsList: |
||||
data := &defs.APIRTMPConnList{ |
||||
Items: []*defs.APIRTMPConn{}, |
||||
} |
||||
|
||||
for c := range s.conns { |
||||
data.Items = append(data.Items, c.apiItem()) |
||||
} |
||||
|
||||
sort.Slice(data.Items, func(i, j int) bool { |
||||
return data.Items[i].Created.Before(data.Items[j].Created) |
||||
}) |
||||
|
||||
req.res <- serverAPIConnsListRes{data: data} |
||||
|
||||
case req := <-s.chAPIConnsGet: |
||||
c := s.findConnByUUID(req.uuid) |
||||
if c == nil { |
||||
req.res <- serverAPIConnsGetRes{err: fmt.Errorf("connection not found")} |
||||
continue |
||||
} |
||||
|
||||
req.res <- serverAPIConnsGetRes{data: c.apiItem()} |
||||
|
||||
case req := <-s.chAPIConnsKick: |
||||
c := s.findConnByUUID(req.uuid) |
||||
if c == nil { |
||||
req.res <- serverAPIConnsKickRes{err: fmt.Errorf("connection not found")} |
||||
continue |
||||
} |
||||
|
||||
delete(s.conns, c) |
||||
c.Close() |
||||
req.res <- serverAPIConnsKickRes{} |
||||
|
||||
case <-s.ctx.Done(): |
||||
break outer |
||||
} |
||||
} |
||||
|
||||
s.ctxCancel() |
||||
|
||||
s.ln.Close() |
||||
} |
||||
|
||||
func (s *Server) findConnByUUID(uuid uuid.UUID) *conn { |
||||
for c := range s.conns { |
||||
if c.uuid == uuid { |
||||
return c |
||||
} |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
// newConn is called by rtmpListener.
|
||||
func (s *Server) newConn(conn net.Conn) { |
||||
select { |
||||
case s.chNewConn <- conn: |
||||
case <-s.ctx.Done(): |
||||
conn.Close() |
||||
} |
||||
} |
||||
|
||||
// acceptError is called by rtmpListener.
|
||||
func (s *Server) acceptError(err error) { |
||||
select { |
||||
case s.chAcceptErr <- err: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// closeConn is called by conn.
|
||||
func (s *Server) closeConn(c *conn) { |
||||
select { |
||||
case s.chCloseConn <- c: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// APIConnsList is called by api.
|
||||
func (s *Server) APIConnsList() (*defs.APIRTMPConnList, error) { |
||||
req := serverAPIConnsListReq{ |
||||
res: make(chan serverAPIConnsListRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsList <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
// APIConnsGet is called by api.
|
||||
func (s *Server) APIConnsGet(uuid uuid.UUID) (*defs.APIRTMPConn, error) { |
||||
req := serverAPIConnsGetReq{ |
||||
uuid: uuid, |
||||
res: make(chan serverAPIConnsGetRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsGet <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
// APIConnsKick is called by api.
|
||||
func (s *Server) APIConnsKick(uuid uuid.UUID) error { |
||||
req := serverAPIConnsKickReq{ |
||||
uuid: uuid, |
||||
res: make(chan serverAPIConnsKickRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsKick <- req: |
||||
res := <-req.res |
||||
return res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
@ -0,0 +1,430 @@
@@ -0,0 +1,430 @@
|
||||
// Package rtsp contains a RTSP server.
|
||||
package rtsp |
||||
|
||||
import ( |
||||
"context" |
||||
"crypto/tls" |
||||
"fmt" |
||||
"sort" |
||||
"strings" |
||||
"sync" |
||||
"time" |
||||
|
||||
"github.com/bluenviron/gortsplib/v4" |
||||
"github.com/bluenviron/gortsplib/v4/pkg/base" |
||||
"github.com/bluenviron/gortsplib/v4/pkg/headers" |
||||
"github.com/bluenviron/gortsplib/v4/pkg/liberrors" |
||||
"github.com/google/uuid" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/defs" |
||||
"github.com/bluenviron/mediamtx/internal/externalcmd" |
||||
"github.com/bluenviron/mediamtx/internal/logger" |
||||
) |
||||
|
||||
func printAddresses(srv *gortsplib.Server) string { |
||||
var ret []string |
||||
|
||||
ret = append(ret, fmt.Sprintf("%s (TCP)", srv.RTSPAddress)) |
||||
|
||||
if srv.UDPRTPAddress != "" { |
||||
ret = append(ret, fmt.Sprintf("%s (UDP/RTP)", srv.UDPRTPAddress)) |
||||
} |
||||
|
||||
if srv.UDPRTCPAddress != "" { |
||||
ret = append(ret, fmt.Sprintf("%s (UDP/RTCP)", srv.UDPRTCPAddress)) |
||||
} |
||||
|
||||
return strings.Join(ret, ", ") |
||||
} |
||||
|
||||
type serverParent interface { |
||||
logger.Writer |
||||
} |
||||
|
||||
// Server is a RTSP server.
|
||||
type Server struct { |
||||
Address string |
||||
AuthMethods []headers.AuthMethod |
||||
ReadTimeout conf.StringDuration |
||||
WriteTimeout conf.StringDuration |
||||
WriteQueueSize int |
||||
UseUDP bool |
||||
UseMulticast bool |
||||
RTPAddress string |
||||
RTCPAddress string |
||||
MulticastIPRange string |
||||
MulticastRTPPort int |
||||
MulticastRTCPPort int |
||||
IsTLS bool |
||||
ServerCert string |
||||
ServerKey string |
||||
RTSPAddress string |
||||
Protocols map[conf.Protocol]struct{} |
||||
RunOnConnect string |
||||
RunOnConnectRestart bool |
||||
RunOnDisconnect string |
||||
ExternalCmdPool *externalcmd.Pool |
||||
PathManager defs.PathManager |
||||
Parent serverParent |
||||
|
||||
ctx context.Context |
||||
ctxCancel func() |
||||
wg sync.WaitGroup |
||||
srv *gortsplib.Server |
||||
mutex sync.RWMutex |
||||
conns map[*gortsplib.ServerConn]*conn |
||||
sessions map[*gortsplib.ServerSession]*session |
||||
} |
||||
|
||||
// Initialize initializes the server.
|
||||
func (s *Server) Initialize() error { |
||||
s.ctx, s.ctxCancel = context.WithCancel(context.Background()) |
||||
|
||||
s.conns = make(map[*gortsplib.ServerConn]*conn) |
||||
s.sessions = make(map[*gortsplib.ServerSession]*session) |
||||
|
||||
s.srv = &gortsplib.Server{ |
||||
Handler: s, |
||||
ReadTimeout: time.Duration(s.ReadTimeout), |
||||
WriteTimeout: time.Duration(s.WriteTimeout), |
||||
WriteQueueSize: s.WriteQueueSize, |
||||
RTSPAddress: s.Address, |
||||
} |
||||
|
||||
if s.UseUDP { |
||||
s.srv.UDPRTPAddress = s.RTPAddress |
||||
s.srv.UDPRTCPAddress = s.RTCPAddress |
||||
} |
||||
|
||||
if s.UseMulticast { |
||||
s.srv.MulticastIPRange = s.MulticastIPRange |
||||
s.srv.MulticastRTPPort = s.MulticastRTPPort |
||||
s.srv.MulticastRTCPPort = s.MulticastRTCPPort |
||||
} |
||||
|
||||
if s.IsTLS { |
||||
cert, err := tls.LoadX509KeyPair(s.ServerCert, s.ServerKey) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
s.srv.TLSConfig = &tls.Config{Certificates: []tls.Certificate{cert}} |
||||
} |
||||
|
||||
err := s.srv.Start() |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
s.Log(logger.Info, "listener opened on %s", printAddresses(s.srv)) |
||||
|
||||
s.wg.Add(1) |
||||
go s.run() |
||||
|
||||
return nil |
||||
} |
||||
|
||||
// Log implements logger.Writer.
|
||||
func (s *Server) Log(level logger.Level, format string, args ...interface{}) { |
||||
label := func() string { |
||||
if s.IsTLS { |
||||
return "RTSPS" |
||||
} |
||||
return "RTSP" |
||||
}() |
||||
s.Parent.Log(level, "[%s] "+format, append([]interface{}{label}, args...)...) |
||||
} |
||||
|
||||
// Close closes the server.
|
||||
func (s *Server) Close() { |
||||
s.Log(logger.Info, "listener is closing") |
||||
s.ctxCancel() |
||||
s.wg.Wait() |
||||
} |
||||
|
||||
func (s *Server) run() { |
||||
defer s.wg.Done() |
||||
|
||||
serverErr := make(chan error) |
||||
go func() { |
||||
serverErr <- s.srv.Wait() |
||||
}() |
||||
|
||||
outer: |
||||
select { |
||||
case err := <-serverErr: |
||||
s.Log(logger.Error, "%s", err) |
||||
break outer |
||||
|
||||
case <-s.ctx.Done(): |
||||
s.srv.Close() |
||||
<-serverErr |
||||
break outer |
||||
} |
||||
|
||||
s.ctxCancel() |
||||
} |
||||
|
||||
// OnConnOpen implements gortsplib.ServerHandlerOnConnOpen.
|
||||
func (s *Server) OnConnOpen(ctx *gortsplib.ServerHandlerOnConnOpenCtx) { |
||||
c := &conn{ |
||||
isTLS: s.IsTLS, |
||||
rtspAddress: s.RTSPAddress, |
||||
authMethods: s.AuthMethods, |
||||
readTimeout: s.ReadTimeout, |
||||
runOnConnect: s.RunOnConnect, |
||||
runOnConnectRestart: s.RunOnConnectRestart, |
||||
runOnDisconnect: s.RunOnDisconnect, |
||||
externalCmdPool: s.ExternalCmdPool, |
||||
pathManager: s.PathManager, |
||||
rconn: ctx.Conn, |
||||
rserver: s.srv, |
||||
parent: s, |
||||
} |
||||
c.initialize() |
||||
s.mutex.Lock() |
||||
s.conns[ctx.Conn] = c |
||||
s.mutex.Unlock() |
||||
|
||||
ctx.Conn.SetUserData(c) |
||||
} |
||||
|
||||
// OnConnClose implements gortsplib.ServerHandlerOnConnClose.
|
||||
func (s *Server) OnConnClose(ctx *gortsplib.ServerHandlerOnConnCloseCtx) { |
||||
s.mutex.Lock() |
||||
c := s.conns[ctx.Conn] |
||||
delete(s.conns, ctx.Conn) |
||||
s.mutex.Unlock() |
||||
c.onClose(ctx.Error) |
||||
} |
||||
|
||||
// OnRequest implements gortsplib.ServerHandlerOnRequest.
|
||||
func (s *Server) OnRequest(sc *gortsplib.ServerConn, req *base.Request) { |
||||
c := sc.UserData().(*conn) |
||||
c.onRequest(req) |
||||
} |
||||
|
||||
// OnResponse implements gortsplib.ServerHandlerOnResponse.
|
||||
func (s *Server) OnResponse(sc *gortsplib.ServerConn, res *base.Response) { |
||||
c := sc.UserData().(*conn) |
||||
c.OnResponse(res) |
||||
} |
||||
|
||||
// OnSessionOpen implements gortsplib.ServerHandlerOnSessionOpen.
|
||||
func (s *Server) OnSessionOpen(ctx *gortsplib.ServerHandlerOnSessionOpenCtx) { |
||||
se := &session{ |
||||
isTLS: s.IsTLS, |
||||
protocols: s.Protocols, |
||||
rsession: ctx.Session, |
||||
rconn: ctx.Conn, |
||||
rserver: s.srv, |
||||
externalCmdPool: s.ExternalCmdPool, |
||||
pathManager: s.PathManager, |
||||
parent: s, |
||||
} |
||||
se.initialize() |
||||
s.mutex.Lock() |
||||
s.sessions[ctx.Session] = se |
||||
s.mutex.Unlock() |
||||
ctx.Session.SetUserData(se) |
||||
} |
||||
|
||||
// OnSessionClose implements gortsplib.ServerHandlerOnSessionClose.
|
||||
func (s *Server) OnSessionClose(ctx *gortsplib.ServerHandlerOnSessionCloseCtx) { |
||||
s.mutex.Lock() |
||||
se := s.sessions[ctx.Session] |
||||
delete(s.sessions, ctx.Session) |
||||
s.mutex.Unlock() |
||||
|
||||
if se != nil { |
||||
se.onClose(ctx.Error) |
||||
} |
||||
} |
||||
|
||||
// OnDescribe implements gortsplib.ServerHandlerOnDescribe.
|
||||
func (s *Server) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx, |
||||
) (*base.Response, *gortsplib.ServerStream, error) { |
||||
c := ctx.Conn.UserData().(*conn) |
||||
return c.onDescribe(ctx) |
||||
} |
||||
|
||||
// OnAnnounce implements gortsplib.ServerHandlerOnAnnounce.
|
||||
func (s *Server) OnAnnounce(ctx *gortsplib.ServerHandlerOnAnnounceCtx) (*base.Response, error) { |
||||
c := ctx.Conn.UserData().(*conn) |
||||
se := ctx.Session.UserData().(*session) |
||||
return se.onAnnounce(c, ctx) |
||||
} |
||||
|
||||
// OnSetup implements gortsplib.ServerHandlerOnSetup.
|
||||
func (s *Server) OnSetup(ctx *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, error) { |
||||
c := ctx.Conn.UserData().(*conn) |
||||
se := ctx.Session.UserData().(*session) |
||||
return se.onSetup(c, ctx) |
||||
} |
||||
|
||||
// OnPlay implements gortsplib.ServerHandlerOnPlay.
|
||||
func (s *Server) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) { |
||||
se := ctx.Session.UserData().(*session) |
||||
return se.onPlay(ctx) |
||||
} |
||||
|
||||
// OnRecord implements gortsplib.ServerHandlerOnRecord.
|
||||
func (s *Server) OnRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.Response, error) { |
||||
se := ctx.Session.UserData().(*session) |
||||
return se.onRecord(ctx) |
||||
} |
||||
|
||||
// OnPause implements gortsplib.ServerHandlerOnPause.
|
||||
func (s *Server) OnPause(ctx *gortsplib.ServerHandlerOnPauseCtx) (*base.Response, error) { |
||||
se := ctx.Session.UserData().(*session) |
||||
return se.onPause(ctx) |
||||
} |
||||
|
||||
// OnPacketLost implements gortsplib.ServerHandlerOnDecodeError.
|
||||
func (s *Server) OnPacketLost(ctx *gortsplib.ServerHandlerOnPacketLostCtx) { |
||||
se := ctx.Session.UserData().(*session) |
||||
se.onPacketLost(ctx) |
||||
} |
||||
|
||||
// OnDecodeError implements gortsplib.ServerHandlerOnDecodeError.
|
||||
func (s *Server) OnDecodeError(ctx *gortsplib.ServerHandlerOnDecodeErrorCtx) { |
||||
se := ctx.Session.UserData().(*session) |
||||
se.onDecodeError(ctx) |
||||
} |
||||
|
||||
// OnStreamWriteError implements gortsplib.ServerHandlerOnStreamWriteError.
|
||||
func (s *Server) OnStreamWriteError(ctx *gortsplib.ServerHandlerOnStreamWriteErrorCtx) { |
||||
se := ctx.Session.UserData().(*session) |
||||
se.onStreamWriteError(ctx) |
||||
} |
||||
|
||||
func (s *Server) findConnByUUID(uuid uuid.UUID) *conn { |
||||
for _, c := range s.conns { |
||||
if c.uuid == uuid { |
||||
return c |
||||
} |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
func (s *Server) findSessionByUUID(uuid uuid.UUID) (*gortsplib.ServerSession, *session) { |
||||
for key, sx := range s.sessions { |
||||
if sx.uuid == uuid { |
||||
return key, sx |
||||
} |
||||
} |
||||
return nil, nil |
||||
} |
||||
|
||||
// APIConnsList is called by api and metrics.
|
||||
func (s *Server) APIConnsList() (*defs.APIRTSPConnsList, error) { |
||||
select { |
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
default: |
||||
} |
||||
|
||||
s.mutex.RLock() |
||||
defer s.mutex.RUnlock() |
||||
|
||||
data := &defs.APIRTSPConnsList{ |
||||
Items: []*defs.APIRTSPConn{}, |
||||
} |
||||
|
||||
for _, c := range s.conns { |
||||
data.Items = append(data.Items, c.apiItem()) |
||||
} |
||||
|
||||
sort.Slice(data.Items, func(i, j int) bool { |
||||
return data.Items[i].Created.Before(data.Items[j].Created) |
||||
}) |
||||
|
||||
return data, nil |
||||
} |
||||
|
||||
// APIConnsGet is called by api.
|
||||
func (s *Server) APIConnsGet(uuid uuid.UUID) (*defs.APIRTSPConn, error) { |
||||
select { |
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
default: |
||||
} |
||||
|
||||
s.mutex.RLock() |
||||
defer s.mutex.RUnlock() |
||||
|
||||
conn := s.findConnByUUID(uuid) |
||||
if conn == nil { |
||||
return nil, fmt.Errorf("connection not found") |
||||
} |
||||
|
||||
return conn.apiItem(), nil |
||||
} |
||||
|
||||
// APISessionsList is called by api and metrics.
|
||||
func (s *Server) APISessionsList() (*defs.APIRTSPSessionList, error) { |
||||
select { |
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
default: |
||||
} |
||||
|
||||
s.mutex.RLock() |
||||
defer s.mutex.RUnlock() |
||||
|
||||
data := &defs.APIRTSPSessionList{ |
||||
Items: []*defs.APIRTSPSession{}, |
||||
} |
||||
|
||||
for _, s := range s.sessions { |
||||
data.Items = append(data.Items, s.apiItem()) |
||||
} |
||||
|
||||
sort.Slice(data.Items, func(i, j int) bool { |
||||
return data.Items[i].Created.Before(data.Items[j].Created) |
||||
}) |
||||
|
||||
return data, nil |
||||
} |
||||
|
||||
// APISessionsGet is called by api.
|
||||
func (s *Server) APISessionsGet(uuid uuid.UUID) (*defs.APIRTSPSession, error) { |
||||
select { |
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
default: |
||||
} |
||||
|
||||
s.mutex.RLock() |
||||
defer s.mutex.RUnlock() |
||||
|
||||
_, sx := s.findSessionByUUID(uuid) |
||||
if sx == nil { |
||||
return nil, fmt.Errorf("session not found") |
||||
} |
||||
|
||||
return sx.apiItem(), nil |
||||
} |
||||
|
||||
// APISessionsKick is called by api.
|
||||
func (s *Server) APISessionsKick(uuid uuid.UUID) error { |
||||
select { |
||||
case <-s.ctx.Done(): |
||||
return fmt.Errorf("terminated") |
||||
default: |
||||
} |
||||
|
||||
s.mutex.RLock() |
||||
defer s.mutex.RUnlock() |
||||
|
||||
key, sx := s.findSessionByUUID(uuid) |
||||
if sx == nil { |
||||
return fmt.Errorf("session not found") |
||||
} |
||||
|
||||
sx.Close() |
||||
delete(s.sessions, key) |
||||
sx.onClose(liberrors.ErrServerTerminated{}) |
||||
return nil |
||||
} |
||||
@ -0,0 +1,310 @@
@@ -0,0 +1,310 @@
|
||||
// Package srt contains a SRT server.
|
||||
package srt |
||||
|
||||
import ( |
||||
"context" |
||||
"fmt" |
||||
"sort" |
||||
"sync" |
||||
"time" |
||||
|
||||
srt "github.com/datarhei/gosrt" |
||||
"github.com/google/uuid" |
||||
|
||||
"github.com/bluenviron/mediamtx/internal/conf" |
||||
"github.com/bluenviron/mediamtx/internal/defs" |
||||
"github.com/bluenviron/mediamtx/internal/externalcmd" |
||||
"github.com/bluenviron/mediamtx/internal/logger" |
||||
) |
||||
|
||||
func srtMaxPayloadSize(u int) int { |
||||
return ((u - 16) / 188) * 188 // 16 = SRT header, 188 = MPEG-TS packet
|
||||
} |
||||
|
||||
type srtNewConnReq struct { |
||||
connReq srt.ConnRequest |
||||
res chan *conn |
||||
} |
||||
|
||||
type serverAPIConnsListRes struct { |
||||
data *defs.APISRTConnList |
||||
err error |
||||
} |
||||
|
||||
type serverAPIConnsListReq struct { |
||||
res chan serverAPIConnsListRes |
||||
} |
||||
|
||||
type serverAPIConnsGetRes struct { |
||||
data *defs.APISRTConn |
||||
err error |
||||
} |
||||
|
||||
type serverAPIConnsGetReq struct { |
||||
uuid uuid.UUID |
||||
res chan serverAPIConnsGetRes |
||||
} |
||||
|
||||
type serverAPIConnsKickRes struct { |
||||
err error |
||||
} |
||||
|
||||
type serverAPIConnsKickReq struct { |
||||
uuid uuid.UUID |
||||
res chan serverAPIConnsKickRes |
||||
} |
||||
|
||||
type serverParent interface { |
||||
logger.Writer |
||||
} |
||||
|
||||
// Server is a SRT server.
|
||||
type Server struct { |
||||
Address string |
||||
RTSPAddress string |
||||
ReadTimeout conf.StringDuration |
||||
WriteTimeout conf.StringDuration |
||||
WriteQueueSize int |
||||
UDPMaxPayloadSize int |
||||
RunOnConnect string |
||||
RunOnConnectRestart bool |
||||
RunOnDisconnect string |
||||
ExternalCmdPool *externalcmd.Pool |
||||
PathManager defs.PathManager |
||||
Parent serverParent |
||||
|
||||
ctx context.Context |
||||
ctxCancel func() |
||||
wg sync.WaitGroup |
||||
ln srt.Listener |
||||
conns map[*conn]struct{} |
||||
|
||||
// in
|
||||
chNewConnRequest chan srtNewConnReq |
||||
chAcceptErr chan error |
||||
chCloseConn chan *conn |
||||
chAPIConnsList chan serverAPIConnsListReq |
||||
chAPIConnsGet chan serverAPIConnsGetReq |
||||
chAPIConnsKick chan serverAPIConnsKickReq |
||||
} |
||||
|
||||
// Initialize initializes the server.
|
||||
func (s *Server) Initialize() error { |
||||
conf := srt.DefaultConfig() |
||||
conf.ConnectionTimeout = time.Duration(s.ReadTimeout) |
||||
conf.PayloadSize = uint32(srtMaxPayloadSize(s.UDPMaxPayloadSize)) |
||||
|
||||
var err error |
||||
s.ln, err = srt.Listen("srt", s.Address, conf) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
s.ctx, s.ctxCancel = context.WithCancel(context.Background()) |
||||
|
||||
s.conns = make(map[*conn]struct{}) |
||||
s.chNewConnRequest = make(chan srtNewConnReq) |
||||
s.chAcceptErr = make(chan error) |
||||
s.chCloseConn = make(chan *conn) |
||||
s.chAPIConnsList = make(chan serverAPIConnsListReq) |
||||
s.chAPIConnsGet = make(chan serverAPIConnsGetReq) |
||||
s.chAPIConnsKick = make(chan serverAPIConnsKickReq) |
||||
|
||||
s.Log(logger.Info, "listener opened on "+s.Address+" (UDP)") |
||||
|
||||
l := &listener{ |
||||
ln: s.ln, |
||||
wg: &s.wg, |
||||
parent: s, |
||||
} |
||||
l.initialize() |
||||
|
||||
s.wg.Add(1) |
||||
go s.run() |
||||
|
||||
return nil |
||||
} |
||||
|
||||
// Log implements logger.Writer.
|
||||
func (s *Server) Log(level logger.Level, format string, args ...interface{}) { |
||||
s.Parent.Log(level, "[SRT] "+format, args...) |
||||
} |
||||
|
||||
// Close closes the server.
|
||||
func (s *Server) Close() { |
||||
s.Log(logger.Info, "listener is closing") |
||||
s.ctxCancel() |
||||
s.wg.Wait() |
||||
} |
||||
|
||||
func (s *Server) run() { |
||||
defer s.wg.Done() |
||||
|
||||
outer: |
||||
for { |
||||
select { |
||||
case err := <-s.chAcceptErr: |
||||
s.Log(logger.Error, "%s", err) |
||||
break outer |
||||
|
||||
case req := <-s.chNewConnRequest: |
||||
c := &conn{ |
||||
parentCtx: s.ctx, |
||||
rtspAddress: s.RTSPAddress, |
||||
readTimeout: s.ReadTimeout, |
||||
writeTimeout: s.WriteTimeout, |
||||
writeQueueSize: s.WriteQueueSize, |
||||
udpMaxPayloadSize: s.UDPMaxPayloadSize, |
||||
connReq: req.connReq, |
||||
runOnConnect: s.RunOnConnect, |
||||
runOnConnectRestart: s.RunOnConnectRestart, |
||||
runOnDisconnect: s.RunOnDisconnect, |
||||
wg: &s.wg, |
||||
externalCmdPool: s.ExternalCmdPool, |
||||
pathManager: s.PathManager, |
||||
parent: s, |
||||
} |
||||
c.initialize() |
||||
s.conns[c] = struct{}{} |
||||
req.res <- c |
||||
|
||||
case c := <-s.chCloseConn: |
||||
delete(s.conns, c) |
||||
|
||||
case req := <-s.chAPIConnsList: |
||||
data := &defs.APISRTConnList{ |
||||
Items: []*defs.APISRTConn{}, |
||||
} |
||||
|
||||
for c := range s.conns { |
||||
data.Items = append(data.Items, c.apiItem()) |
||||
} |
||||
|
||||
sort.Slice(data.Items, func(i, j int) bool { |
||||
return data.Items[i].Created.Before(data.Items[j].Created) |
||||
}) |
||||
|
||||
req.res <- serverAPIConnsListRes{data: data} |
||||
|
||||
case req := <-s.chAPIConnsGet: |
||||
c := s.findConnByUUID(req.uuid) |
||||
if c == nil { |
||||
req.res <- serverAPIConnsGetRes{err: fmt.Errorf("connection not found")} |
||||
continue |
||||
} |
||||
|
||||
req.res <- serverAPIConnsGetRes{data: c.apiItem()} |
||||
|
||||
case req := <-s.chAPIConnsKick: |
||||
c := s.findConnByUUID(req.uuid) |
||||
if c == nil { |
||||
req.res <- serverAPIConnsKickRes{err: fmt.Errorf("connection not found")} |
||||
continue |
||||
} |
||||
|
||||
delete(s.conns, c) |
||||
c.Close() |
||||
req.res <- serverAPIConnsKickRes{} |
||||
|
||||
case <-s.ctx.Done(): |
||||
break outer |
||||
} |
||||
} |
||||
|
||||
s.ctxCancel() |
||||
|
||||
s.ln.Close() |
||||
} |
||||
|
||||
func (s *Server) findConnByUUID(uuid uuid.UUID) *conn { |
||||
for sx := range s.conns { |
||||
if sx.uuid == uuid { |
||||
return sx |
||||
} |
||||
} |
||||
return nil |
||||
} |
||||
|
||||
// newConnRequest is called by srtListener.
|
||||
func (s *Server) newConnRequest(connReq srt.ConnRequest) *conn { |
||||
req := srtNewConnReq{ |
||||
connReq: connReq, |
||||
res: make(chan *conn), |
||||
} |
||||
|
||||
select { |
||||
case s.chNewConnRequest <- req: |
||||
c := <-req.res |
||||
|
||||
return c.new(req) |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil |
||||
} |
||||
} |
||||
|
||||
// acceptError is called by srtListener.
|
||||
func (s *Server) acceptError(err error) { |
||||
select { |
||||
case s.chAcceptErr <- err: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// closeConn is called by conn.
|
||||
func (s *Server) closeConn(c *conn) { |
||||
select { |
||||
case s.chCloseConn <- c: |
||||
case <-s.ctx.Done(): |
||||
} |
||||
} |
||||
|
||||
// APIConnsList is called by api.
|
||||
func (s *Server) APIConnsList() (*defs.APISRTConnList, error) { |
||||
req := serverAPIConnsListReq{ |
||||
res: make(chan serverAPIConnsListRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsList <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
// APIConnsGet is called by api.
|
||||
func (s *Server) APIConnsGet(uuid uuid.UUID) (*defs.APISRTConn, error) { |
||||
req := serverAPIConnsGetReq{ |
||||
uuid: uuid, |
||||
res: make(chan serverAPIConnsGetRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsGet <- req: |
||||
res := <-req.res |
||||
return res.data, res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return nil, fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
|
||||
// APIConnsKick is called by api.
|
||||
func (s *Server) APIConnsKick(uuid uuid.UUID) error { |
||||
req := serverAPIConnsKickReq{ |
||||
uuid: uuid, |
||||
res: make(chan serverAPIConnsKickRes), |
||||
} |
||||
|
||||
select { |
||||
case s.chAPIConnsKick <- req: |
||||
res := <-req.res |
||||
return res.err |
||||
|
||||
case <-s.ctx.Done(): |
||||
return fmt.Errorf("terminated") |
||||
} |
||||
} |
||||
Loading…
Reference in new issue