hls: fix crash when muxer is being recreated, improve performance (#3017)

.github/workflows/code_lint.yml

@@ -17,6 +17,8 @@ jobs:
       with:
         go-version: "1.19"
 
+    - run: touch internal/servers/hls/hls.min.js
+
     - uses: golangci/golangci-lint-action@v3
       with:
         version: v1.55.0

internal/core/api_test.go

@@ -18,7 +18,6 @@ import (
 	"github.com/bluenviron/gortsplib/v4"
 	"github.com/bluenviron/gortsplib/v4/pkg/description"
 	"github.com/bluenviron/gortsplib/v4/pkg/format"
-	"github.com/bluenviron/mediacommon/pkg/codecs/mpeg4audio"
 	"github.com/bluenviron/mediacommon/pkg/formats/mpegts"
 	srt "github.com/datarhei/gosrt"
 	"github.com/google/uuid"
@@ -30,35 +29,14 @@ import (
 	"github.com/bluenviron/mediamtx/internal/test"
 )
 
-var testFormatH264 = &format.H264{
-	PayloadTyp: 96,
-	SPS: []byte{ // 1920x1080 baseline
-		0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-		0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-		0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20,
-	},
-	PPS:               []byte{0x08, 0x06, 0x07, 0x08},
-	PacketizationMode: 1,
-}
-
 var testMediaH264 = &description.Media{
 	Type:    description.MediaTypeVideo,
-	Formats: []format.Format{testFormatH264},
+	Formats: []format.Format{test.FormatH264},
 }
 
 var testMediaAAC = &description.Media{
-	Type: description.MediaTypeAudio,
-	Formats: []format.Format{&format.MPEG4Audio{
-		PayloadTyp: 96,
-		Config: &mpeg4audio.Config{
-			Type:         2,
-			SampleRate:   44100,
-			ChannelCount: 2,
-		},
-		SizeLength:       13,
-		IndexLength:      3,
-		IndexDeltaLength: 3,
-	}},
+	Type:    description.MediaTypeAudio,
+	Formats: []format.Format{test.FormatMPEG4Audio},
 }
 
 func checkClose(t *testing.T, closeFunc func() error) {
@@ -450,7 +428,7 @@ func TestAPIProtocolList(t *testing.T) {
 				conn, err := rtmp.NewClientConn(nconn, u, true)
 				require.NoError(t, err)
 
-				_, err = rtmp.NewWriter(conn, testFormatH264, nil)
+				_, err = rtmp.NewWriter(conn, test.FormatH264, nil)
 				require.NoError(t, err)
 
 				time.Sleep(500 * time.Millisecond)
@@ -750,7 +728,7 @@ func TestAPIProtocolGet(t *testing.T) {
 				conn, err := rtmp.NewClientConn(nconn, u, true)
 				require.NoError(t, err)
 
-				_, err = rtmp.NewWriter(conn, testFormatH264, nil)
+				_, err = rtmp.NewWriter(conn, test.FormatH264, nil)
 				require.NoError(t, err)
 
 				time.Sleep(500 * time.Millisecond)
@@ -1127,7 +1105,7 @@ func TestAPIProtocolKick(t *testing.T) {
 				conn, err := rtmp.NewClientConn(nconn, u, true)
 				require.NoError(t, err)
 
-				_, err = rtmp.NewWriter(conn, testFormatH264, nil)
+				_, err = rtmp.NewWriter(conn, test.FormatH264, nil)
 				require.NoError(t, err)
 
 			case "webrtc":

internal/core/hls_server_test.go

@@ -1,222 +0,0 @@
-package core
-
-import (
-	"context"
-	"encoding/json"
-	"io"
-	"net"
-	"net/http"
-	"testing"
-	"time"
-
-	"github.com/bluenviron/gortsplib/v4"
-	"github.com/bluenviron/gortsplib/v4/pkg/description"
-	"github.com/bluenviron/gortsplib/v4/pkg/format"
-	"github.com/pion/rtp"
-	"github.com/stretchr/testify/require"
-)
-
-type testHTTPAuthenticator struct {
-	*http.Server
-}
-
-func newTestHTTPAuthenticator(t *testing.T, protocol string, action string) *testHTTPAuthenticator {
-	firstReceived := false
-
-	ts := &testHTTPAuthenticator{}
-
-	ts.Server = &http.Server{
-		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			require.Equal(t, http.MethodPost, r.Method)
-			require.Equal(t, "/auth", r.URL.Path)
-
-			var in struct {
-				IP       string `json:"ip"`
-				User     string `json:"user"`
-				Password string `json:"password"`
-				Path     string `json:"path"`
-				Protocol string `json:"protocol"`
-				ID       string `json:"id"`
-				Action   string `json:"action"`
-				Query    string `json:"query"`
-			}
-			err := json.NewDecoder(r.Body).Decode(&in)
-			require.NoError(t, err)
-
-			var user string
-			if action == "publish" {
-				user = "testpublisher"
-			} else {
-				user = "testreader"
-			}
-
-			if in.IP != "127.0.0.1" ||
-				in.User != user ||
-				in.Password != "testpass" ||
-				in.Path != "teststream" ||
-				in.Protocol != protocol ||
-				(firstReceived && in.ID == "") ||
-				in.Action != action ||
-				(in.Query != "user=testreader&pass=testpass&param=value" &&
-					in.Query != "user=testpublisher&pass=testpass&param=value" &&
-					in.Query != "param=value") {
-				w.WriteHeader(http.StatusBadRequest)
-				return
-			}
-
-			firstReceived = true
-		}),
-	}
-
-	ln, err := net.Listen("tcp", "127.0.0.1:9120")
-	require.NoError(t, err)
-
-	go ts.Server.Serve(ln)
-
-	return ts
-}
-
-func (ts *testHTTPAuthenticator) close() {
-	ts.Server.Shutdown(context.Background())
-}
-
-func httpPullFile(t *testing.T, hc *http.Client, u string) []byte {
-	res, err := hc.Get(u)
-	require.NoError(t, err)
-	defer res.Body.Close()
-
-	if res.StatusCode != http.StatusOK {
-		t.Errorf("bad status code: %v", res.StatusCode)
-	}
-
-	byts, err := io.ReadAll(res.Body)
-	require.NoError(t, err)
-
-	return byts
-}
-
-func TestHLSReadNotFound(t *testing.T) {
-	p, ok := newInstance("")
-	require.Equal(t, true, ok)
-	defer p.Close()
-
-	req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:8888/stream/", nil)
-	require.NoError(t, err)
-
-	hc := &http.Client{Transport: &http.Transport{}}
-
-	res, err := hc.Do(req)
-	require.NoError(t, err)
-	defer res.Body.Close()
-	require.Equal(t, http.StatusNotFound, res.StatusCode)
-}
-
-func TestHLSRead(t *testing.T) {
-	p, ok := newInstance("hlsAlwaysRemux: yes\n" +
-		"paths:\n" +
-		"  all_others:\n")
-	require.Equal(t, true, ok)
-	defer p.Close()
-
-	medi := &description.Media{
-		Type: description.MediaTypeVideo,
-		Formats: []format.Format{&format.H264{
-			PayloadTyp:        96,
-			PacketizationMode: 1,
-			SPS: []byte{ // 1920x1080 baseline
-				0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-				0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-				0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20,
-			},
-			PPS: []byte{0x08, 0x06, 0x07, 0x08},
-		}},
-	}
-
-	v := gortsplib.TransportTCP
-	source := gortsplib.Client{
-		Transport: &v,
-	}
-	err := source.StartRecording("rtsp://localhost:8554/stream",
-		&description.Session{Medias: []*description.Media{medi}})
-	require.NoError(t, err)
-	defer source.Close()
-
-	time.Sleep(500 * time.Millisecond)
-
-	for i := 0; i < 2; i++ {
-		err = source.WritePacketRTP(medi, &rtp.Packet{
-			Header: rtp.Header{
-				Version:        2,
-				Marker:         true,
-				PayloadType:    96,
-				SequenceNumber: 123 + uint16(i),
-				Timestamp:      45343 + uint32(i*90000),
-				SSRC:           563423,
-			},
-			Payload: []byte{
-				0x05, 0x02, 0x03, 0x04, // IDR
-			},
-		})
-		require.NoError(t, err)
-	}
-
-	hc := &http.Client{Transport: &http.Transport{}}
-
-	cnt := httpPullFile(t, hc, "http://localhost:8888/stream/index.m3u8")
-	require.Equal(t, "#EXTM3U\n"+
-		"#EXT-X-VERSION:9\n"+
-		"#EXT-X-INDEPENDENT-SEGMENTS\n"+
-		"\n"+
-		"#EXT-X-STREAM-INF:BANDWIDTH=1192,AVERAGE-BANDWIDTH=1192,"+
-		"CODECS=\"avc1.42c028\",RESOLUTION=1920x1080,FRAME-RATE=30.000\n"+
-		"stream.m3u8\n", string(cnt))
-
-	cnt = httpPullFile(t, hc, "http://localhost:8888/stream/stream.m3u8")
-	require.Regexp(t, "#EXTM3U\n"+
-		"#EXT-X-VERSION:9\n"+
-		"#EXT-X-TARGETDURATION:1\n"+
-		"#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,PART-HOLD-BACK=2\\.50000,CAN-SKIP-UNTIL=6\\.00000\n"+
-		"#EXT-X-PART-INF:PART-TARGET=1\\.00000\n"+
-		"#EXT-X-MEDIA-SEQUENCE:1\n"+
-		"#EXT-X-MAP:URI=\".*?_init.mp4\"\n"+
-		"#EXT-X-GAP\n"+
-		"#EXTINF:1\\.00000,\n"+
-		"gap.mp4\n"+
-		"#EXT-X-GAP\n"+
-		"#EXTINF:1\\.00000,\n"+
-		"gap.mp4\n"+
-		"#EXT-X-GAP\n"+
-		"#EXTINF:1\\.00000,\n"+
-		"gap.mp4\n"+
-		"#EXT-X-GAP\n"+
-		"#EXTINF:1\\.00000,\n"+
-		"gap.mp4\n"+
-		"#EXT-X-GAP\n"+
-		"#EXTINF:1\\.00000,\n"+
-		"gap.mp4\n"+
-		"#EXT-X-GAP\n"+
-		"#EXTINF:1\\.00000,\n"+
-		"gap.mp4\n"+
-		"#EXT-X-PROGRAM-DATE-TIME:.+?\n"+
-		"#EXT-X-PART:DURATION=1\\.00000,URI=\".*?_part0.mp4\",INDEPENDENT=YES\n"+
-		"#EXTINF:1\\.00000,\n"+
-		".*?_seg7.mp4\n"+
-		"#EXT-X-PRELOAD-HINT:TYPE=PART,URI=\".*?_part1.mp4\"\n", string(cnt))
-
-	/*trak := <-c.track
-
-	pkt, _, err := trak.ReadRTP()
-	require.NoError(t, err)
-	require.Equal(t, &rtp.Packet{
-		Header: rtp.Header{
-			Version:        2,
-			Marker:         true,
-			PayloadType:    102,
-			SequenceNumber: pkt.SequenceNumber,
-			Timestamp:      pkt.Timestamp,
-			SSRC:           pkt.SSRC,
-			CSRC:           []uint32{},
-		},
-		Payload: []byte{0x01, 0x02, 0x03, 0x04},
-	}, pkt)*/
-}

internal/core/metrics_test.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"crypto/tls"
+	"io"
 	"net"
 	"net/http"
 	"net/url"
@@ -25,6 +26,21 @@ import (
 	"github.com/bluenviron/mediamtx/internal/test"
 )
 
+func httpPullFile(t *testing.T, hc *http.Client, u string) []byte {
+	res, err := hc.Get(u)
+	require.NoError(t, err)
+	defer res.Body.Close()
+
+	if res.StatusCode != http.StatusOK {
+		t.Errorf("bad status code: %v", res.StatusCode)
+	}
+
+	byts, err := io.ReadAll(res.Body)
+	require.NoError(t, err)
+
+	return byts
+}
+
 func TestMetrics(t *testing.T) {
 	serverCertFpath, err := writeTempFile(serverCert)
 	require.NoError(t, err)
@@ -96,7 +112,7 @@ webrtc_sessions_bytes_sent 0
 			err := source.StartRecording("rtsp://localhost:8554/rtsp_path",
 				&description.Session{Medias: []*description.Media{{
 					Type:    description.MediaTypeVideo,
-					Formats: []format.Format{testFormatH264},
+					Formats: []format.Format{test.FormatH264},
 				}}})
 			require.NoError(t, err)
 			defer source.Close()
@@ -109,7 +125,7 @@ webrtc_sessions_bytes_sent 0
 			err := source2.StartRecording("rtsps://localhost:8322/rtsps_path",
 				&description.Session{Medias: []*description.Media{{
 					Type:    description.MediaTypeVideo,
-					Formats: []format.Format{testFormatH264},
+					Formats: []format.Format{test.FormatH264},
 				}}})
 			require.NoError(t, err)
 			defer source2.Close()
@@ -128,7 +144,7 @@ webrtc_sessions_bytes_sent 0
 			conn, err := rtmp.NewClientConn(nconn, u, true)
 			require.NoError(t, err)
 
-			_, err = rtmp.NewWriter(conn, testFormatH264, nil)
+			_, err = rtmp.NewWriter(conn, test.FormatH264, nil)
 			require.NoError(t, err)
 			<-terminate
 		}()
@@ -145,7 +161,7 @@ webrtc_sessions_bytes_sent 0
 			conn, err := rtmp.NewClientConn(nconn, u, true)
 			require.NoError(t, err)
 
-			_, err = rtmp.NewWriter(conn, testFormatH264, nil)
+			_, err = rtmp.NewWriter(conn, test.FormatH264, nil)
 			require.NoError(t, err)
 			<-terminate
 		}()
@@ -204,18 +220,9 @@ webrtc_sessions_bytes_sent 0
 			require.NoError(t, err)
 
 			err = w.WriteH26x(track, 0, 0, true, [][]byte{
-				{ // SPS
-					0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-					0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-					0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9,
-					0x20,
-				},
-				{ // PPS
-					0x08, 0x06, 0x07, 0x08,
-				},
-				{ // IDR
-					0x05, 1,
-				},
+				test.FormatH264.SPS,
+				test.FormatH264.PPS,
+				{0x05, 1}, // IDR
 			})
 			require.NoError(t, err)
 

internal/core/path.go

@@ -928,12 +928,13 @@ func (pa *path) describe(req defs.PathDescribeReq) defs.PathDescribeRes {
 }
 
 // addPublisher is called by a publisher through pathManager.
-func (pa *path) addPublisher(req defs.PathAddPublisherReq) defs.PathAddPublisherRes {
+func (pa *path) addPublisher(req defs.PathAddPublisherReq) (defs.Path, error) {
 	select {
 	case pa.chAddPublisher <- req:
-		return <-req.Res
+		res := <-req.Res
+		return res.Path, res.Err
 	case <-pa.ctx.Done():
-		return defs.PathAddPublisherRes{Err: fmt.Errorf("terminated")}
+		return nil, fmt.Errorf("terminated")
 	}
 }
 
@@ -948,13 +949,14 @@ func (pa *path) RemovePublisher(req defs.PathRemovePublisherReq) {
 }
 
 // StartPublisher is called by a publisher.
-func (pa *path) StartPublisher(req defs.PathStartPublisherReq) defs.PathStartPublisherRes {
+func (pa *path) StartPublisher(req defs.PathStartPublisherReq) (*stream.Stream, error) {
 	req.Res = make(chan defs.PathStartPublisherRes)
 	select {
 	case pa.chStartPublisher <- req:
-		return <-req.Res
+		res := <-req.Res
+		return res.Stream, res.Err
 	case <-pa.ctx.Done():
-		return defs.PathStartPublisherRes{Err: fmt.Errorf("terminated")}
+		return nil, fmt.Errorf("terminated")
 	}
 }
 
@@ -969,12 +971,13 @@ func (pa *path) StopPublisher(req defs.PathStopPublisherReq) {
 }
 
 // addReader is called by a reader through pathManager.
-func (pa *path) addReader(req defs.PathAddReaderReq) defs.PathAddReaderRes {
+func (pa *path) addReader(req defs.PathAddReaderReq) (defs.Path, *stream.Stream, error) {
 	select {
 	case pa.chAddReader <- req:
-		return <-req.Res
+		res := <-req.Res
+		return res.Path, res.Stream, res.Err
 	case <-pa.ctx.Done():
-		return defs.PathAddReaderRes{Err: fmt.Errorf("terminated")}
+		return nil, nil, fmt.Errorf("terminated")
 	}
 }
 

internal/core/path_manager.go

@@ -10,6 +10,7 @@ import (
 	"github.com/bluenviron/mediamtx/internal/defs"
 	"github.com/bluenviron/mediamtx/internal/externalcmd"
 	"github.com/bluenviron/mediamtx/internal/logger"
+	"github.com/bluenviron/mediamtx/internal/stream"
 )
 
 func pathConfCanBeUpdated(oldPathConf *conf.Path, newPathConf *conf.Path) bool {
@@ -411,14 +412,15 @@ func (pm *pathManager) closePath(pa *path) {
 }
 
 // GetConfForPath is called by a reader or publisher.
-func (pm *pathManager) FindPathConf(req defs.PathFindPathConfReq) defs.PathFindPathConfRes {
+func (pm *pathManager) FindPathConf(req defs.PathFindPathConfReq) (*conf.Path, error) {
 	req.Res = make(chan defs.PathFindPathConfRes)
 	select {
 	case pm.chFindPathConf <- req:
-		return <-req.Res
+		res := <-req.Res
+		return res.Conf, res.Err
 
 	case <-pm.ctx.Done():
-		return defs.PathFindPathConfRes{Err: fmt.Errorf("terminated")}
+		return nil, fmt.Errorf("terminated")
 	}
 }
 
@@ -446,36 +448,36 @@ func (pm *pathManager) Describe(req defs.PathDescribeReq) defs.PathDescribeRes {
 }
 
 // AddPublisher is called by a publisher.
-func (pm *pathManager) AddPublisher(req defs.PathAddPublisherReq) defs.PathAddPublisherRes {
+func (pm *pathManager) AddPublisher(req defs.PathAddPublisherReq) (defs.Path, error) {
 	req.Res = make(chan defs.PathAddPublisherRes)
 	select {
 	case pm.chAddPublisher <- req:
 		res := <-req.Res
 		if res.Err != nil {
-			return res
+			return nil, res.Err
 		}
 
 		return res.Path.(*path).addPublisher(req)
 
 	case <-pm.ctx.Done():
-		return defs.PathAddPublisherRes{Err: fmt.Errorf("terminated")}
+		return nil, fmt.Errorf("terminated")
 	}
 }
 
 // AddReader is called by a reader.
-func (pm *pathManager) AddReader(req defs.PathAddReaderReq) defs.PathAddReaderRes {
+func (pm *pathManager) AddReader(req defs.PathAddReaderReq) (defs.Path, *stream.Stream, error) {
 	req.Res = make(chan defs.PathAddReaderRes)
 	select {
 	case pm.chAddReader <- req:
 		res := <-req.Res
 		if res.Err != nil {
-			return res
+			return nil, nil, res.Err
 		}
 
 		return res.Path.(*path).addReader(req)
 
 	case <-pm.ctx.Done():
-		return defs.PathAddReaderRes{Err: fmt.Errorf("terminated")}
+		return nil, nil, fmt.Errorf("terminated")
 	}
 }
 

internal/core/rtmp_server_test.go

@@ -1,20 +1,87 @@
 package core //nolint:dupl
 
 import (
+	"context"
 	"crypto/tls"
+	"encoding/json"
 	"net"
+	"net/http"
 	"net/url"
 	"os"
 	"testing"
 	"time"
 
 	"github.com/bluenviron/gortsplib/v4/pkg/format"
-	"github.com/bluenviron/mediacommon/pkg/codecs/mpeg4audio"
 	"github.com/stretchr/testify/require"
 
 	"github.com/bluenviron/mediamtx/internal/protocols/rtmp"
+	"github.com/bluenviron/mediamtx/internal/test"
 )
 
+type testHTTPAuthenticator struct {
+	*http.Server
+}
+
+func newTestHTTPAuthenticator(t *testing.T, protocol string, action string) *testHTTPAuthenticator {
+	firstReceived := false
+
+	ts := &testHTTPAuthenticator{}
+
+	ts.Server = &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			require.Equal(t, http.MethodPost, r.Method)
+			require.Equal(t, "/auth", r.URL.Path)
+
+			var in struct {
+				IP       string `json:"ip"`
+				User     string `json:"user"`
+				Password string `json:"password"`
+				Path     string `json:"path"`
+				Protocol string `json:"protocol"`
+				ID       string `json:"id"`
+				Action   string `json:"action"`
+				Query    string `json:"query"`
+			}
+			err := json.NewDecoder(r.Body).Decode(&in)
+			require.NoError(t, err)
+
+			var user string
+			if action == "publish" {
+				user = "testpublisher"
+			} else {
+				user = "testreader"
+			}
+
+			if in.IP != "127.0.0.1" ||
+				in.User != user ||
+				in.Password != "testpass" ||
+				in.Path != "teststream" ||
+				in.Protocol != protocol ||
+				(firstReceived && in.ID == "") ||
+				in.Action != action ||
+				(in.Query != "user=testreader&pass=testpass&param=value" &&
+					in.Query != "user=testpublisher&pass=testpass&param=value" &&
+					in.Query != "param=value") {
+				w.WriteHeader(http.StatusBadRequest)
+				return
+			}
+
+			firstReceived = true
+		}),
+	}
+
+	ln, err := net.Listen("tcp", "127.0.0.1:9120")
+	require.NoError(t, err)
+
+	go ts.Server.Serve(ln)
+
+	return ts
+}
+
+func (ts *testHTTPAuthenticator) close() {
+	ts.Server.Shutdown(context.Background())
+}
+
 func TestRTMPServer(t *testing.T) {
 	for _, encrypt := range []string{
 		"plain",
@@ -98,30 +165,7 @@ func TestRTMPServer(t *testing.T) {
 				conn1, err := rtmp.NewClientConn(nconn1, u1, true)
 				require.NoError(t, err)
 
-				videoTrack := &format.H264{
-					PayloadTyp: 96,
-					SPS: []byte{ // 1920x1080 baseline
-						0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-						0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-						0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20,
-					},
-					PPS:               []byte{0x08, 0x06, 0x07, 0x08},
-					PacketizationMode: 1,
-				}
-
-				audioTrack := &format.MPEG4Audio{
-					PayloadTyp: 96,
-					Config: &mpeg4audio.Config{
-						Type:         2,
-						SampleRate:   44100,
-						ChannelCount: 2,
-					},
-					SizeLength:       13,
-					IndexLength:      3,
-					IndexDeltaLength: 3,
-				}
-
-				w, err := rtmp.NewWriter(conn1, videoTrack, audioTrack)
+				w, err := rtmp.NewWriter(conn1, test.FormatH264, test.FormatMPEG4Audio)
 				require.NoError(t, err)
 
 				time.Sleep(500 * time.Millisecond)
@@ -151,8 +195,8 @@ func TestRTMPServer(t *testing.T) {
 				require.NoError(t, err)
 
 				videoTrack1, audioTrack2 := r.Tracks()
-				require.Equal(t, videoTrack, videoTrack1)
-				require.Equal(t, audioTrack, audioTrack2)
+				require.Equal(t, test.FormatH264, videoTrack1)
+				require.Equal(t, test.FormatMPEG4Audio, audioTrack2)
 
 				err = w.WriteH264(0, 0, true, [][]byte{
 					{0x05, 0x02, 0x03, 0x04}, // IDR 1
@@ -162,15 +206,8 @@ func TestRTMPServer(t *testing.T) {
 
 				r.OnDataH264(func(pts time.Duration, au [][]byte) {
 					require.Equal(t, [][]byte{
-						{ // SPS
-							0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-							0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-							0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9,
-							0x20,
-						},
-						{ // PPS
-							0x08, 0x06, 0x07, 0x08,
-						},
+						test.FormatH264.SPS,
+						test.FormatH264.PPS,
 						{ // IDR 1
 							0x05, 0x02, 0x03, 0x04,
 						},

internal/core/srt_server_test.go

@@ -6,7 +6,8 @@ import (
 	"time"
 
 	"github.com/bluenviron/mediacommon/pkg/formats/mpegts"
-	"github.com/datarhei/gosrt"
+	"github.com/bluenviron/mediamtx/internal/test"
+	srt "github.com/datarhei/gosrt"
 	"github.com/stretchr/testify/require"
 )
 
@@ -57,18 +58,9 @@ func TestSRTServer(t *testing.T) {
 			require.NoError(t, err)
 
 			err = w.WriteH26x(track, 0, 0, true, [][]byte{
-				{ // SPS
-					0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-					0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-					0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9,
-					0x20,
-				},
-				{ // PPS
-					0x08, 0x06, 0x07, 0x08,
-				},
-				{ // IDR
-					0x05, 1,
-				},
+				test.FormatH264.SPS,
+				test.FormatH264.PPS,
+				{0x05, 1}, // IDR
 			})
 			require.NoError(t, err)
 
@@ -117,19 +109,10 @@ func TestSRTServer(t *testing.T) {
 				require.Equal(t, int64(0), pts)
 				require.Equal(t, int64(0), dts)
 				require.Equal(t, [][]byte{
-					{ // SPS
-						0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-						0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-						0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9,
-						0x20,
-					},
-					{ // PPS
-						0x08, 0x06, 0x07, 0x08,
-					},
-					{ // IDR
-						0x05, 1,
-					},
-				}, au)
+					test.FormatH264.SPS,
+					test.FormatH264.PPS,
+					{0x05, 1},
+				}, au) // IDR
 				received = true
 				return nil
 			})

internal/defs/path.go

@@ -28,7 +28,7 @@ type Path interface {
 	Name() string
 	SafeConf() *conf.Path
 	ExternalCmdEnv() externalcmd.Environment
-	StartPublisher(req PathStartPublisherReq) PathStartPublisherRes
+	StartPublisher(req PathStartPublisherReq) (*stream.Stream, error)
 	StopPublisher(req PathStopPublisherReq)
 	RemovePublisher(req PathRemovePublisherReq)
 	RemoveReader(req PathRemoveReaderReq)

internal/defs/path_manager.go

@@ -1,9 +1,14 @@
 package defs
 
+import (
+	"github.com/bluenviron/mediamtx/internal/conf"
+	"github.com/bluenviron/mediamtx/internal/stream"
+)
+
 // PathManager is a path manager.
 type PathManager interface {
-	FindPathConf(req PathFindPathConfReq) PathFindPathConfRes
+	FindPathConf(req PathFindPathConfReq) (*conf.Path, error)
 	Describe(req PathDescribeReq) PathDescribeRes
-	AddPublisher(req PathAddPublisherReq) PathAddPublisherRes
-	AddReader(req PathAddReaderReq) PathAddReaderRes
+	AddPublisher(req PathAddPublisherReq) (Path, error)
+	AddReader(req PathAddReaderReq) (Path, *stream.Stream, error)
 }

internal/playback/fmp4_test.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/bluenviron/mediacommon/pkg/codecs/mpeg4audio"
 	"github.com/bluenviron/mediacommon/pkg/formats/fmp4"
+	"github.com/bluenviron/mediamtx/internal/test"
 )
 
 func writeBenchInit(f io.WriteSeeker) {
@@ -16,13 +17,8 @@ func writeBenchInit(f io.WriteSeeker) {
 				ID:        1,
 				TimeScale: 90000,
 				Codec: &fmp4.CodecH264{
-					SPS: []byte{
-						0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-						0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-						0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9,
-						0x20,
-					},
-					PPS: []byte{0x08},
+					SPS: test.FormatH264.SPS,
+					PPS: test.FormatH264.PPS,
 				},
 			},
 			{

internal/playback/server_test.go

@@ -23,13 +23,8 @@ func writeSegment1(t *testing.T, fpath string) {
 			ID:        1,
 			TimeScale: 90000,
 			Codec: &fmp4.CodecH264{
-				SPS: []byte{
-					0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-					0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-					0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9,
-					0x20,
-				},
-				PPS: []byte{0x08},
+				SPS: test.FormatH264.SPS,
+				PPS: test.FormatH264.PPS,
 			},
 		}},
 	}
@@ -86,13 +81,8 @@ func writeSegment2(t *testing.T, fpath string) {
 			ID:        1,
 			TimeScale: 90000,
 			Codec: &fmp4.CodecH264{
-				SPS: []byte{
-					0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-					0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-					0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9,
-					0x20,
-				},
-				PPS: []byte{0x08},
+				SPS: test.FormatH264.SPS,
+				PPS: test.FormatH264.PPS,
 			},
 		}},
 	}

internal/record/agent_test.go

@@ -103,14 +103,8 @@ func TestAgent(t *testing.T) {
 					PTS: (50 + time.Duration(i)) * time.Second,
 				},
 				AU: [][]byte{
-					{ // SPS
-						0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-						0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-						0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20,
-					},
-					{ // PPS
-						0x08, 0x06, 0x07, 0x08,
-					},
+					test.FormatH264.SPS,
+					test.FormatH264.PPS,
 					{5}, // IDR
 				},
 			})
@@ -290,14 +284,8 @@ func TestAgentFMP4NegativeDTS(t *testing.T) {
 				NTP: time.Date(2008, 0o5, 20, 22, 15, 25, 0, time.UTC),
 			},
 			AU: [][]byte{
-				{ // SPS
-					0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-					0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-					0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20,
-				},
-				{ // PPS
-					0x08, 0x06, 0x07, 0x08,
-				},
+				test.FormatH264.SPS,
+				test.FormatH264.PPS,
 				{5}, // IDR
 			},
 		})

internal/record/format_fmp4.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/bluenviron/mediamtx/internal/defs"
 	"github.com/bluenviron/mediamtx/internal/logger"
+	"github.com/bluenviron/mediamtx/internal/test"
 	"github.com/bluenviron/mediamtx/internal/unit"
 )
 
@@ -374,16 +375,8 @@ func (f *formatFMP4) initialize() {
 				sps, pps := forma.SafeParams()
 
 				if sps == nil || pps == nil {
-					sps = []byte{
-						0x67, 0x42, 0xc0, 0x1f, 0xd9, 0x00, 0xf0, 0x11,
-						0x7e, 0xf0, 0x11, 0x00, 0x00, 0x03, 0x00, 0x01,
-						0x00, 0x00, 0x03, 0x00, 0x30, 0x8f, 0x18, 0x32,
-						0x48,
-					}
-
-					pps = []byte{
-						0x68, 0xcb, 0x8c, 0xb2,
-					}
+					sps = test.FormatH264.SPS
+					pps = test.FormatH264.PPS
 				}
 
 				codec := &fmp4.CodecH264{

internal/servers/hls/http_server.go

@@ -40,7 +40,7 @@ type httpServer struct {
 	allowOrigin    string
 	trustedProxies conf.IPsOrCIDRs
 	readTimeout    conf.StringDuration
-	pathManager    defs.PathManager
+	pathManager    serverPathManager
 	parent         *Server
 
 	inner *httpp.WrappedServer
@@ -150,7 +150,7 @@ func (s *httpServer) onRequest(ctx *gin.Context) {
 
 	user, pass, hasCredentials := ctx.Request.BasicAuth()
 
-	res := s.pathManager.FindPathConf(defs.PathFindPathConfReq{
+	pathConf, err := s.pathManager.FindPathConf(defs.PathFindPathConfReq{
 		AccessRequest: defs.PathAccessRequest{
 			Name:    dir,
 			Query:   ctx.Request.URL.RawQuery,
@@ -161,9 +161,9 @@ func (s *httpServer) onRequest(ctx *gin.Context) {
 			Proto:   defs.AuthProtocolHLS,
 		},
 	})
-	if res.Err != nil {
+	if err != nil {
 		var terr defs.AuthenticationError
-		if errors.As(res.Err, &terr) {
+		if errors.As(err, &terr) {
 			if !hasCredentials {
 				ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`)
 				ctx.Writer.WriteHeader(http.StatusUnauthorized)
@@ -192,23 +192,22 @@ func (s *httpServer) onRequest(ctx *gin.Context) {
 
 	default:
 		mux, err := s.parent.getMuxer(serverGetMuxerReq{
-			path:       dir,
-			remoteAddr: httpp.RemoteAddr(ctx),
+			path:           dir,
+			remoteAddr:     httpp.RemoteAddr(ctx),
+			sourceOnDemand: pathConf.SourceOnDemand,
 		})
 		if err != nil {
-			s.Log(logger.Error, err.Error())
-			ctx.Writer.WriteHeader(http.StatusInternalServerError)
+			ctx.Writer.WriteHeader(http.StatusNotFound)
 			return
 		}
 
-		err = mux.processRequest(muxerProcessRequestReq{})
-		if err != nil {
-			s.Log(logger.Error, err.Error())
-			ctx.Writer.WriteHeader(http.StatusInternalServerError)
+		mi := mux.getInstance()
+		if mi == nil {
+			ctx.Writer.WriteHeader(http.StatusNotFound)
 			return
 		}
 
 		ctx.Request.URL.Path = fname
-		mux.handleRequest(ctx)
+		mi.handleRequest(ctx)
 	}
 }

internal/servers/hls/muxer.go

@@ -5,29 +5,19 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
-	"os"
-	"path/filepath"
 	"sync"
 	"sync/atomic"
 	"time"
 
-	"github.com/bluenviron/gohlslib"
-	"github.com/bluenviron/gohlslib/pkg/codecs"
-	"github.com/bluenviron/gortsplib/v4/pkg/format"
-	"github.com/gin-gonic/gin"
-
-	"github.com/bluenviron/mediamtx/internal/asyncwriter"
 	"github.com/bluenviron/mediamtx/internal/conf"
 	"github.com/bluenviron/mediamtx/internal/defs"
 	"github.com/bluenviron/mediamtx/internal/logger"
-	"github.com/bluenviron/mediamtx/internal/stream"
-	"github.com/bluenviron/mediamtx/internal/unit"
 )
 
 const (
 	closeCheckPeriod     = 1 * time.Second
 	closeAfterInactivity = 60 * time.Second
-	muxerRecreatePause   = 10 * time.Second
+	recreatePause        = 10 * time.Second
 )
 
 func int64Ptr(v int64) *int64 {
@@ -51,8 +41,8 @@ func (w *responseWriterWithCounter) Write(p []byte) (int, error) {
 	return n, err
 }
 
-type muxerProcessRequestReq struct {
-	res chan error
+type muxerGetInstanceReq struct {
+	res chan *muxerInstance
 }
 
 type muxer struct {
@@ -68,21 +58,18 @@ type muxer struct {
 	writeQueueSize            int
 	wg                        *sync.WaitGroup
 	pathName                  string
-	pathManager               defs.PathManager
+	pathManager               serverPathManager
 	parent                    *Server
 
 	ctx             context.Context
 	ctxCancel       func()
 	created         time.Time
 	path            defs.Path
-	writer          *asyncwriter.Writer
 	lastRequestTime *int64
-	muxer           *gohlslib.Muxer
-	requests        []*muxerProcessRequestReq
 	bytesSent       *uint64
 
 	// in
-	chProcessRequest chan muxerProcessRequestReq
+	chGetInstance chan muxerGetInstanceReq
 }
 
 func (m *muxer) initialize() {
@@ -93,7 +80,7 @@ func (m *muxer) initialize() {
 	m.created = time.Now()
 	m.lastRequestTime = int64Ptr(time.Now().UnixNano())
 	m.bytesSent = new(uint64)
-	m.chProcessRequest = make(chan muxerProcessRequestReq)
+	m.chGetInstance = make(chan muxerGetInstanceReq)
 
 	m.Log(logger.Info, "created %s", func() string {
 		if m.remoteAddr == "" {
@@ -123,368 +110,139 @@ func (m *muxer) PathName() string {
 func (m *muxer) run() {
 	defer m.wg.Done()
 
-	err := func() error {
-		var innerReady chan struct{}
-		var innerErr chan error
-		var innerCtx context.Context
-		var innerCtxCancel func()
-
-		createInner := func() {
-			innerReady = make(chan struct{})
-			innerErr = make(chan error)
-			innerCtx, innerCtxCancel = context.WithCancel(context.Background())
-			go func() {
-				innerErr <- m.runInner(innerCtx, innerReady)
-			}()
-		}
-
-		createInner()
-
-		isReady := false
-		isRecreating := false
-		recreateTimer := emptyTimer()
-
-		for {
-			select {
-			case <-m.ctx.Done():
-				if !isRecreating {
-					innerCtxCancel()
-					<-innerErr
-				}
-				return errors.New("terminated")
-
-			case req := <-m.chProcessRequest:
-				switch {
-				case isRecreating:
-					req.res <- fmt.Errorf("recreating")
-
-				case isReady:
-					req.res <- nil
-
-				default:
-					m.requests = append(m.requests, &req)
-				}
-
-			case <-innerReady:
-				isReady = true
-				for _, req := range m.requests {
-					req.res <- nil
-				}
-				m.requests = nil
-
-			case err := <-innerErr:
-				innerCtxCancel()
-
-				if m.remoteAddr == "" { // created with "always remux"
-					m.Log(logger.Error, err.Error())
-					m.clearQueuedRequests()
-					isReady = false
-					isRecreating = true
-					recreateTimer = time.NewTimer(muxerRecreatePause)
-				} else {
-					return err
-				}
-
-			case <-recreateTimer.C:
-				isRecreating = false
-				createInner()
-			}
-		}
-	}()
+	err := m.runInner()
 
 	m.ctxCancel()
 
-	m.clearQueuedRequests()
-
 	m.parent.closeMuxer(m)
 
 	m.Log(logger.Info, "destroyed: %v", err)
 }
 
-func (m *muxer) clearQueuedRequests() {
-	for _, req := range m.requests {
-		req.res <- fmt.Errorf("terminated")
-	}
-	m.requests = nil
-}
-
-func (m *muxer) runInner(innerCtx context.Context, innerReady chan struct{}) error {
-	res := m.pathManager.AddReader(defs.PathAddReaderReq{
+func (m *muxer) runInner() error {
+	path, stream, err := m.pathManager.AddReader(defs.PathAddReaderReq{
 		Author: m,
 		AccessRequest: defs.PathAccessRequest{
 			Name:     m.pathName,
 			SkipAuth: true,
 		},
 	})
-	if res.Err != nil {
-		return res.Err
+	if err != nil {
+		return err
 	}
 
-	m.path = res.Path
+	m.path = path
 
 	defer m.path.RemoveReader(defs.PathRemoveReaderReq{Author: m})
 
-	m.writer = asyncwriter.New(m.writeQueueSize, m)
+	var instanceError chan error
+	var recreateTimer *time.Timer
 
-	defer res.Stream.RemoveReader(m.writer)
-
-	videoTrack := m.createVideoTrack(res.Stream)
-	audioTrack := m.createAudioTrack(res.Stream)
-
-	if videoTrack == nil && audioTrack == nil {
-		return fmt.Errorf(
-			"the stream doesn't contain any supported codec, which are currently H265, H264, Opus, MPEG-4 Audio")
+	mi := &muxerInstance{
+		variant:         m.variant,
+		segmentCount:    m.segmentCount,
+		segmentDuration: m.segmentDuration,
+		partDuration:    m.partDuration,
+		segmentMaxSize:  m.segmentMaxSize,
+		directory:       m.directory,
+		writeQueueSize:  m.writeQueueSize,
+		pathName:        m.pathName,
+		stream:          stream,
+		bytesSent:       m.bytesSent,
+		parent:          m,
 	}
-
-	var muxerDirectory string
-	if m.directory != "" {
-		muxerDirectory = filepath.Join(m.directory, m.pathName)
-		os.MkdirAll(muxerDirectory, 0o755)
-		defer os.Remove(muxerDirectory)
-	}
-
-	m.muxer = &gohlslib.Muxer{
-		Variant:         gohlslib.MuxerVariant(m.variant),
-		SegmentCount:    m.segmentCount,
-		SegmentDuration: time.Duration(m.segmentDuration),
-		PartDuration:    time.Duration(m.partDuration),
-		SegmentMaxSize:  uint64(m.segmentMaxSize),
-		VideoTrack:      videoTrack,
-		AudioTrack:      audioTrack,
-		Directory:       muxerDirectory,
-	}
-
-	err := m.muxer.Start()
+	err = mi.initialize()
 	if err != nil {
-		return fmt.Errorf("muxer error: %w", err)
+		if m.remoteAddr != "" {
+			return err
+		}
+
+		m.Log(logger.Error, err.Error())
+		mi = nil
+		instanceError = make(chan error)
+		recreateTimer = time.NewTimer(recreatePause)
+	} else {
+		instanceError = mi.errorChan()
+		recreateTimer = emptyTimer()
 	}
-	defer m.muxer.Close()
 
-	innerReady <- struct{}{}
-
-	m.Log(logger.Info, "is converting into HLS, %s",
-		defs.FormatsInfo(res.Stream.FormatsForReader(m.writer)))
-
-	m.writer.Start()
-
-	closeCheckTicker := time.NewTicker(closeCheckPeriod)
-	defer closeCheckTicker.Stop()
+	var activityCheckTimer *time.Timer
+	if m.remoteAddr != "" {
+		activityCheckTimer = time.NewTimer(closeCheckPeriod)
+	} else {
+		activityCheckTimer = emptyTimer()
+	}
 
 	for {
 		select {
-		case <-closeCheckTicker.C:
+		case req := <-m.chGetInstance:
+			req.res <- mi
+
+		case err := <-instanceError:
+			mi.close()
+
 			if m.remoteAddr != "" {
-				t := time.Unix(0, atomic.LoadInt64(m.lastRequestTime))
-				if time.Since(t) >= closeAfterInactivity {
-					m.writer.Stop()
-					return fmt.Errorf("not used anymore")
+				return err
+			}
+
+			m.Log(logger.Error, err.Error())
+			mi = nil
+			instanceError = make(chan error)
+			recreateTimer = time.NewTimer(recreatePause)
+
+		case <-recreateTimer.C:
+			mi = &muxerInstance{
+				variant:         m.variant,
+				segmentCount:    m.segmentCount,
+				segmentDuration: m.segmentDuration,
+				partDuration:    m.partDuration,
+				segmentMaxSize:  m.segmentMaxSize,
+				directory:       m.directory,
+				writeQueueSize:  m.writeQueueSize,
+				pathName:        m.pathName,
+				stream:          stream,
+				bytesSent:       m.bytesSent,
+				parent:          m,
+			}
+			err := mi.initialize()
+			if err != nil {
+				m.Log(logger.Error, err.Error())
+				mi = nil
+				recreateTimer = time.NewTimer(recreatePause)
+			} else {
+				instanceError = mi.errorChan()
+			}
+
+		case <-activityCheckTimer.C:
+			t := time.Unix(0, atomic.LoadInt64(m.lastRequestTime))
+			if time.Since(t) >= closeAfterInactivity {
+				if mi != nil {
+					mi.close()
 				}
+				return fmt.Errorf("not used anymore")
 			}
+			activityCheckTimer = time.NewTimer(closeCheckPeriod)
 
-		case err := <-m.writer.Error():
-			return err
-
-		case <-innerCtx.Done():
-			m.writer.Stop()
-			return fmt.Errorf("terminated")
+		case <-m.ctx.Done():
+			if mi != nil {
+				mi.close()
+			}
+			return errors.New("terminated")
 		}
 	}
 }
 
-func (m *muxer) createVideoTrack(stream *stream.Stream) *gohlslib.Track {
-	var videoFormatAV1 *format.AV1
-	videoMedia := stream.Desc().FindFormat(&videoFormatAV1)
-
-	if videoFormatAV1 != nil {
-		stream.AddReader(m.writer, videoMedia, videoFormatAV1, func(u unit.Unit) error {
-			tunit := u.(*unit.AV1)
-
-			if tunit.TU == nil {
-				return nil
-			}
-
-			err := m.muxer.WriteAV1(tunit.NTP, tunit.PTS, tunit.TU)
-			if err != nil {
-				return fmt.Errorf("muxer error: %w", err)
-			}
-
-			return nil
-		})
-
-		return &gohlslib.Track{
-			Codec: &codecs.AV1{},
-		}
-	}
-
-	var videoFormatVP9 *format.VP9
-	videoMedia = stream.Desc().FindFormat(&videoFormatVP9)
-
-	if videoFormatVP9 != nil {
-		stream.AddReader(m.writer, videoMedia, videoFormatVP9, func(u unit.Unit) error {
-			tunit := u.(*unit.VP9)
-
-			if tunit.Frame == nil {
-				return nil
-			}
-
-			err := m.muxer.WriteVP9(tunit.NTP, tunit.PTS, tunit.Frame)
-			if err != nil {
-				return fmt.Errorf("muxer error: %w", err)
-			}
-
-			return nil
-		})
-
-		return &gohlslib.Track{
-			Codec: &codecs.VP9{},
-		}
-	}
-
-	var videoFormatH265 *format.H265
-	videoMedia = stream.Desc().FindFormat(&videoFormatH265)
-
-	if videoFormatH265 != nil {
-		stream.AddReader(m.writer, videoMedia, videoFormatH265, func(u unit.Unit) error {
-			tunit := u.(*unit.H265)
-
-			if tunit.AU == nil {
-				return nil
-			}
-
-			err := m.muxer.WriteH26x(tunit.NTP, tunit.PTS, tunit.AU)
-			if err != nil {
-				return fmt.Errorf("muxer error: %w", err)
-			}
-
-			return nil
-		})
-
-		vps, sps, pps := videoFormatH265.SafeParams()
-
-		return &gohlslib.Track{
-			Codec: &codecs.H265{
-				VPS: vps,
-				SPS: sps,
-				PPS: pps,
-			},
-		}
-	}
-
-	var videoFormatH264 *format.H264
-	videoMedia = stream.Desc().FindFormat(&videoFormatH264)
-
-	if videoFormatH264 != nil {
-		stream.AddReader(m.writer, videoMedia, videoFormatH264, func(u unit.Unit) error {
-			tunit := u.(*unit.H264)
-
-			if tunit.AU == nil {
-				return nil
-			}
-
-			err := m.muxer.WriteH26x(tunit.NTP, tunit.PTS, tunit.AU)
-			if err != nil {
-				return fmt.Errorf("muxer error: %w", err)
-			}
-
-			return nil
-		})
-
-		sps, pps := videoFormatH264.SafeParams()
-
-		return &gohlslib.Track{
-			Codec: &codecs.H264{
-				SPS: sps,
-				PPS: pps,
-			},
-		}
-	}
-
-	return nil
-}
-
-func (m *muxer) createAudioTrack(stream *stream.Stream) *gohlslib.Track {
-	var audioFormatOpus *format.Opus
-	audioMedia := stream.Desc().FindFormat(&audioFormatOpus)
-
-	if audioMedia != nil {
-		stream.AddReader(m.writer, audioMedia, audioFormatOpus, func(u unit.Unit) error {
-			tunit := u.(*unit.Opus)
-
-			err := m.muxer.WriteOpus(
-				tunit.NTP,
-				tunit.PTS,
-				tunit.Packets)
-			if err != nil {
-				return fmt.Errorf("muxer error: %w", err)
-			}
-
-			return nil
-		})
-
-		return &gohlslib.Track{
-			Codec: &codecs.Opus{
-				ChannelCount: func() int {
-					if audioFormatOpus.IsStereo {
-						return 2
-					}
-					return 1
-				}(),
-			},
-		}
-	}
-
-	var audioFormatMPEG4Audio *format.MPEG4Audio
-	audioMedia = stream.Desc().FindFormat(&audioFormatMPEG4Audio)
-
-	if audioMedia != nil {
-		stream.AddReader(m.writer, audioMedia, audioFormatMPEG4Audio, func(u unit.Unit) error {
-			tunit := u.(*unit.MPEG4Audio)
-
-			if tunit.AUs == nil {
-				return nil
-			}
-
-			err := m.muxer.WriteMPEG4Audio(
-				tunit.NTP,
-				tunit.PTS,
-				tunit.AUs)
-			if err != nil {
-				return fmt.Errorf("muxer error: %w", err)
-			}
-
-			return nil
-		})
-
-		return &gohlslib.Track{
-			Codec: &codecs.MPEG4Audio{
-				Config: *audioFormatMPEG4Audio.GetConfig(),
-			},
-		}
-	}
-
-	return nil
-}
-
-func (m *muxer) handleRequest(ctx *gin.Context) {
+func (m *muxer) getInstance() *muxerInstance {
 	atomic.StoreInt64(m.lastRequestTime, time.Now().UnixNano())
 
-	w := &responseWriterWithCounter{
-		ResponseWriter: ctx.Writer,
-		bytesSent:      m.bytesSent,
-	}
-
-	m.muxer.Handle(w, ctx.Request)
-}
-
-func (m *muxer) processRequest(req muxerProcessRequestReq) error {
-	req.res = make(chan error)
+	req := muxerGetInstanceReq{res: make(chan *muxerInstance)}
 
 	select {
-	case m.chProcessRequest <- req:
+	case m.chGetInstance <- req:
 		return <-req.res
 
 	case <-m.ctx.Done():
-		return fmt.Errorf("terminated")
+		return nil
 	}
 }
 

internal/servers/hls/muxer_instance.go

@@ -0,0 +1,278 @@
+package hls
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/bluenviron/gohlslib"
+	"github.com/bluenviron/gohlslib/pkg/codecs"
+	"github.com/bluenviron/gortsplib/v4/pkg/format"
+	"github.com/bluenviron/mediamtx/internal/asyncwriter"
+	"github.com/bluenviron/mediamtx/internal/conf"
+	"github.com/bluenviron/mediamtx/internal/defs"
+	"github.com/bluenviron/mediamtx/internal/logger"
+	"github.com/bluenviron/mediamtx/internal/stream"
+	"github.com/bluenviron/mediamtx/internal/unit"
+	"github.com/gin-gonic/gin"
+)
+
+type muxerInstance struct {
+	variant         conf.HLSVariant
+	segmentCount    int
+	segmentDuration conf.StringDuration
+	partDuration    conf.StringDuration
+	segmentMaxSize  conf.StringSize
+	directory       string
+	writeQueueSize  int
+	pathName        string
+	stream          *stream.Stream
+	bytesSent       *uint64
+	parent          logger.Writer
+
+	writer *asyncwriter.Writer
+	hmuxer *gohlslib.Muxer
+}
+
+func (mi *muxerInstance) initialize() error {
+	mi.writer = asyncwriter.New(mi.writeQueueSize, mi)
+
+	videoTrack := mi.createVideoTrack()
+	audioTrack := mi.createAudioTrack()
+
+	if videoTrack == nil && audioTrack == nil {
+		mi.stream.RemoveReader(mi.writer)
+		return fmt.Errorf(
+			"the stream doesn't contain any supported codec, which are currently H265, H264, Opus, MPEG-4 Audio")
+	}
+
+	var muxerDirectory string
+	if mi.directory != "" {
+		muxerDirectory = filepath.Join(mi.directory, mi.pathName)
+		os.MkdirAll(muxerDirectory, 0o755)
+		defer os.Remove(muxerDirectory)
+	}
+
+	mi.hmuxer = &gohlslib.Muxer{
+		Variant:         gohlslib.MuxerVariant(mi.variant),
+		SegmentCount:    mi.segmentCount,
+		SegmentDuration: time.Duration(mi.segmentDuration),
+		PartDuration:    time.Duration(mi.partDuration),
+		SegmentMaxSize:  uint64(mi.segmentMaxSize),
+		VideoTrack:      videoTrack,
+		AudioTrack:      audioTrack,
+		Directory:       muxerDirectory,
+	}
+
+	err := mi.hmuxer.Start()
+	if err != nil {
+		mi.stream.RemoveReader(mi.writer)
+		return err
+	}
+
+	mi.Log(logger.Info, "is converting into HLS, %s",
+		defs.FormatsInfo(mi.stream.FormatsForReader(mi.writer)))
+
+	mi.writer.Start()
+
+	return nil
+}
+
+// Log implements logger.Writer.
+func (mi *muxerInstance) Log(level logger.Level, format string, args ...interface{}) {
+	mi.parent.Log(level, format, args...)
+}
+
+func (mi *muxerInstance) close() {
+	mi.writer.Stop()
+	mi.hmuxer.Close()
+	mi.stream.RemoveReader(mi.writer)
+}
+
+func (mi *muxerInstance) createVideoTrack() *gohlslib.Track {
+	var videoFormatAV1 *format.AV1
+	videoMedia := mi.stream.Desc().FindFormat(&videoFormatAV1)
+
+	if videoFormatAV1 != nil {
+		mi.stream.AddReader(mi.writer, videoMedia, videoFormatAV1, func(u unit.Unit) error {
+			tunit := u.(*unit.AV1)
+
+			if tunit.TU == nil {
+				return nil
+			}
+
+			err := mi.hmuxer.WriteAV1(tunit.NTP, tunit.PTS, tunit.TU)
+			if err != nil {
+				return fmt.Errorf("muxer error: %w", err)
+			}
+
+			return nil
+		})
+
+		return &gohlslib.Track{
+			Codec: &codecs.AV1{},
+		}
+	}
+
+	var videoFormatVP9 *format.VP9
+	videoMedia = mi.stream.Desc().FindFormat(&videoFormatVP9)
+
+	if videoFormatVP9 != nil {
+		mi.stream.AddReader(mi.writer, videoMedia, videoFormatVP9, func(u unit.Unit) error {
+			tunit := u.(*unit.VP9)
+
+			if tunit.Frame == nil {
+				return nil
+			}
+
+			err := mi.hmuxer.WriteVP9(tunit.NTP, tunit.PTS, tunit.Frame)
+			if err != nil {
+				return fmt.Errorf("muxer error: %w", err)
+			}
+
+			return nil
+		})
+
+		return &gohlslib.Track{
+			Codec: &codecs.VP9{},
+		}
+	}
+
+	var videoFormatH265 *format.H265
+	videoMedia = mi.stream.Desc().FindFormat(&videoFormatH265)
+
+	if videoFormatH265 != nil {
+		mi.stream.AddReader(mi.writer, videoMedia, videoFormatH265, func(u unit.Unit) error {
+			tunit := u.(*unit.H265)
+
+			if tunit.AU == nil {
+				return nil
+			}
+
+			err := mi.hmuxer.WriteH26x(tunit.NTP, tunit.PTS, tunit.AU)
+			if err != nil {
+				return fmt.Errorf("muxer error: %w", err)
+			}
+
+			return nil
+		})
+
+		vps, sps, pps := videoFormatH265.SafeParams()
+
+		return &gohlslib.Track{
+			Codec: &codecs.H265{
+				VPS: vps,
+				SPS: sps,
+				PPS: pps,
+			},
+		}
+	}
+
+	var videoFormatH264 *format.H264
+	videoMedia = mi.stream.Desc().FindFormat(&videoFormatH264)
+
+	if videoFormatH264 != nil {
+		mi.stream.AddReader(mi.writer, videoMedia, videoFormatH264, func(u unit.Unit) error {
+			tunit := u.(*unit.H264)
+
+			if tunit.AU == nil {
+				return nil
+			}
+
+			err := mi.hmuxer.WriteH26x(tunit.NTP, tunit.PTS, tunit.AU)
+			if err != nil {
+				return fmt.Errorf("muxer error: %w", err)
+			}
+
+			return nil
+		})
+
+		sps, pps := videoFormatH264.SafeParams()
+
+		return &gohlslib.Track{
+			Codec: &codecs.H264{
+				SPS: sps,
+				PPS: pps,
+			},
+		}
+	}
+
+	return nil
+}
+
+func (mi *muxerInstance) createAudioTrack() *gohlslib.Track {
+	var audioFormatOpus *format.Opus
+	audioMedia := mi.stream.Desc().FindFormat(&audioFormatOpus)
+
+	if audioMedia != nil {
+		mi.stream.AddReader(mi.writer, audioMedia, audioFormatOpus, func(u unit.Unit) error {
+			tunit := u.(*unit.Opus)
+
+			err := mi.hmuxer.WriteOpus(
+				tunit.NTP,
+				tunit.PTS,
+				tunit.Packets)
+			if err != nil {
+				return fmt.Errorf("muxer error: %w", err)
+			}
+
+			return nil
+		})
+
+		return &gohlslib.Track{
+			Codec: &codecs.Opus{
+				ChannelCount: func() int {
+					if audioFormatOpus.IsStereo {
+						return 2
+					}
+					return 1
+				}(),
+			},
+		}
+	}
+
+	var audioFormatMPEG4Audio *format.MPEG4Audio
+	audioMedia = mi.stream.Desc().FindFormat(&audioFormatMPEG4Audio)
+
+	if audioMedia != nil {
+		mi.stream.AddReader(mi.writer, audioMedia, audioFormatMPEG4Audio, func(u unit.Unit) error {
+			tunit := u.(*unit.MPEG4Audio)
+
+			if tunit.AUs == nil {
+				return nil
+			}
+
+			err := mi.hmuxer.WriteMPEG4Audio(
+				tunit.NTP,
+				tunit.PTS,
+				tunit.AUs)
+			if err != nil {
+				return fmt.Errorf("muxer error: %w", err)
+			}
+
+			return nil
+		})
+
+		return &gohlslib.Track{
+			Codec: &codecs.MPEG4Audio{
+				Config: *audioFormatMPEG4Audio.GetConfig(),
+			},
+		}
+	}
+
+	return nil
+}
+
+func (mi *muxerInstance) errorChan() chan error {
+	return mi.writer.Error()
+}
+
+func (mi *muxerInstance) handleRequest(ctx *gin.Context) {
+	w := &responseWriterWithCounter{
+		ResponseWriter: ctx.Writer,
+		bytesSent:      mi.bytesSent,
+	}
+
+	mi.hmuxer.Handle(w, ctx.Request)
+}

internal/servers/hls/server.go

@@ -11,6 +11,7 @@ import (
 	"github.com/bluenviron/mediamtx/internal/conf"
 	"github.com/bluenviron/mediamtx/internal/defs"
 	"github.com/bluenviron/mediamtx/internal/logger"
+	"github.com/bluenviron/mediamtx/internal/stream"
 )
 
 // ErrMuxerNotFound is returned when a muxer is not found.
@@ -22,9 +23,10 @@ type serverGetMuxerRes struct {
 }
 
 type serverGetMuxerReq struct {
-	path       string
-	remoteAddr string
-	res        chan serverGetMuxerRes
+	path           string
+	remoteAddr     string
+	sourceOnDemand bool
+	res            chan serverGetMuxerRes
 }
 
 type serverAPIMuxersListRes struct {
@@ -46,6 +48,11 @@ type serverAPIMuxersGetReq struct {
 	res  chan serverAPIMuxersGetRes
 }
 
+type serverPathManager interface {
+	FindPathConf(req defs.PathFindPathConfReq) (*conf.Path, error)
+	AddReader(req defs.PathAddReaderReq) (defs.Path, *stream.Stream, error)
+}
+
 type serverParent interface {
 	logger.Writer
 }
@@ -68,7 +75,7 @@ type Server struct {
 	Directory                 string
 	ReadTimeout               conf.StringDuration
 	WriteQueueSize            int
-	PathManager               defs.PathManager
+	PathManager               serverPathManager
 	Parent                    serverParent
 
 	ctx        context.Context
@@ -162,17 +169,16 @@ outer:
 			switch {
 			case ok:
 				req.res <- serverGetMuxerRes{muxer: mux}
-			case !s.AlwaysRemux:
-				req.res <- serverGetMuxerRes{muxer: s.createMuxer(req.path, req.remoteAddr)}
-			default:
+			case s.AlwaysRemux && !req.sourceOnDemand:
 				req.res <- serverGetMuxerRes{err: fmt.Errorf("muxer is waiting to be created")}
+			default:
+				req.res <- serverGetMuxerRes{muxer: s.createMuxer(req.path, req.remoteAddr)}
 			}
 
 		case c := <-s.chCloseMuxer:
-			if c2, ok := s.muxers[c.PathName()]; !ok || c2 != c {
-				continue
+			if c2, ok := s.muxers[c.PathName()]; ok && c2 == c {
+				delete(s.muxers, c.PathName())
 			}
-			delete(s.muxers, c.PathName())
 
 		case req := <-s.chAPIMuxerList:
 			data := &defs.APIHLSMuxerList{

internal/servers/hls/server_test.go

@@ -0,0 +1,303 @@
+package hls
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/bluenviron/gohlslib"
+	"github.com/bluenviron/gohlslib/pkg/codecs"
+	"github.com/bluenviron/gortsplib/v4/pkg/description"
+	"github.com/bluenviron/gortsplib/v4/pkg/format"
+	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
+	"github.com/bluenviron/mediamtx/internal/conf"
+	"github.com/bluenviron/mediamtx/internal/defs"
+	"github.com/bluenviron/mediamtx/internal/externalcmd"
+	"github.com/bluenviron/mediamtx/internal/stream"
+	"github.com/bluenviron/mediamtx/internal/test"
+	"github.com/bluenviron/mediamtx/internal/unit"
+	"github.com/stretchr/testify/require"
+)
+
+type dummyPath struct{}
+
+func (pa *dummyPath) Name() string {
+	return "mystream"
+}
+
+func (pa *dummyPath) SafeConf() *conf.Path {
+	return &conf.Path{}
+}
+
+func (pa *dummyPath) ExternalCmdEnv() externalcmd.Environment {
+	return nil
+}
+
+func (pa *dummyPath) StartPublisher(_ defs.PathStartPublisherReq) (*stream.Stream, error) {
+	return nil, fmt.Errorf("unimplemented")
+}
+
+func (pa *dummyPath) StopPublisher(_ defs.PathStopPublisherReq) {
+}
+
+func (pa *dummyPath) RemovePublisher(_ defs.PathRemovePublisherReq) {
+}
+
+func (pa *dummyPath) RemoveReader(_ defs.PathRemoveReaderReq) {
+}
+
+type dummyPathManager struct {
+	stream *stream.Stream
+}
+
+func (pm *dummyPathManager) FindPathConf(_ defs.PathFindPathConfReq) (*conf.Path, error) {
+	return &conf.Path{}, nil
+}
+
+func (pm *dummyPathManager) AddReader(req defs.PathAddReaderReq) (defs.Path, *stream.Stream, error) {
+	if req.AccessRequest.Name == "nonexisting" {
+		return nil, nil, fmt.Errorf("not found")
+	}
+	return &dummyPath{}, pm.stream, nil
+}
+
+func TestServerNotFound(t *testing.T) {
+	for _, ca := range []string{
+		"always remux off",
+		"always remux on",
+	} {
+		t.Run(ca, func(t *testing.T) {
+			s := &Server{
+				Address:                   "127.0.0.1:8888",
+				Encryption:                false,
+				ServerKey:                 "",
+				ServerCert:                "",
+				ExternalAuthenticationURL: "",
+				AlwaysRemux:               ca == "always remux on",
+				Variant:                   conf.HLSVariant(gohlslib.MuxerVariantMPEGTS),
+				SegmentCount:              7,
+				SegmentDuration:           conf.StringDuration(1 * time.Second),
+				PartDuration:              conf.StringDuration(200 * time.Millisecond),
+				SegmentMaxSize:            50 * 1024 * 1024,
+				AllowOrigin:               "",
+				TrustedProxies:            conf.IPsOrCIDRs{},
+				Directory:                 "",
+				ReadTimeout:               conf.StringDuration(10 * time.Second),
+				WriteQueueSize:            512,
+				PathManager:               &dummyPathManager{},
+				Parent:                    &test.NilLogger{},
+			}
+			err := s.Initialize()
+			require.NoError(t, err)
+			defer s.Close()
+
+			hc := &http.Client{Transport: &http.Transport{}}
+
+			func() {
+				req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:8888/nonexisting/", nil)
+				require.NoError(t, err)
+
+				res, err := hc.Do(req)
+				require.NoError(t, err)
+				defer res.Body.Close()
+				require.Equal(t, http.StatusOK, res.StatusCode)
+			}()
+
+			func() {
+				req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:8888/nonexisting/index.m3u8", nil)
+				require.NoError(t, err)
+
+				res, err := hc.Do(req)
+				require.NoError(t, err)
+				defer res.Body.Close()
+				require.Equal(t, http.StatusNotFound, res.StatusCode)
+			}()
+		})
+	}
+}
+
+func TestServerRead(t *testing.T) {
+	t.Run("always remux off", func(t *testing.T) {
+		testMediaH264 := &description.Media{
+			Type:    description.MediaTypeVideo,
+			Formats: []format.Format{test.FormatH264},
+		}
+
+		desc := &description.Session{Medias: []*description.Media{testMediaH264}}
+
+		stream, err := stream.New(
+			1460,
+			desc,
+			true,
+			test.NilLogger{},
+		)
+		require.NoError(t, err)
+
+		pathManager := &dummyPathManager{stream: stream}
+
+		s := &Server{
+			Address:                   "127.0.0.1:8888",
+			Encryption:                false,
+			ServerKey:                 "",
+			ServerCert:                "",
+			ExternalAuthenticationURL: "",
+			AlwaysRemux:               false,
+			Variant:                   conf.HLSVariant(gohlslib.MuxerVariantMPEGTS),
+			SegmentCount:              7,
+			SegmentDuration:           conf.StringDuration(1 * time.Second),
+			PartDuration:              conf.StringDuration(200 * time.Millisecond),
+			SegmentMaxSize:            50 * 1024 * 1024,
+			AllowOrigin:               "",
+			TrustedProxies:            conf.IPsOrCIDRs{},
+			Directory:                 "",
+			ReadTimeout:               conf.StringDuration(10 * time.Second),
+			WriteQueueSize:            512,
+			PathManager:               pathManager,
+			Parent:                    &test.NilLogger{},
+		}
+		err = s.Initialize()
+		require.NoError(t, err)
+		defer s.Close()
+
+		c := &gohlslib.Client{
+			URI: "http://127.0.0.1:8888/mystream/index.m3u8",
+		}
+
+		recv := make(chan struct{})
+
+		c.OnTracks = func(tracks []*gohlslib.Track) error {
+			require.Equal(t, []*gohlslib.Track{{
+				Codec: &codecs.H264{},
+			}}, tracks)
+
+			c.OnDataH26x(tracks[0], func(pts, dts time.Duration, au [][]byte) {
+				require.Equal(t, time.Duration(0), pts)
+				require.Equal(t, time.Duration(0), dts)
+				require.Equal(t, [][]byte{
+					{byte(h264.NALUTypeAccessUnitDelimiter), 0xf0},
+					test.FormatH264.SPS,
+					test.FormatH264.PPS,
+					{5, 1},
+				}, au)
+				close(recv)
+			})
+
+			return nil
+		}
+
+		err = c.Start()
+		require.NoError(t, err)
+		defer func() { <-c.Wait() }()
+		defer c.Close()
+
+		go func() {
+			time.Sleep(100 * time.Millisecond)
+			for i := 0; i < 4; i++ {
+				stream.WriteUnit(testMediaH264, test.FormatH264, &unit.H264{
+					Base: unit.Base{
+						NTP: time.Time{},
+						PTS: time.Duration(i) * time.Second,
+					},
+					AU: [][]byte{
+						{5, 1}, // IDR
+					},
+				})
+			}
+		}()
+
+		<-recv
+	})
+
+	t.Run("always remux on", func(t *testing.T) {
+		testMediaH264 := &description.Media{
+			Type:    description.MediaTypeVideo,
+			Formats: []format.Format{test.FormatH264},
+		}
+
+		desc := &description.Session{Medias: []*description.Media{testMediaH264}}
+
+		stream, err := stream.New(
+			1460,
+			desc,
+			true,
+			test.NilLogger{},
+		)
+		require.NoError(t, err)
+
+		pathManager := &dummyPathManager{stream: stream}
+
+		s := &Server{
+			Address:                   "127.0.0.1:8888",
+			Encryption:                false,
+			ServerKey:                 "",
+			ServerCert:                "",
+			ExternalAuthenticationURL: "",
+			AlwaysRemux:               true,
+			Variant:                   conf.HLSVariant(gohlslib.MuxerVariantMPEGTS),
+			SegmentCount:              7,
+			SegmentDuration:           conf.StringDuration(1 * time.Second),
+			PartDuration:              conf.StringDuration(200 * time.Millisecond),
+			SegmentMaxSize:            50 * 1024 * 1024,
+			AllowOrigin:               "",
+			TrustedProxies:            conf.IPsOrCIDRs{},
+			Directory:                 "",
+			ReadTimeout:               conf.StringDuration(10 * time.Second),
+			WriteQueueSize:            512,
+			PathManager:               pathManager,
+			Parent:                    &test.NilLogger{},
+		}
+		err = s.Initialize()
+		require.NoError(t, err)
+		defer s.Close()
+
+		s.PathReady(&dummyPath{})
+
+		time.Sleep(100 * time.Millisecond)
+
+		for i := 0; i < 4; i++ {
+			stream.WriteUnit(testMediaH264, test.FormatH264, &unit.H264{
+				Base: unit.Base{
+					NTP: time.Time{},
+					PTS: time.Duration(i) * time.Second,
+				},
+				AU: [][]byte{
+					{5, 1}, // IDR
+				},
+			})
+		}
+
+		c := &gohlslib.Client{
+			URI: "http://127.0.0.1:8888/mystream/index.m3u8",
+		}
+
+		recv := make(chan struct{})
+
+		c.OnTracks = func(tracks []*gohlslib.Track) error {
+			require.Equal(t, []*gohlslib.Track{{
+				Codec: &codecs.H264{},
+			}}, tracks)
+
+			c.OnDataH26x(tracks[0], func(pts, dts time.Duration, au [][]byte) {
+				require.Equal(t, time.Duration(0), pts)
+				require.Equal(t, time.Duration(0), dts)
+				require.Equal(t, [][]byte{
+					{0x09, 0xf0},
+					test.FormatH264.SPS,
+					test.FormatH264.PPS,
+					{5, 1},
+				}, au)
+				close(recv)
+			})
+
+			return nil
+		}
+
+		err = c.Start()
+		require.NoError(t, err)
+		defer func() { <-c.Wait() }()
+		defer c.Close()
+
+		<-recv
+	})
+}

internal/servers/rtmp/conn.go

@@ -165,7 +165,7 @@ func (c *conn) runReader() error {
 func (c *conn) runRead(conn *rtmp.Conn, u *url.URL) error {
 	pathName, query, rawQuery := pathNameAndQuery(u)
 
-	res := c.pathManager.AddReader(defs.PathAddReaderReq{
+	path, stream, err := c.pathManager.AddReader(defs.PathAddReaderReq{
 		Author: c,
 		AccessRequest: defs.PathAccessRequest{
 			Name:  pathName,
@@ -177,18 +177,17 @@ func (c *conn) runRead(conn *rtmp.Conn, u *url.URL) error {
 			ID:    &c.uuid,
 		},
 	})
-
-	if res.Err != nil {
+	if err != nil {
 		var terr defs.AuthenticationError
-		if errors.As(res.Err, &terr) {
+		if errors.As(err, &terr) {
 			// wait some seconds to mitigate brute force attacks
 			<-time.After(pauseAfterAuthError)
 			return terr
 		}
-		return res.Err
+		return err
 	}
 
-	defer res.Path.RemoveReader(defs.PathRemoveReaderReq{Author: c})
+	defer path.RemoveReader(defs.PathRemoveReaderReq{Author: c})
 
 	c.mutex.Lock()
 	c.state = connStateRead
@@ -198,18 +197,18 @@ func (c *conn) runRead(conn *rtmp.Conn, u *url.URL) error {
 
 	writer := asyncwriter.New(c.writeQueueSize, c)
 
-	defer res.Stream.RemoveReader(writer)
+	defer stream.RemoveReader(writer)
 
 	var w *rtmp.Writer
 
 	videoFormat := c.setupVideo(
 		&w,
-		res.Stream,
+		stream,
 		writer)
 
 	audioFormat := c.setupAudio(
 		&w,
-		res.Stream,
+		stream,
 		writer)
 
 	if videoFormat == nil && audioFormat == nil {
@@ -218,19 +217,18 @@ func (c *conn) runRead(conn *rtmp.Conn, u *url.URL) error {
 	}
 
 	c.Log(logger.Info, "is reading from path '%s', %s",
-		res.Path.Name(), defs.FormatsInfo(res.Stream.FormatsForReader(writer)))
+		path.Name(), defs.FormatsInfo(stream.FormatsForReader(writer)))
 
 	onUnreadHook := hooks.OnRead(hooks.OnReadParams{
 		Logger:          c,
 		ExternalCmdPool: c.externalCmdPool,
-		Conf:            res.Path.SafeConf(),
-		ExternalCmdEnv:  res.Path.ExternalCmdEnv(),
+		Conf:            path.SafeConf(),
+		ExternalCmdEnv:  path.ExternalCmdEnv(),
 		Reader:          c.APISourceDescribe(),
 		Query:           rawQuery,
 	})
 	defer onUnreadHook()
 
-	var err error
 	w, err = rtmp.NewWriter(conn, videoFormat, audioFormat)
 	if err != nil {
 		return err
@@ -396,7 +394,7 @@ func (c *conn) setupAudio(
 func (c *conn) runPublish(conn *rtmp.Conn, u *url.URL) error {
 	pathName, query, rawQuery := pathNameAndQuery(u)
 
-	res := c.pathManager.AddPublisher(defs.PathAddPublisherReq{
+	path, err := c.pathManager.AddPublisher(defs.PathAddPublisherReq{
 		Author: c,
 		AccessRequest: defs.PathAccessRequest{
 			Name:    pathName,
@@ -409,18 +407,17 @@ func (c *conn) runPublish(conn *rtmp.Conn, u *url.URL) error {
 			ID:      &c.uuid,
 		},
 	})
-
-	if res.Err != nil {
+	if err != nil {
 		var terr defs.AuthenticationError
-		if errors.As(res.Err, &terr) {
+		if errors.As(err, &terr) {
 			// wait some seconds to mitigate brute force attacks
 			<-time.After(pauseAfterAuthError)
 			return terr
 		}
-		return res.Err
+		return err
 	}
 
-	defer res.Path.RemovePublisher(defs.PathRemovePublisherReq{Author: c})
+	defer path.RemovePublisher(defs.PathRemovePublisherReq{Author: c})
 
 	c.mutex.Lock()
 	c.state = connStatePublish
@@ -551,17 +548,15 @@ func (c *conn) runPublish(conn *rtmp.Conn, u *url.URL) error {
 		}
 	}
 
-	rres := res.Path.StartPublisher(defs.PathStartPublisherReq{
+	stream, err = path.StartPublisher(defs.PathStartPublisherReq{
 		Author:             c,
 		Desc:               &description.Session{Medias: medias},
 		GenerateRTPPackets: true,
 	})
-	if rres.Err != nil {
-		return rres.Err
+	if err != nil {
+		return err
 	}
 
-	stream = rres.Stream
-
 	// disable write deadline to allow outgoing acknowledges
 	c.nconn.SetWriteDeadline(time.Time{})
 

internal/servers/rtsp/session.go

@@ -110,7 +110,7 @@ func (s *session) onAnnounce(c *conn, ctx *gortsplib.ServerHandlerOnAnnounceCtx)
 		}
 	}
 
-	res := s.pathManager.AddPublisher(defs.PathAddPublisherReq{
+	path, err := s.pathManager.AddPublisher(defs.PathAddPublisherReq{
 		Author: s,
 		AccessRequest: defs.PathAccessRequest{
 			Name:        ctx.Path,
@@ -124,19 +124,18 @@ func (s *session) onAnnounce(c *conn, ctx *gortsplib.ServerHandlerOnAnnounceCtx)
 			RTSPNonce:   c.authNonce,
 		},
 	})
-
-	if res.Err != nil {
+	if err != nil {
 		var terr defs.AuthenticationError
-		if errors.As(res.Err, &terr) {
+		if errors.As(err, &terr) {
 			return c.handleAuthError(terr)
 		}
 
 		return &base.Response{
 			StatusCode: base.StatusBadRequest,
-		}, res.Err
+		}, err
 	}
 
-	s.path = res.Path
+	s.path = path
 
 	s.mutex.Lock()
 	s.state = gortsplib.ServerSessionStatePreRecord
@@ -196,7 +195,7 @@ func (s *session) onSetup(c *conn, ctx *gortsplib.ServerHandlerOnSetupCtx,
 			}
 		}
 
-		res := s.pathManager.AddReader(defs.PathAddReaderReq{
+		path, stream, err := s.pathManager.AddReader(defs.PathAddReaderReq{
 			Author: s,
 			AccessRequest: defs.PathAccessRequest{
 				Name:        ctx.Path,
@@ -209,28 +208,27 @@ func (s *session) onSetup(c *conn, ctx *gortsplib.ServerHandlerOnSetupCtx,
 				RTSPNonce:   c.authNonce,
 			},
 		})
-
-		if res.Err != nil {
+		if err != nil {
 			var terr defs.AuthenticationError
-			if errors.As(res.Err, &terr) {
+			if errors.As(err, &terr) {
 				res, err := c.handleAuthError(terr)
 				return res, nil, err
 			}
 
 			var terr2 defs.PathNoOnePublishingError
-			if errors.As(res.Err, &terr2) {
+			if errors.As(err, &terr2) {
 				return &base.Response{
 					StatusCode: base.StatusNotFound,
-				}, nil, res.Err
+				}, nil, err
 			}
 
 			return &base.Response{
 				StatusCode: base.StatusBadRequest,
-			}, nil, res.Err
+			}, nil, err
 		}
 
-		s.path = res.Path
-		s.stream = res.Stream
+		s.path = path
+		s.stream = stream
 
 		s.mutex.Lock()
 		s.state = gortsplib.ServerSessionStatePrePlay
@@ -238,16 +236,16 @@ func (s *session) onSetup(c *conn, ctx *gortsplib.ServerHandlerOnSetupCtx,
 		s.query = ctx.Query
 		s.mutex.Unlock()
 
-		var stream *gortsplib.ServerStream
+		var rstream *gortsplib.ServerStream
 		if !s.isTLS {
-			stream = res.Stream.RTSPStream(s.rserver)
+			rstream = stream.RTSPStream(s.rserver)
 		} else {
-			stream = res.Stream.RTSPSStream(s.rserver)
+			rstream = stream.RTSPSStream(s.rserver)
 		}
 
 		return &base.Response{
 			StatusCode: base.StatusOK,
-		}, stream, nil
+		}, rstream, nil
 
 	default: // record
 		return &base.Response{
@@ -289,18 +287,18 @@ func (s *session) onPlay(_ *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, e
 
 // onRecord is called by rtspServer.
 func (s *session) onRecord(_ *gortsplib.ServerHandlerOnRecordCtx) (*base.Response, error) {
-	res := s.path.StartPublisher(defs.PathStartPublisherReq{
+	stream, err := s.path.StartPublisher(defs.PathStartPublisherReq{
 		Author:             s,
 		Desc:               s.rsession.AnnouncedDescription(),
 		GenerateRTPPackets: false,
 	})
-	if res.Err != nil {
+	if err != nil {
 		return &base.Response{
 			StatusCode: base.StatusBadRequest,
-		}, res.Err
+		}, err
 	}
 
-	s.stream = res.Stream
+	s.stream = stream
 
 	for _, medi := range s.rsession.AnnouncedDescription().Medias {
 		for _, forma := range medi.Formats {
@@ -313,7 +311,7 @@ func (s *session) onRecord(_ *gortsplib.ServerHandlerOnRecordCtx) (*base.Respons
 					return
 				}
 
-				res.Stream.WriteRTPPacket(cmedi, cforma, pkt, time.Now(), pts)
+				stream.WriteRTPPacket(cmedi, cforma, pkt, time.Now(), pts)
 			})
 		}
 	}

internal/servers/srt/conn.go

@@ -163,7 +163,7 @@ func (c *conn) runInner2(req srtNewConnReq) (bool, error) {
 }
 
 func (c *conn) runPublish(req srtNewConnReq, streamID *streamID) (bool, error) {
-	res := c.pathManager.AddPublisher(defs.PathAddPublisherReq{
+	path, err := c.pathManager.AddPublisher(defs.PathAddPublisherReq{
 		Author: c,
 		AccessRequest: defs.PathAccessRequest{
 			Name:    streamID.path,
@@ -176,20 +176,19 @@ func (c *conn) runPublish(req srtNewConnReq, streamID *streamID) (bool, error) {
 			Query:   streamID.query,
 		},
 	})
-
-	if res.Err != nil {
+	if err != nil {
 		var terr defs.AuthenticationError
-		if errors.As(res.Err, &terr) {
+		if errors.As(err, &terr) {
 			// wait some seconds to mitigate brute force attacks
 			<-time.After(pauseAfterAuthError)
 			return false, terr
 		}
-		return false, res.Err
+		return false, err
 	}
 
-	defer res.Path.RemovePublisher(defs.PathRemovePublisherReq{Author: c})
+	defer path.RemovePublisher(defs.PathRemovePublisherReq{Author: c})
 
-	err := srtCheckPassphrase(req.connReq, res.Path.SafeConf().SRTPublishPassphrase)
+	err = srtCheckPassphrase(req.connReq, path.SafeConf().SRTPublishPassphrase)
 	if err != nil {
 		return false, err
 	}
@@ -208,7 +207,7 @@ func (c *conn) runPublish(req srtNewConnReq, streamID *streamID) (bool, error) {
 
 	readerErr := make(chan error)
 	go func() {
-		readerErr <- c.runPublishReader(sconn, res.Path)
+		readerErr <- c.runPublishReader(sconn, path)
 	}()
 
 	select {
@@ -243,17 +242,15 @@ func (c *conn) runPublishReader(sconn srt.Conn, path defs.Path) error {
 		return err
 	}
 
-	rres := path.StartPublisher(defs.PathStartPublisherReq{
+	stream, err = path.StartPublisher(defs.PathStartPublisherReq{
 		Author:             c,
 		Desc:               &description.Session{Medias: medias},
 		GenerateRTPPackets: true,
 	})
-	if rres.Err != nil {
-		return rres.Err
+	if err != nil {
+		return err
 	}
 
-	stream = rres.Stream
-
 	for {
 		err := r.Read()
 		if err != nil {
@@ -263,7 +260,7 @@ func (c *conn) runPublishReader(sconn srt.Conn, path defs.Path) error {
 }
 
 func (c *conn) runRead(req srtNewConnReq, streamID *streamID) (bool, error) {
-	res := c.pathManager.AddReader(defs.PathAddReaderReq{
+	path, stream, err := c.pathManager.AddReader(defs.PathAddReaderReq{
 		Author: c,
 		AccessRequest: defs.PathAccessRequest{
 			Name:  streamID.path,
@@ -275,20 +272,19 @@ func (c *conn) runRead(req srtNewConnReq, streamID *streamID) (bool, error) {
 			Query: streamID.query,
 		},
 	})
-
-	if res.Err != nil {
+	if err != nil {
 		var terr defs.AuthenticationError
-		if errors.As(res.Err, &terr) {
+		if errors.As(err, &terr) {
 			// wait some seconds to mitigate brute force attacks
 			<-time.After(pauseAfterAuthError)
-			return false, res.Err
+			return false, err
 		}
-		return false, res.Err
+		return false, err
 	}
 
-	defer res.Path.RemoveReader(defs.PathRemoveReaderReq{Author: c})
+	defer path.RemoveReader(defs.PathRemoveReaderReq{Author: c})
 
-	err := srtCheckPassphrase(req.connReq, res.Path.SafeConf().SRTReadPassphrase)
+	err = srtCheckPassphrase(req.connReq, path.SafeConf().SRTReadPassphrase)
 	if err != nil {
 		return false, err
 	}
@@ -308,23 +304,23 @@ func (c *conn) runRead(req srtNewConnReq, streamID *streamID) (bool, error) {
 
 	writer := asyncwriter.New(c.writeQueueSize, c)
 
-	defer res.Stream.RemoveReader(writer)
+	defer stream.RemoveReader(writer)
 
 	bw := bufio.NewWriterSize(sconn, srtMaxPayloadSize(c.udpMaxPayloadSize))
 
-	err = mpegts.FromStream(res.Stream, writer, bw, sconn, time.Duration(c.writeTimeout))
+	err = mpegts.FromStream(stream, writer, bw, sconn, time.Duration(c.writeTimeout))
 	if err != nil {
 		return true, err
 	}
 
 	c.Log(logger.Info, "is reading from path '%s', %s",
-		res.Path.Name(), defs.FormatsInfo(res.Stream.FormatsForReader(writer)))
+		path.Name(), defs.FormatsInfo(stream.FormatsForReader(writer)))
 
 	onUnreadHook := hooks.OnRead(hooks.OnReadParams{
 		Logger:          c,
 		ExternalCmdPool: c.externalCmdPool,
-		Conf:            res.Path.SafeConf(),
-		ExternalCmdEnv:  res.Path.ExternalCmdEnv(),
+		Conf:            path.SafeConf(),
+		ExternalCmdEnv:  path.ExternalCmdEnv(),
 		Reader:          c.APIReaderDescribe(),
 		Query:           streamID.query,
 	})

internal/servers/webrtc/http_server.go

@@ -109,7 +109,7 @@ func (s *httpServer) close() {
 func (s *httpServer) checkAuthOutsideSession(ctx *gin.Context, path string, publish bool) bool {
 	user, pass, hasCredentials := ctx.Request.BasicAuth()
 
-	res := s.pathManager.FindPathConf(defs.PathFindPathConfReq{
+	_, err := s.pathManager.FindPathConf(defs.PathFindPathConfReq{
 		AccessRequest: defs.PathAccessRequest{
 			Name:    path,
 			Query:   ctx.Request.URL.RawQuery,
@@ -120,9 +120,9 @@ func (s *httpServer) checkAuthOutsideSession(ctx *gin.Context, path string, publ
 			Proto:   defs.AuthProtocolWebRTC,
 		},
 	})
-	if res.Err != nil {
+	if err != nil {
 		var terr defs.AuthenticationError
-		if errors.As(res.Err, &terr) {
+		if errors.As(err, &terr) {
 			if !hasCredentials {
 				ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`)
 				ctx.Writer.WriteHeader(http.StatusUnauthorized)
@@ -138,7 +138,7 @@ func (s *httpServer) checkAuthOutsideSession(ctx *gin.Context, path string, publ
 			return false
 		}
 
-		writeError(ctx, http.StatusInternalServerError, res.Err)
+		writeError(ctx, http.StatusInternalServerError, err)
 		return false
 	}
 

internal/servers/webrtc/session.go

@@ -363,7 +363,7 @@ func (s *session) runInner2() (int, error) {
 func (s *session) runPublish() (int, error) {
 	ip, _, _ := net.SplitHostPort(s.req.remoteAddr)
 
-	res := s.pathManager.AddPublisher(defs.PathAddPublisherReq{
+	path, err := s.pathManager.AddPublisher(defs.PathAddPublisherReq{
 		Author: s,
 		AccessRequest: defs.PathAccessRequest{
 			Name:    s.req.pathName,
@@ -376,19 +376,19 @@ func (s *session) runPublish() (int, error) {
 			ID:      &s.uuid,
 		},
 	})
-	if res.Err != nil {
+	if err != nil {
 		var terr defs.AuthenticationError
-		if errors.As(res.Err, &terr) {
+		if errors.As(err, &terr) {
 			// wait some seconds to mitigate brute force attacks
 			<-time.After(pauseAfterAuthError)
 
-			return http.StatusUnauthorized, res.Err
+			return http.StatusUnauthorized, err
 		}
 
-		return http.StatusBadRequest, res.Err
+		return http.StatusBadRequest, err
 	}
 
-	defer res.Path.RemovePublisher(defs.PathRemovePublisherReq{Author: s})
+	defer path.RemovePublisher(defs.PathRemovePublisherReq{Author: s})
 
 	iceServers, err := s.parent.generateICEServers()
 	if err != nil {
@@ -450,13 +450,13 @@ func (s *session) runPublish() (int, error) {
 
 	medias := webrtc.TracksToMedias(tracks)
 
-	rres := res.Path.StartPublisher(defs.PathStartPublisherReq{
+	stream, err := path.StartPublisher(defs.PathStartPublisherReq{
 		Author:             s,
 		Desc:               &description.Session{Medias: medias},
 		GenerateRTPPackets: false,
 	})
-	if rres.Err != nil {
-		return 0, rres.Err
+	if err != nil {
+		return 0, err
 	}
 
 	timeDecoder := rtptime.NewGlobalDecoder()
@@ -478,7 +478,7 @@ func (s *session) runPublish() (int, error) {
 					continue
 				}
 
-				rres.Stream.WriteRTPPacket(cmedia, cmedia.Formats[0], pkt, time.Now(), pts)
+				stream.WriteRTPPacket(cmedia, cmedia.Formats[0], pkt, time.Now(), pts)
 			}
 		}()
 	}
@@ -495,7 +495,7 @@ func (s *session) runPublish() (int, error) {
 func (s *session) runRead() (int, error) {
 	ip, _, _ := net.SplitHostPort(s.req.remoteAddr)
 
-	res := s.pathManager.AddReader(defs.PathAddReaderReq{
+	path, stream, err := s.pathManager.AddReader(defs.PathAddReaderReq{
 		Author: s,
 		AccessRequest: defs.PathAccessRequest{
 			Name:  s.req.pathName,
@@ -507,22 +507,22 @@ func (s *session) runRead() (int, error) {
 			ID:    &s.uuid,
 		},
 	})
-	if res.Err != nil {
+	if err != nil {
 		var terr defs.AuthenticationError
-		if errors.As(res.Err, &terr) {
+		if errors.As(err, &terr) {
 			// wait some seconds to mitigate brute force attacks
 			<-time.After(pauseAfterAuthError)
-			return http.StatusUnauthorized, res.Err
+			return http.StatusUnauthorized, err
 		}
 
-		if strings.HasPrefix(res.Err.Error(), "no one is publishing") {
-			return http.StatusNotFound, res.Err
+		if strings.HasPrefix(err.Error(), "no one is publishing") {
+			return http.StatusNotFound, err
 		}
 
-		return http.StatusBadRequest, res.Err
+		return http.StatusBadRequest, err
 	}
 
-	defer res.Path.RemoveReader(defs.PathRemoveReaderReq{Author: s})
+	defer path.RemoveReader(defs.PathRemoveReaderReq{Author: s})
 
 	iceServers, err := s.parent.generateICEServers()
 	if err != nil {
@@ -543,8 +543,8 @@ func (s *session) runRead() (int, error) {
 
 	writer := asyncwriter.New(s.writeQueueSize, s)
 
-	videoTrack, videoSetup := findVideoTrack(res.Stream, writer)
-	audioTrack, audioSetup := findAudioTrack(res.Stream, writer)
+	videoTrack, videoSetup := findVideoTrack(stream, writer)
+	audioTrack, audioSetup := findAudioTrack(stream, writer)
 
 	if videoTrack == nil && audioTrack == nil {
 		return http.StatusBadRequest, fmt.Errorf(
@@ -576,7 +576,7 @@ func (s *session) runRead() (int, error) {
 	s.pc = pc
 	s.mutex.Unlock()
 
-	defer res.Stream.RemoveReader(writer)
+	defer stream.RemoveReader(writer)
 
 	n := 0
 
@@ -596,13 +596,13 @@ func (s *session) runRead() (int, error) {
 	}
 
 	s.Log(logger.Info, "is reading from path '%s', %s",
-		res.Path.Name(), defs.FormatsInfo(res.Stream.FormatsForReader(writer)))
+		path.Name(), defs.FormatsInfo(stream.FormatsForReader(writer)))
 
 	onUnreadHook := hooks.OnRead(hooks.OnReadParams{
 		Logger:          s,
 		ExternalCmdPool: s.externalCmdPool,
-		Conf:            res.Path.SafeConf(),
-		ExternalCmdEnv:  res.Path.ExternalCmdEnv(),
+		Conf:            path.SafeConf(),
+		ExternalCmdEnv:  path.ExternalCmdEnv(),
 		Reader:          s.APIReaderDescribe(),
 		Query:           s.req.query,
 	})

internal/staticsources/rtmp/source_test.go

@@ -7,8 +7,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/bluenviron/gortsplib/v4/pkg/format"
-	"github.com/bluenviron/mediacommon/pkg/codecs/mpeg4audio"
 	"github.com/stretchr/testify/require"
 
 	"github.com/bluenviron/mediamtx/internal/conf"
@@ -120,30 +118,7 @@ func TestSource(t *testing.T) {
 				conn, _, _, err := rtmp.NewServerConn(nconn)
 				require.NoError(t, err)
 
-				videoTrack := &format.H264{
-					PayloadTyp: 96,
-					SPS: []byte{ // 1920x1080 baseline
-						0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-						0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-						0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20,
-					},
-					PPS:               []byte{0x08, 0x06, 0x07, 0x08},
-					PacketizationMode: 1,
-				}
-
-				audioTrack := &format.MPEG4Audio{
-					PayloadTyp: 96,
-					Config: &mpeg4audio.Config{
-						Type:         2,
-						SampleRate:   44100,
-						ChannelCount: 2,
-					},
-					SizeLength:       13,
-					IndexLength:      3,
-					IndexDeltaLength: 3,
-				}
-
-				w, err := rtmp.NewWriter(conn, videoTrack, audioTrack)
+				w, err := rtmp.NewWriter(conn, test.FormatH264, test.FormatMPEG4Audio)
 				require.NoError(t, err)
 
 				err = w.WriteH264(0, 0, true, [][]byte{{0x05, 0x02, 0x03, 0x04}})

internal/staticsources/rtsp/source_test.go

@@ -106,17 +106,8 @@ func (sh *testServer) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo
 }
 
 var testMediaH264 = &description.Media{
-	Type: description.MediaTypeVideo,
-	Formats: []format.Format{&format.H264{
-		PayloadTyp: 96,
-		SPS: []byte{ // 1920x1080 baseline
-			0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
-			0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
-			0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20,
-		},
-		PPS:               []byte{0x08, 0x06, 0x07, 0x08},
-		PacketizationMode: 1,
-	}},
+	Type:    description.MediaTypeVideo,
+	Formats: []format.Format{test.FormatH264},
 }
 
 func TestSource(t *testing.T) {

internal/test/formats.go

@@ -0,0 +1,31 @@
+package test
+
+import (
+	"github.com/bluenviron/gortsplib/v4/pkg/format"
+	"github.com/bluenviron/mediacommon/pkg/codecs/mpeg4audio"
+)
+
+// FormatH264 is a test H264 format.
+var FormatH264 = &format.H264{
+	PayloadTyp: 96,
+	SPS: []byte{ // 1920x1080 baseline
+		0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
+		0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
+		0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20,
+	},
+	PPS:               []byte{0x08, 0x06, 0x07, 0x08},
+	PacketizationMode: 1,
+}
+
+// FormatMPEG4Audio is a test MPEG-4 audio format.
+var FormatMPEG4Audio = &format.MPEG4Audio{
+	PayloadTyp: 96,
+	Config: &mpeg4audio.Config{
+		Type:         2,
+		SampleRate:   44100,
+		ChannelCount: 2,
+	},
+	SizeLength:       13,
+	IndexLength:      3,
+	IndexDeltaLength: 3,
+}

scripts/lint.mk

@@ -1,4 +1,5 @@
 lint:
+	touch internal/servers/hls/hls.min.js
 	docker run --rm -v $(PWD):/app -w /app \
 	$(LINT_IMAGE) \
 	golangci-lint run -v