pimhwang
/
mediamtx
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

hls: rename remuxer into muxer

pull/523/head
aler9 5 years ago
parent
ca499a27c3
commit
4fdd42fa58
3 changed files with 62 additions and 62 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      apidocs/openapi.yaml
  2. 84
      internal/core/hls_muxer.go
  3. 38
      internal/core/hls_server.go

2
apidocs/openapi.yaml
Unescape View File

@ -254,7 +254,7 @@ components:
properties: properties:
type: type:
type: string type: string
enum: [hlsmuxer] enum: [hlsMuxer]
RTSPSession: RTSPSession:
type: object type: object

84
internal/core/hls_remuxer.go → internal/core/hls_muxer.go
Unescape View File

@ -88,7 +88,7 @@ create();
</html> </html>
` `
type hlsRemuxerRequest struct { type hlsMuxerRequest struct {
Dir string Dir string
File string File string
Req *http.Request Req *http.Request
@ -96,29 +96,29 @@ type hlsRemuxerRequest struct {
Res chan io.Reader Res chan io.Reader
} }
type hlsRemuxerTrackIDPayloadPair struct { type hlsMuxerTrackIDPayloadPair struct {
trackID int trackID int
buf []byte buf []byte
} }
type hlsRemuxerPathManager interface { type hlsMuxerPathManager interface {
OnReaderSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes OnReaderSetupPlay(req pathReaderSetupPlayReq) pathReaderSetupPlayRes
} }
type hlsRemuxerParent interface { type hlsMuxerParent interface {
Log(logger.Level, string, ...interface{}) Log(logger.Level, string, ...interface{})
OnRemuxerClose(*hlsRemuxer) OnMuxerClose(*hlsMuxer)
} }
type hlsRemuxer struct { type hlsMuxer struct {
hlsAlwaysRemux bool hlsAlwaysRemux bool
hlsSegmentCount int hlsSegmentCount int
hlsSegmentDuration time.Duration hlsSegmentDuration time.Duration
readBufferCount int readBufferCount int
wg *sync.WaitGroup wg *sync.WaitGroup
pathName string pathName string
pathManager hlsRemuxerPathManager pathManager hlsMuxerPathManager
parent hlsRemuxerParent parent hlsMuxerParent
ctx context.Context ctx context.Context
ctxCancel func() ctxCancel func()
@ -126,13 +126,13 @@ type hlsRemuxer struct {
ringBuffer *ringbuffer.RingBuffer ringBuffer *ringbuffer.RingBuffer
lastRequestTime *int64 lastRequestTime *int64
muxer *hls.Muxer muxer *hls.Muxer
requests []hlsRemuxerRequest requests []hlsMuxerRequest
// in // in
request chan hlsRemuxerRequest request chan hlsMuxerRequest
} }
func newHLSRemuxer( func newHLSMuxer(
parentCtx context.Context, parentCtx context.Context,
hlsAlwaysRemux bool, hlsAlwaysRemux bool,
hlsSegmentCount int, hlsSegmentCount int,
@ -140,11 +140,11 @@ func newHLSRemuxer(
readBufferCount int, readBufferCount int,
wg *sync.WaitGroup, wg *sync.WaitGroup,
pathName string, pathName string,
pathManager hlsRemuxerPathManager, pathManager hlsMuxerPathManager,
parent hlsRemuxerParent) *hlsRemuxer { parent hlsMuxerParent) *hlsMuxer {
ctx, ctxCancel := context.WithCancel(parentCtx) ctx, ctxCancel := context.WithCancel(parentCtx)
r := &hlsRemuxer{ r := &hlsMuxer{
hlsAlwaysRemux: hlsAlwaysRemux, hlsAlwaysRemux: hlsAlwaysRemux,
hlsSegmentCount: hlsSegmentCount, hlsSegmentCount: hlsSegmentCount,
hlsSegmentDuration: hlsSegmentDuration, hlsSegmentDuration: hlsSegmentDuration,
@ -159,7 +159,7 @@ func newHLSRemuxer(
v := time.Now().Unix() v := time.Now().Unix()
return &v return &v
}(), }(),
request: make(chan hlsRemuxerRequest), request: make(chan hlsMuxerRequest),
} }
r.log(logger.Info, "created") r.log(logger.Info, "created")
@ -170,28 +170,28 @@ func newHLSRemuxer(
return r return r
} }
func (r *hlsRemuxer) Close() { func (r *hlsMuxer) Close() {
r.ctxCancel() r.ctxCancel()
} }
func (r *hlsRemuxer) log(level logger.Level, format string, args ...interface{}) { func (r *hlsMuxer) log(level logger.Level, format string, args ...interface{}) {
r.parent.Log(level, "[remuxer %s] "+format, append([]interface{}{r.pathName}, args...)...) r.parent.Log(level, "[muxer %s] "+format, append([]interface{}{r.pathName}, args...)...)
} }
// PathName returns the path name. // PathName returns the path name.
func (r *hlsRemuxer) PathName() string { func (r *hlsMuxer) PathName() string {
return r.pathName return r.pathName
} }
func (r *hlsRemuxer) run() { func (r *hlsMuxer) run() {
defer r.wg.Done() defer r.wg.Done()
defer r.log(logger.Info, "destroyed") defer r.log(logger.Info, "destroyed")
remuxerCtx, remuxerCtxCancel := context.WithCancel(context.Background()) innerCtx, innerCtxCancel := context.WithCancel(context.Background())
remuxerReady := make(chan struct{}) innerReady := make(chan struct{})
remuxerErr := make(chan error) innerErr := make(chan error)
go func() { go func() {
remuxerErr <- r.runRemuxer(remuxerCtx, remuxerReady) innerErr <- r.runInner(innerCtx, innerReady)
}() }()
isReady := false isReady := false
@ -200,8 +200,8 @@ outer:
for { for {
select { select {
case <-r.ctx.Done(): case <-r.ctx.Done():
remuxerCtxCancel() innerCtxCancel()
<-remuxerErr <-innerErr
break outer break outer
case req := <-r.request: case req := <-r.request:
@ -211,15 +211,15 @@ outer:
r.requests = append(r.requests, req) r.requests = append(r.requests, req)
} }
case <-remuxerReady: case <-innerReady:
isReady = true isReady = true
for _, req := range r.requests { for _, req := range r.requests {
r.handleRequest(req) r.handleRequest(req)
} }
r.requests = nil r.requests = nil
case err := <-remuxerErr: case err := <-innerErr:
remuxerCtxCancel() innerCtxCancel()
if err != nil { if err != nil {
r.log(logger.Info, "ERR: %s", err) r.log(logger.Info, "ERR: %s", err)
} }
@ -234,10 +234,10 @@ outer:
req.Res <- nil req.Res <- nil
} }
r.parent.OnRemuxerClose(r) r.parent.OnMuxerClose(r)
} }
func (r *hlsRemuxer) runRemuxer(remuxerCtx context.Context, remuxerReady chan struct{}) error { func (r *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{}) error {
res := r.pathManager.OnReaderSetupPlay(pathReaderSetupPlayReq{ res := r.pathManager.OnReaderSetupPlay(pathReaderSetupPlayReq{
Author: r, Author: r,
PathName: r.pathName, PathName: r.pathName,
@ -311,7 +311,7 @@ func (r *hlsRemuxer) runRemuxer(remuxerCtx context.Context, remuxerReady chan st
} }
defer r.muxer.Close() defer r.muxer.Close()
remuxerReady <- struct{}{} innerReady <- struct{}{}
r.ringBuffer = ringbuffer.New(uint64(r.readBufferCount)) r.ringBuffer = ringbuffer.New(uint64(r.readBufferCount))
@ -327,7 +327,7 @@ func (r *hlsRemuxer) runRemuxer(remuxerCtx context.Context, remuxerReady chan st
if !ok { if !ok {
return fmt.Errorf("terminated") return fmt.Errorf("terminated")
} }
pair := data.(hlsRemuxerTrackIDPayloadPair) pair := data.(hlsMuxerTrackIDPayloadPair)
if videoTrack != nil && pair.trackID == videoTrackID { if videoTrack != nil && pair.trackID == videoTrackID {
var pkt rtp.Packet var pkt rtp.Packet
@ -399,7 +399,7 @@ func (r *hlsRemuxer) runRemuxer(remuxerCtx context.Context, remuxerReady chan st
case err := <-writerDone: case err := <-writerDone:
return err return err
case <-remuxerCtx.Done(): case <-innerCtx.Done():
r.ringBuffer.Close() r.ringBuffer.Close()
<-writerDone <-writerDone
return nil return nil
@ -407,7 +407,7 @@ func (r *hlsRemuxer) runRemuxer(remuxerCtx context.Context, remuxerReady chan st
} }
} }
func (r *hlsRemuxer) handleRequest(req hlsRemuxerRequest) { func (r *hlsMuxer) handleRequest(req hlsMuxerRequest) {
atomic.StoreInt64(r.lastRequestTime, time.Now().Unix()) atomic.StoreInt64(r.lastRequestTime, time.Now().Unix())
conf := r.path.Conf() conf := r.path.Conf()
@ -463,7 +463,7 @@ func (r *hlsRemuxer) handleRequest(req hlsRemuxerRequest) {
} }
// OnRequest is called by hlsserver.Server (forwarded from ServeHTTP). // OnRequest is called by hlsserver.Server (forwarded from ServeHTTP).
func (r *hlsRemuxer) OnRequest(req hlsRemuxerRequest) { func (r *hlsMuxer) OnRequest(req hlsMuxerRequest) {
select { select {
case r.request <- req: case r.request <- req:
case <-r.ctx.Done(): case <-r.ctx.Done():
@ -473,20 +473,20 @@ func (r *hlsRemuxer) OnRequest(req hlsRemuxerRequest) {
} }
// OnReaderAccepted implements reader. // OnReaderAccepted implements reader.
func (r *hlsRemuxer) OnReaderAccepted() { func (r *hlsMuxer) OnReaderAccepted() {
r.log(logger.Info, "is remuxing into HLS") r.log(logger.Info, "is converting into HLS")
} }
// OnReaderFrame implements reader. // OnReaderFrame implements reader.
func (r *hlsRemuxer) OnReaderFrame(trackID int, streamType gortsplib.StreamType, payload []byte) { func (r *hlsMuxer) OnReaderFrame(trackID int, streamType gortsplib.StreamType, payload []byte) {
if streamType == gortsplib.StreamTypeRTP { if streamType == gortsplib.StreamTypeRTP {
r.ringBuffer.Push(hlsRemuxerTrackIDPayloadPair{trackID, payload}) r.ringBuffer.Push(hlsMuxerTrackIDPayloadPair{trackID, payload})
} }
} }
// OnReaderAPIDescribe implements reader. // OnReaderAPIDescribe implements reader.
func (r *hlsRemuxer) OnReaderAPIDescribe() interface{} { func (r *hlsMuxer) OnReaderAPIDescribe() interface{} {
return struct { return struct {
Type string `json:"type"` Type string `json:"type"`
}{"hlsremuxer"} }{"hlsMuxer"}
} }

38
internal/core/hls_server.go
Unescape View File

@ -30,12 +30,12 @@ type hlsServer struct {
ctxCancel func() ctxCancel func()
wg sync.WaitGroup wg sync.WaitGroup
ln net.Listener ln net.Listener
remuxers map[string]*hlsRemuxer muxers map[string]*hlsMuxer
// in // in
pathSourceReady chan *path pathSourceReady chan *path
request chan hlsRemuxerRequest request chan hlsMuxerRequest
remuxerClose chan *hlsRemuxer muxerClose chan *hlsMuxer
} }
func newHLSServer( func newHLSServer(
@ -67,10 +67,10 @@ func newHLSServer(
ctx: ctx, ctx: ctx,
ctxCancel: ctxCancel, ctxCancel: ctxCancel,
ln: ln, ln: ln,
remuxers: make(map[string]*hlsRemuxer), muxers: make(map[string]*hlsMuxer),
pathSourceReady: make(chan *path), pathSourceReady: make(chan *path),
request: make(chan hlsRemuxerRequest), request: make(chan hlsMuxerRequest),
remuxerClose: make(chan *hlsRemuxer), muxerClose: make(chan *hlsMuxer),
} }
s.Log(logger.Info, "listener opened on "+address) s.Log(logger.Info, "listener opened on "+address)
@ -105,18 +105,18 @@ outer:
select { select {
case pa := <-s.pathSourceReady: case pa := <-s.pathSourceReady:
if s.hlsAlwaysRemux { if s.hlsAlwaysRemux {
s.findOrCreateRemuxer(pa.Name()) s.findOrCreateMuxer(pa.Name())
} }
case req := <-s.request: case req := <-s.request:
r := s.findOrCreateRemuxer(req.Dir) r := s.findOrCreateMuxer(req.Dir)
r.OnRequest(req) r.OnRequest(req)
case c := <-s.remuxerClose: case c := <-s.muxerClose:
if c2, ok := s.remuxers[c.PathName()]; !ok || c2 != c { if c2, ok := s.muxers[c.PathName()]; !ok || c2 != c {
continue continue
} }
delete(s.remuxers, c.PathName()) delete(s.muxers, c.PathName())
case <-s.ctx.Done(): case <-s.ctx.Done():
break outer break outer
@ -176,7 +176,7 @@ func (s *hlsServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
dir = strings.TrimSuffix(dir, "/") dir = strings.TrimSuffix(dir, "/")
cres := make(chan io.Reader) cres := make(chan io.Reader)
hreq := hlsRemuxerRequest{ hreq := hlsMuxerRequest{
Dir: dir, Dir: dir,
File: fname, File: fname,
Req: r, Req: r,
@ -209,10 +209,10 @@ func (s *hlsServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
} }
} }
func (s *hlsServer) findOrCreateRemuxer(pathName string) *hlsRemuxer { func (s *hlsServer) findOrCreateMuxer(pathName string) *hlsMuxer {
r, ok := s.remuxers[pathName] r, ok := s.muxers[pathName]
if !ok { if !ok {
r = newHLSRemuxer( r = newHLSMuxer(
s.ctx, s.ctx,
s.hlsAlwaysRemux, s.hlsAlwaysRemux,
s.hlsSegmentCount, s.hlsSegmentCount,
@ -222,15 +222,15 @@ func (s *hlsServer) findOrCreateRemuxer(pathName string) *hlsRemuxer {
pathName, pathName,
s.pathManager, s.pathManager,
s) s)
s.remuxers[pathName] = r s.muxers[pathName] = r
} }
return r return r
} }
// OnRemuxerClose is called by hlsRemuxer. // OnMuxerClose is called by hlsMuxer.
func (s *hlsServer) OnRemuxerClose(c *hlsRemuxer) { func (s *hlsServer) OnMuxerClose(c *hlsMuxer) {
select { select {
case s.remuxerClose <- c: case s.muxerClose <- c:
case <-s.ctx.Done(): case <-s.ctx.Done():
} }
} }

Write Preview
Loading…
Cancel
Save