diff --git a/.github/workflows/code_lint.yml b/.github/workflows/code_lint.yml index e21c9a71..e8ba45da 100644 --- a/.github/workflows/code_lint.yml +++ b/.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 diff --git a/internal/core/api_test.go b/internal/core/api_test.go index 9aa23013..04769624 100644 --- a/internal/core/api_test.go +++ b/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": diff --git a/internal/core/hls_server_test.go b/internal/core/hls_server_test.go deleted file mode 100644 index 71cbf6a6..00000000 --- a/internal/core/hls_server_test.go +++ /dev/null @@ -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¶m=value" && - in.Query != "user=testpublisher&pass=testpass¶m=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)*/ -} diff --git a/internal/core/metrics_test.go b/internal/core/metrics_test.go index 2f380b45..bf2a54e1 100644 --- a/internal/core/metrics_test.go +++ b/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) diff --git a/internal/core/path.go b/internal/core/path.go index 40a7287b..fd2aba37 100644 --- a/internal/core/path.go +++ b/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") } } diff --git a/internal/core/path_manager.go b/internal/core/path_manager.go index ae38c47c..011275a5 100644 --- a/internal/core/path_manager.go +++ b/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") } } diff --git a/internal/core/rtmp_server_test.go b/internal/core/rtmp_server_test.go index f7c81a7b..2fdbc821 100644 --- a/internal/core/rtmp_server_test.go +++ b/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¶m=value" && + in.Query != "user=testpublisher&pass=testpass¶m=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, }, diff --git a/internal/core/srt_server_test.go b/internal/core/srt_server_test.go index fa55d6f7..6dc5a984 100644 --- a/internal/core/srt_server_test.go +++ b/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 }) diff --git a/internal/defs/path.go b/internal/defs/path.go index fb5be368..d9d98bb9 100644 --- a/internal/defs/path.go +++ b/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) diff --git a/internal/defs/path_manager.go b/internal/defs/path_manager.go index 98cd1ca3..2e997622 100644 --- a/internal/defs/path_manager.go +++ b/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) } diff --git a/internal/playback/fmp4_test.go b/internal/playback/fmp4_test.go index 1df5ef19..8ecd003d 100644 --- a/internal/playback/fmp4_test.go +++ b/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, }, }, { diff --git a/internal/playback/server_test.go b/internal/playback/server_test.go index 99bd4bf6..89f9adb4 100644 --- a/internal/playback/server_test.go +++ b/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, }, }}, } diff --git a/internal/record/agent_test.go b/internal/record/agent_test.go index 3fb8689d..4e4dd3dc 100644 --- a/internal/record/agent_test.go +++ b/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 }, }) diff --git a/internal/record/format_fmp4.go b/internal/record/format_fmp4.go index 309b7513..b26b2a84 100644 --- a/internal/record/format_fmp4.go +++ b/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{ diff --git a/internal/servers/hls/http_server.go b/internal/servers/hls/http_server.go index 8d6398b0..aa70a3c9 100644 --- a/internal/servers/hls/http_server.go +++ b/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) } } diff --git a/internal/servers/hls/muxer.go b/internal/servers/hls/muxer.go index 1f7211a9..a840bfc5 100644 --- a/internal/servers/hls/muxer.go +++ b/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 } } diff --git a/internal/servers/hls/muxer_instance.go b/internal/servers/hls/muxer_instance.go new file mode 100644 index 00000000..cd122c72 --- /dev/null +++ b/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) +} diff --git a/internal/servers/hls/server.go b/internal/servers/hls/server.go index 98856226..34b8fc9f 100644 --- a/internal/servers/hls/server.go +++ b/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{ diff --git a/internal/servers/hls/server_test.go b/internal/servers/hls/server_test.go new file mode 100644 index 00000000..d9a5603a --- /dev/null +++ b/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 + }) +} diff --git a/internal/servers/rtmp/conn.go b/internal/servers/rtmp/conn.go index b6c1cf54..4d145c86 100644 --- a/internal/servers/rtmp/conn.go +++ b/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{}) diff --git a/internal/servers/rtsp/session.go b/internal/servers/rtsp/session.go index b4995bf0..2bd95715 100644 --- a/internal/servers/rtsp/session.go +++ b/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) }) } } diff --git a/internal/servers/srt/conn.go b/internal/servers/srt/conn.go index 1719bf83..4de843b1 100644 --- a/internal/servers/srt/conn.go +++ b/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, }) diff --git a/internal/servers/webrtc/http_server.go b/internal/servers/webrtc/http_server.go index 8b8e5fe6..fdd48761 100644 --- a/internal/servers/webrtc/http_server.go +++ b/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 } diff --git a/internal/servers/webrtc/session.go b/internal/servers/webrtc/session.go index d604f4e3..db922c99 100644 --- a/internal/servers/webrtc/session.go +++ b/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, }) diff --git a/internal/staticsources/rtmp/source_test.go b/internal/staticsources/rtmp/source_test.go index 4aaed6ba..c14dd5fd 100644 --- a/internal/staticsources/rtmp/source_test.go +++ b/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}}) diff --git a/internal/staticsources/rtsp/source_test.go b/internal/staticsources/rtsp/source_test.go index 3a9eb999..be46aeb6 100644 --- a/internal/staticsources/rtsp/source_test.go +++ b/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) { diff --git a/internal/test/formats.go b/internal/test/formats.go new file mode 100644 index 00000000..e863137f --- /dev/null +++ b/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, +} diff --git a/scripts/lint.mk b/scripts/lint.mk index 9c73673c..5245ac8c 100644 --- a/scripts/lint.mk +++ b/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