hls: when alwaysRemux is true, prevent clients from creating muxers (#3015)

internal/servers/hls/http_server.go

@@ -191,10 +191,24 @@ func (s *httpServer) onRequest(ctx *gin.Context) {
 		ctx.Writer.Write(hlsIndex)
 
 	default:
-		s.parent.handleRequest(muxerHandleRequestReq{
-			path: dir,
-			file: fname,
-			ctx:  ctx,
+		mux, err := s.parent.getMuxer(serverGetMuxerReq{
+			path:       dir,
+			remoteAddr: httpp.RemoteAddr(ctx),
 		})
+		if err != nil {
+			s.Log(logger.Error, err.Error())
+			ctx.Writer.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+
+		err = mux.processRequest(muxerProcessRequestReq{})
+		if err != nil {
+			s.Log(logger.Error, err.Error())
+			ctx.Writer.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+
+		ctx.Request.URL.Path = fname
+		mux.handleRequest(ctx)
 	}
 }

internal/servers/hls/muxer.go

@@ -51,11 +51,8 @@ func (w *responseWriterWithCounter) Write(p []byte) (int, error) {
 	return n, err
 }
 
-type muxerHandleRequestReq struct {
-	path string
-	file string
-	ctx  *gin.Context
-	res  chan *muxer
+type muxerProcessRequestReq struct {
+	res chan error
 }
 
 type muxer struct {
@@ -81,11 +78,11 @@ type muxer struct {
 	writer          *asyncwriter.Writer
 	lastRequestTime *int64
 	muxer           *gohlslib.Muxer
-	requests        []*muxerHandleRequestReq
+	requests        []*muxerProcessRequestReq
 	bytesSent       *uint64
 
 	// in
-	chRequest chan *muxerHandleRequestReq
+	chProcessRequest chan muxerProcessRequestReq
 }
 
 func (m *muxer) initialize() {
@@ -96,7 +93,7 @@ func (m *muxer) initialize() {
 	m.created = time.Now()
 	m.lastRequestTime = int64Ptr(time.Now().UnixNano())
 	m.bytesSent = new(uint64)
-	m.chRequest = make(chan *muxerHandleRequestReq)
+	m.chProcessRequest = make(chan muxerProcessRequestReq)
 
 	m.Log(logger.Info, "created %s", func() string {
 		if m.remoteAddr == "" {
@@ -156,22 +153,22 @@ func (m *muxer) run() {
 				}
 				return errors.New("terminated")
 
-			case req := <-m.chRequest:
+			case req := <-m.chProcessRequest:
 				switch {
 				case isRecreating:
-					req.res <- nil
+					req.res <- fmt.Errorf("recreating")
 
 				case isReady:
-					req.res <- m
+					req.res <- nil
 
 				default:
-					m.requests = append(m.requests, req)
+					m.requests = append(m.requests, &req)
 				}
 
 			case <-innerReady:
 				isReady = true
 				for _, req := range m.requests {
-					req.res <- m
+					req.res <- nil
 				}
 				m.requests = nil
 
@@ -206,7 +203,7 @@ func (m *muxer) run() {
 
 func (m *muxer) clearQueuedRequests() {
 	for _, req := range m.requests {
-		req.res <- nil
+		req.res <- fmt.Errorf("terminated")
 	}
 	m.requests = nil
 }
@@ -479,12 +476,15 @@ func (m *muxer) handleRequest(ctx *gin.Context) {
 	m.muxer.Handle(w, ctx.Request)
 }
 
-// processRequest is called by hlsserver.Server (forwarded from ServeHTTP).
-func (m *muxer) processRequest(req *muxerHandleRequestReq) {
+func (m *muxer) processRequest(req muxerProcessRequestReq) error {
+	req.res = make(chan error)
+
 	select {
-	case m.chRequest <- req:
+	case m.chProcessRequest <- req:
+		return <-req.res
+
 	case <-m.ctx.Done():
-		req.res <- nil
+		return fmt.Errorf("terminated")
 	}
 }
 

internal/servers/hls/server.go

@@ -11,12 +11,22 @@ import (
 	"github.com/bluenviron/mediamtx/internal/conf"
 	"github.com/bluenviron/mediamtx/internal/defs"
 	"github.com/bluenviron/mediamtx/internal/logger"
-	"github.com/bluenviron/mediamtx/internal/protocols/httpp"
 )
 
 // ErrMuxerNotFound is returned when a muxer is not found.
 var ErrMuxerNotFound = errors.New("muxer not found")
 
+type serverGetMuxerRes struct {
+	muxer *muxer
+	err   error
+}
+
+type serverGetMuxerReq struct {
+	path       string
+	remoteAddr string
+	res        chan serverGetMuxerRes
+}
+
 type serverAPIMuxersListRes struct {
 	data *defs.APIHLSMuxerList
 	err  error
@@ -68,12 +78,12 @@ type Server struct {
 	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
+	chPathReady    chan defs.Path
+	chPathNotReady chan defs.Path
+	chGetMuxer     chan serverGetMuxerReq
+	chCloseMuxer   chan *muxer
+	chAPIMuxerList chan serverAPIMuxersListReq
+	chAPIMuxerGet  chan serverAPIMuxersGetReq
 }
 
 // Initialize initializes the server.
@@ -85,7 +95,7 @@ func (s *Server) Initialize() error {
 	s.muxers = make(map[string]*muxer)
 	s.chPathReady = make(chan defs.Path)
 	s.chPathNotReady = make(chan defs.Path)
-	s.chHandleRequest = make(chan muxerHandleRequestReq)
+	s.chGetMuxer = make(chan serverGetMuxerReq)
 	s.chCloseMuxer = make(chan *muxer)
 	s.chAPIMuxerList = make(chan serverAPIMuxersListReq)
 	s.chAPIMuxerGet = make(chan serverAPIMuxersGetReq)
@@ -147,15 +157,15 @@ outer:
 				delete(s.muxers, pa.Name())
 			}
 
-		case req := <-s.chHandleRequest:
-			r, ok := s.muxers[req.path]
+		case req := <-s.chGetMuxer:
+			mux, ok := s.muxers[req.path]
 			switch {
 			case ok:
-				r.processRequest(&req)
-
+				req.res <- serverGetMuxerRes{muxer: mux}
+			case !s.AlwaysRemux:
+				req.res <- serverGetMuxerRes{muxer: s.createMuxer(req.path, req.remoteAddr)}
 			default:
-				r := s.createMuxer(req.path, httpp.RemoteAddr(req.ctx))
-				r.processRequest(&req)
+				req.res <- serverGetMuxerRes{err: fmt.Errorf("muxer is waiting to be created")}
 			}
 
 		case c := <-s.chCloseMuxer:
@@ -230,18 +240,16 @@ func (s *Server) closeMuxer(c *muxer) {
 	}
 }
 
-func (s *Server) handleRequest(req muxerHandleRequestReq) {
-	req.res = make(chan *muxer)
+func (s *Server) getMuxer(req serverGetMuxerReq) (*muxer, error) {
+	req.res = make(chan serverGetMuxerRes)
 
 	select {
-	case s.chHandleRequest <- req:
-		muxer := <-req.res
-		if muxer != nil {
-			req.ctx.Request.URL.Path = req.file
-			muxer.handleRequest(req.ctx)
-		}
+	case s.chGetMuxer <- req:
+		res := <-req.res
+		return res.muxer, res.err
 
 	case <-s.ctx.Done():
+		return nil, fmt.Errorf("terminated")
 	}
 }