From f87d9e659e1cd982b04711c022e3a4f14becbe6a Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Sat, 25 Apr 2026 21:10:34 +0200 Subject: [PATCH] hls: track sessions (#962) (#5683) sessions are now tracked through cookies or query parameters. This provides the ability to inspect sessions through logs, metrics and API, allows more precise tracking of outbound bytes, decreases load on external HTTP authentication URLs since they are now called once per session and not once per request. --- api/openapi.yaml | 154 ++++++++- docs/2-features/21-metrics.md | 7 +- docs/4-read/13-web-browsers.md | 2 + internal/api/api.go | 3 + internal/api/api_hls.go | 61 ++++ internal/api/api_hls_test.go | 222 ++++++++++++- internal/api/api_paths_test.go | 2 +- internal/auth/error.go | 2 +- internal/core/api_test.go | 103 +++++- internal/core/core.go | 1 + internal/core/metrics_test.go | 4 + internal/core/path.go | 14 +- internal/defs/api_hls.go | 41 ++- internal/defs/api_path.go | 20 +- internal/defs/path.go | 2 +- internal/linters/go2api/go2api_test.go | 11 +- internal/metrics/metrics.go | 29 ++ internal/metrics/metrics_test.go | 57 ++++ internal/protocols/httpp/handler_origin.go | 5 +- .../protocols/httpp/handler_origin_test.go | 2 +- internal/recorder/recorder_instance.go | 4 +- internal/servers/hls/http_server.go | 235 +++++++++++--- internal/servers/hls/muxer.go | 307 +++++++++++++----- internal/servers/hls/muxer_instance.go | 81 ++++- .../servers/hls/response_writer_counter.go | 18 + .../servers/hls/response_writer_no_cache.go | 19 ++ internal/servers/hls/server.go | 148 ++++++++- internal/servers/hls/server_test.go | 152 ++++++++- internal/servers/hls/session.go | 168 ++++++++++ internal/servers/rtsp/conn.go | 2 +- internal/servers/rtsp/session.go | 2 +- internal/servers/webrtc/server_test.go | 2 +- internal/servers/webrtc/session.go | 22 +- internal/staticsources/rpicamera/source.go | 4 +- internal/stream/reader.go | 4 +- internal/stream/stream.go | 12 +- internal/stream/stream_format.go | 5 +- internal/stream/stream_media.go | 9 +- internal/stream/stream_standard_test.go | 4 +- internal/stream/sub_stream_format.go | 6 +- 40 files changed, 1694 insertions(+), 252 deletions(-) create mode 100644 internal/servers/hls/response_writer_counter.go create mode 100644 internal/servers/hls/response_writer_no_cache.go create mode 100644 internal/servers/hls/session.go diff --git a/api/openapi.yaml b/api/openapi.yaml index afffcb31..98099606 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -45,8 +45,7 @@ components: PathReaderType: type: string enum: - - hlsMuxer - - rpiCameraSecondary + - hlsSession - rtmpConn - rtmpsConn - rtspConn @@ -1196,6 +1195,40 @@ components: items: $ref: '#/components/schemas/HLSMuxer' + HLSSession: + type: object + properties: + id: + type: string + format: uuid + created: + type: string + remoteAddr: + type: string + path: + type: string + query: + type: string + user: + type: string + outboundBytes: + type: integer + format: uint64 + + HLSSessionList: + type: object + properties: + pageCount: + type: integer + format: int64 + itemCount: + type: integer + format: int64 + items: + type: array + items: + $ref: '#/components/schemas/HLSSession' + Recording: type: object properties: @@ -2291,6 +2324,123 @@ paths: schema: $ref: '#/components/schemas/Error' + /v3/hlssessions/list: + get: + operationId: hlssessionsList + tags: [HLS] + summary: returns all HLS sessions. + description: '' + parameters: + - name: page + in: query + description: page number. + schema: + type: integer + default: 0 + - name: itemsPerPage + in: query + description: items per page. + schema: + type: integer + default: 100 + responses: + '200': + description: the request was successful. + content: + application/json: + schema: + $ref: '#/components/schemas/HLSSessionList' + '400': + description: invalid request. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '500': + description: server error. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + + /v3/hlssessions/get/{id}: + get: + operationId: hlssessionsGet + tags: [HLS] + summary: returns a HLS session. + description: '' + parameters: + - name: id + in: path + required: true + description: ID of the session. + schema: + type: string + responses: + '200': + description: the request was successful. + content: + application/json: + schema: + $ref: '#/components/schemas/HLSSession' + '400': + description: invalid request. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '404': + description: session not found. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '500': + description: server error. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + + /v3/hlssessions/kick/{id}: + post: + operationId: hlssessionsKick + tags: [HLS] + summary: kicks out a HLS session from the server. + description: '' + parameters: + - name: id + in: path + required: true + description: ID of the session. + schema: + type: string + responses: + '200': + description: the request was successful. + content: + application/json: + schema: + $ref: '#/components/schemas/OK' + '400': + description: invalid request. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '404': + description: session not found. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '500': + description: server error. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + /v3/paths/list: get: operationId: pathsList diff --git a/docs/2-features/21-metrics.md b/docs/2-features/21-metrics.md index ccc5e003..d3c30a00 100644 --- a/docs/2-features/21-metrics.md +++ b/docs/2-features/21-metrics.md @@ -20,6 +20,10 @@ paths_inbound_bytes{name="[path_name]",state="[state]"} 1234 paths_outbound_bytes{name="[path_name]",state="[state]"} 1234 paths_inbound_frames_in_error{name="[path_name]",state="[state]"} 1234 +# HLS sessions +hls_sessions{id="[id]",path="[path]",remoteAddr="[remoteAddr]"} 1 +hls_sessions_outbound_bytes{id="[id]",path="[path]",remoteAddr="[remoteAddr]"} 187 + # HLS muxers hls_muxers{name="[name]"} 1 hls_muxers_outbound_bytes{name="[name]"} 187 @@ -151,9 +155,10 @@ Bitrates are not provided directly as metrics because they can be computed from Metrics can be filtered by using HTTP query parameters: -- `type=[TYPE]`: show metrics of a certain type only. TYPE can be `paths`, `hls_muxers`, `rtsp_conns`, `rtsp_sessions`, `rtsps_conns`, `rtsps_sessions`, `rtmp_conns`, `rtmps_conns`, `srt_conns`, `webrtc_sessions`. +- `type=[TYPE]`: show metrics of a certain type only. TYPE can be `paths`, `hls_sessions`, `hls_muxers`, `rtsp_conns`, `rtsp_sessions`, `rtsps_conns`, `rtsps_sessions`, `rtmp_conns`, `rtmps_conns`, `srt_conns`, `webrtc_sessions`. - `path=[PATH]`: show metrics belonging to a specific path only - `hls_muxer=[PATH]`: show metrics belonging to a specific HLS muxer only +- `hls_session=[ID]`: show metrics belonging to a specific HLS session only - `rtsp_conn=[ID]` show metrics belonging to a specific RTSP connection only - `rtsp_session=[SESSION]`: show metrics belonging to a specific RTSP session only - `rtsps_conn=[ID]` show metrics belonging to a specific RTSPS connection only diff --git a/docs/4-read/13-web-browsers.md b/docs/4-read/13-web-browsers.md index bfb6a7cd..c0feff45 100644 --- a/docs/4-read/13-web-browsers.md +++ b/docs/4-read/13-web-browsers.md @@ -164,6 +164,8 @@ After the video tag, add a script that initializes the stream when the page is f if (Hls.isSupported()) { const hls = new Hls({ xhrSetup: function (xhr, url) { + xhr.withCredentials = true; + let user = ""; // fill if needed let pass = ""; // fill if needed let token = ""; // fill if needed diff --git a/internal/api/api.go b/internal/api/api.go index 512b04dd..2da565e3 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -120,6 +120,9 @@ func (a *API) Initialize() error { if !interfaceIsEmpty(a.HLSServer) { group.GET("/hlsmuxers/list", a.onHLSMuxersList) group.GET("/hlsmuxers/get/*name", a.onHLSMuxersGet) + group.GET("/hlssessions/list", a.onHLSSessionsList) + group.GET("/hlssessions/get/:id", a.onHLSSessionsGet) + group.POST("/hlssessions/kick/:id", a.onHLSSessionsKick) } if !interfaceIsEmpty(a.RTSPServer) { diff --git a/internal/api/api_hls.go b/internal/api/api_hls.go index 5b9abfd3..961512f5 100644 --- a/internal/api/api_hls.go +++ b/internal/api/api_hls.go @@ -8,6 +8,7 @@ import ( "github.com/bluenviron/mediamtx/internal/servers/hls" "github.com/gin-gonic/gin" + "github.com/google/uuid" ) func (a *API) onHLSMuxersList(ctx *gin.Context) { @@ -47,3 +48,63 @@ func (a *API) onHLSMuxersGet(ctx *gin.Context) { ctx.JSON(http.StatusOK, data) } + +func (a *API) onHLSSessionsList(ctx *gin.Context) { + data, err := a.HLSServer.APISessionsList() + if err != nil { + a.writeError(ctx, http.StatusInternalServerError, err) + return + } + + data.ItemCount = len(data.Items) + pageCount, err := paginate(&data.Items, ctx.Query("itemsPerPage"), ctx.Query("page")) + if err != nil { + a.writeError(ctx, http.StatusBadRequest, err) + return + } + data.PageCount = pageCount + + ctx.JSON(http.StatusOK, data) +} + +func (a *API) onHLSSessionsGet(ctx *gin.Context) { + id := ctx.Param("id") + + uuid, err := uuid.Parse(id) + if err != nil { + a.writeError(ctx, http.StatusBadRequest, err) + return + } + + data, err := a.HLSServer.APISessionsGet(uuid) + if err != nil { + if errors.Is(err, hls.ErrSessionNotFound) { + a.writeError(ctx, http.StatusNotFound, err) + } else { + a.writeError(ctx, http.StatusInternalServerError, err) + } + return + } + + ctx.JSON(http.StatusOK, data) +} + +func (a *API) onHLSSessionsKick(ctx *gin.Context) { + uuid, err := uuid.Parse(ctx.Param("id")) + if err != nil { + a.writeError(ctx, http.StatusBadRequest, err) + return + } + + err = a.HLSServer.APISessionsKick(uuid) + if err != nil { + if errors.Is(err, hls.ErrSessionNotFound) { + a.writeError(ctx, http.StatusNotFound, err) + } else { + a.writeError(ctx, http.StatusInternalServerError, err) + } + return + } + + a.writeOK(ctx) +} diff --git a/internal/api/api_hls_test.go b/internal/api/api_hls_test.go index 71d57000..f370dd26 100644 --- a/internal/api/api_hls_test.go +++ b/internal/api/api_hls_test.go @@ -1,7 +1,9 @@ package api //nolint:revive import ( + "fmt" "net/http" + "sort" "testing" "time" @@ -9,13 +11,17 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/servers/hls" "github.com/bluenviron/mediamtx/internal/test" + "github.com/google/uuid" "github.com/stretchr/testify/require" ) type testHLSServer struct { - muxers map[string]*defs.APIHLSMuxer + muxers map[string]*defs.APIHLSMuxer + sessions map[string]*defs.APIHLSSession } +var testTime = time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC) + func (s *testHLSServer) APIMuxersList() (*defs.APIHLSMuxerList, error) { items := make([]defs.APIHLSMuxer, 0, len(s.muxers)) for _, muxer := range s.muxers { @@ -32,8 +38,36 @@ func (s *testHLSServer) APIMuxersGet(name string) (*defs.APIHLSMuxer, error) { return muxer, nil } +func (s *testHLSServer) APISessionsList() (*defs.APIHLSSessionList, error) { + items := make([]defs.APIHLSSession, 0, len(s.sessions)) + for _, session := range s.sessions { + items = append(items, *session) + } + sort.Slice(items, func(i, j int) bool { + return items[i].Created.Before(items[j].Created) + }) + return &defs.APIHLSSessionList{Items: items}, nil +} + +func (s *testHLSServer) APISessionsGet(id uuid.UUID) (*defs.APIHLSSession, error) { + session, ok := s.sessions[id.String()] + if !ok { + return nil, hls.ErrSessionNotFound + } + return session, nil +} + +func (s *testHLSServer) APISessionsKick(id uuid.UUID) error { + _, ok := s.sessions[id.String()] + if !ok { + return hls.ErrSessionNotFound + } + delete(s.sessions, id.String()) + return nil +} + func TestHLSMuxersList(t *testing.T) { - now := time.Now() + now := testTime hlsServer := &testHLSServer{ muxers: map[string]*defs.APIHLSMuxer{ "test1": { @@ -80,7 +114,7 @@ func TestHLSMuxersList(t *testing.T) { } func TestHLSMuxersGet(t *testing.T) { - now := time.Now() + now := testTime hlsServer := &testHLSServer{ muxers: map[string]*defs.APIHLSMuxer{ "mypath": { @@ -118,3 +152,185 @@ func TestHLSMuxersGet(t *testing.T) { require.Equal(t, uint64(12), out.OutboundFramesDiscarded) require.Equal(t, uint64(9999), out.BytesSent) } + +func TestHLSSessionsList(t *testing.T) { + now := testTime + hlsServer := &testHLSServer{ + sessions: map[string]*defs.APIHLSSession{ + "session1": { + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb41"), + Created: now, + RemoteAddr: "192.168.1.1:5000", + Path: "stream1", + Query: "key=val1", + User: "user1", + OutboundBytes: 111, + }, + "session2": { + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb42"), + Created: now.Add(time.Minute), + RemoteAddr: "192.168.1.2:5001", + Path: "stream2", + Query: "key=val2", + User: "user2", + OutboundBytes: 222, + }, + }, + } + + api := API{ + Address: "localhost:9997", + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + AuthManager: test.NilAuthManager, + HLSServer: hlsServer, + Parent: &testParent{}, + } + err := api.Initialize() + require.NoError(t, err) + defer api.Close() + + tr := &http.Transport{} + defer tr.CloseIdleConnections() + hc := &http.Client{Transport: tr} + + var out defs.APIHLSSessionList + httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/hlssessions/list", nil, &out) + + require.Equal(t, 2, out.ItemCount) + require.Equal(t, 1, out.PageCount) + require.Len(t, out.Items, 2) + require.Equal(t, []defs.APIHLSSession{ + { + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb41"), + Created: now, + RemoteAddr: "192.168.1.1:5000", + Path: "stream1", + Query: "key=val1", + User: "user1", + OutboundBytes: 111, + }, + { + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb42"), + Created: now.Add(time.Minute), + RemoteAddr: "192.168.1.2:5001", + Path: "stream2", + Query: "key=val2", + User: "user2", + OutboundBytes: 222, + }, + }, out.Items) +} + +func TestHLSSessionsGet(t *testing.T) { + now := testTime + hlsServer := &testHLSServer{ + sessions: map[string]*defs.APIHLSSession{ + "18294761-f9d1-4ea9-9a35-fe265b62eb41": { + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb41"), + Created: now, + RemoteAddr: "192.168.1.100:5000", + Path: "mystream", + Query: "key=val", + User: "myuser", + OutboundBytes: 345, + }, + }, + } + + api := API{ + Address: "localhost:9997", + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + AuthManager: test.NilAuthManager, + HLSServer: hlsServer, + Parent: &testParent{}, + } + err := api.Initialize() + require.NoError(t, err) + defer api.Close() + + tr := &http.Transport{} + defer tr.CloseIdleConnections() + hc := &http.Client{Transport: tr} + + sessionID := "18294761-f9d1-4ea9-9a35-fe265b62eb41" + + var out defs.APIHLSSession + httpRequest(t, hc, http.MethodGet, fmt.Sprintf("http://localhost:9997/v3/hlssessions/get/%s", sessionID), nil, &out) + + require.Equal(t, uuid.MustParse(sessionID), out.ID) + require.Equal(t, "192.168.1.100:5000", out.RemoteAddr) + require.Equal(t, "mystream", out.Path) + require.Equal(t, "key=val", out.Query) + require.Equal(t, "myuser", out.User) + require.Equal(t, uint64(345), out.OutboundBytes) +} + +func TestHLSSessionsKick(t *testing.T) { + now := testTime + sessionID := uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb41") + hlsServer := &testHLSServer{ + sessions: map[string]*defs.APIHLSSession{ + sessionID.String(): { + ID: sessionID, + Created: now, + RemoteAddr: "192.168.1.100:5000", + Path: "mystream", + OutboundBytes: 345, + }, + }, + } + + api := API{ + Address: "localhost:9997", + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + AuthManager: test.NilAuthManager, + HLSServer: hlsServer, + Parent: &testParent{}, + } + err := api.Initialize() + require.NoError(t, err) + defer api.Close() + + tr := &http.Transport{} + defer tr.CloseIdleConnections() + hc := &http.Client{Transport: tr} + + httpRequest(t, hc, http.MethodPost, fmt.Sprintf("http://localhost:9997/v3/hlssessions/kick/%s", sessionID), nil, nil) + + _, ok := hlsServer.sessions[sessionID.String()] + require.False(t, ok) +} + +func TestHLSSessionsKickNotFound(t *testing.T) { + hlsServer := &testHLSServer{} + + api := API{ + Address: "localhost:9997", + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + AuthManager: test.NilAuthManager, + HLSServer: hlsServer, + Parent: &testParent{}, + } + err := api.Initialize() + require.NoError(t, err) + defer api.Close() + + tr := &http.Transport{} + defer tr.CloseIdleConnections() + hc := &http.Client{Transport: tr} + + req, err := http.NewRequest(http.MethodPost, + fmt.Sprintf("http://localhost:9997/v3/hlssessions/kick/%s", uuid.New()), 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) + checkError(t, res.Body, "session not found") +} diff --git a/internal/api/api_paths_test.go b/internal/api/api_paths_test.go index a5d4ee74..e27dc985 100644 --- a/internal/api/api_paths_test.go +++ b/internal/api/api_paths_test.go @@ -108,7 +108,7 @@ func TestPathsGet(t *testing.T) { BytesReceived: 123456, BytesSent: 789012, Readers: []defs.APIPathReader{ - {Type: defs.APIPathReaderTypeHLSMuxer, ID: "muxer1"}, + {Type: defs.APIPathReaderTypeHLSSession, ID: "session6123"}, {Type: defs.APIPathReaderTypeWebRTCSession, ID: "session456"}, }, }, diff --git a/internal/auth/error.go b/internal/auth/error.go index 37922bdb..ce7bafd7 100644 --- a/internal/auth/error.go +++ b/internal/auth/error.go @@ -7,6 +7,6 @@ type Error struct { } // Error implements the error interface. -func (e Error) Error() string { +func (e *Error) Error() string { return "authentication failed: " + e.Wrapped.Error() } diff --git a/internal/core/api_test.go b/internal/core/api_test.go index 3db5ff08..998a12ed 100644 --- a/internal/core/api_test.go +++ b/internal/core/api_test.go @@ -14,10 +14,13 @@ import ( "testing" "time" + "github.com/bluenviron/gohlslib/v2" "github.com/bluenviron/gortmplib" rtmpcodecs "github.com/bluenviron/gortmplib/pkg/codecs" "github.com/bluenviron/gortsplib/v5" "github.com/bluenviron/gortsplib/v5/pkg/description" + "github.com/bluenviron/gortsplib/v5/pkg/format" + "github.com/bluenviron/gortsplib/v5/pkg/format/rtph264" "github.com/bluenviron/mediacommon/v2/pkg/formats/mpegts" tscodecs "github.com/bluenviron/mediacommon/v2/pkg/formats/mpegts/codecs" srt "github.com/datarhei/gosrt" @@ -369,7 +372,8 @@ func TestAPIProtocolListGet(t *testing.T) { "rtsps sessions", "rtmp", "rtmps", - "hls", + "hls sessions", + "hls muxers", "webrtc", "srt", } { @@ -470,7 +474,7 @@ func TestAPIProtocolListGet(t *testing.T) { time.Sleep(500 * time.Millisecond) - case "hls": + case "hls sessions", "hls muxers": source := gortsplib.Client{} err = source.StartRecording("rtsp://localhost:8554/mypath", &description.Session{Medias: []*description.Media{medi}}) @@ -612,7 +616,10 @@ func TestAPIProtocolListGet(t *testing.T) { case "rtmps": pa = "rtmpsconns" - case "hls": + case "hls sessions": + pa = "hlssessions" + + case "hls muxers": pa = "hlsmuxers" case "webrtc": @@ -792,7 +799,24 @@ func TestAPIProtocolListGet(t *testing.T) { }, }, out1) - case "hls": + case "hls sessions": + require.Equal(t, map[string]any{ + "itemCount": float64(1), + "pageCount": float64(1), + "items": []any{ + map[string]any{ + "id": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["id"], + "created": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["created"], + "remoteAddr": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["remoteAddr"], + "path": "mypath", + "query": "", + "user": "", + "outboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"], + }, + }, + }, out1) + + case "hls muxers": require.Equal(t, map[string]any{ "itemCount": float64(1), "pageCount": float64(1), @@ -919,7 +943,7 @@ func TestAPIProtocolListGet(t *testing.T) { var out2 any - if ca == "hls" { + if ca == "hls muxers" { httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/"+pa+"/get/"+ out1.(map[string]any)["items"].([]any)[0].(map[string]any)["path"].(string), nil, &out2) @@ -979,7 +1003,7 @@ func TestAPIProtocolListGet(t *testing.T) { out2.(map[string]any)["rtcpPacketsReceived"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtcpPacketsReceived"] out2.(map[string]any)["rtcpPacketsSent"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtcpPacketsSent"] - case "hls": + case "hls muxers": out2.(map[string]any)["lastRequest"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["lastRequest"] } @@ -1004,7 +1028,8 @@ func TestAPIProtocolGetNotFound(t *testing.T) { "rtsps sessions", "rtmp", "rtmps", - "hls", + "hls sessions", + "hls muxers", "webrtc", "srt", } { @@ -1055,7 +1080,10 @@ func TestAPIProtocolGetNotFound(t *testing.T) { case "rtmps": pa = "rtmpsconns" - case "hls": + case "hls sessions": + pa = "hlssessions" + + case "hls muxers": pa = "hlsmuxers" case "webrtc": @@ -1081,10 +1109,10 @@ func TestAPIProtocolGetNotFound(t *testing.T) { case "rtsp conns", "rtsps conns", "rtmp", "rtmps", "srt": checkError(t, "connection not found", res.Body) - case "rtsp sessions", "rtsps sessions", "webrtc": + case "rtsp sessions", "rtsps sessions", "hls sessions", "webrtc": checkError(t, "session not found", res.Body) - case "hls": + case "hls muxers": checkError(t, "muxer not found", res.Body) } }() @@ -1105,6 +1133,7 @@ func TestAPIProtocolKick(t *testing.T) { "rtsp", "rtsps", "rtmp", + "hls", "webrtc", "srt", } { @@ -1134,7 +1163,6 @@ func TestAPIProtocolKick(t *testing.T) { switch ca { case "rtsp": source := gortsplib.Client{} - err = source.StartRecording("rtsp://localhost:8554/mypath", &description.Session{Medias: []*description.Media{medi}}) require.NoError(t, err) @@ -1144,7 +1172,6 @@ func TestAPIProtocolKick(t *testing.T) { source := gortsplib.Client{ TLSConfig: &tls.Config{InsecureSkipVerify: true}, } - err = source.StartRecording("rtsps://localhost:8322/mypath", &description.Session{Medias: []*description.Media{medi}}) require.NoError(t, err) @@ -1180,6 +1207,45 @@ func TestAPIProtocolKick(t *testing.T) { err = w.WriteH264(track, 2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}}) require.NoError(t, err) + case "hls": + source := gortsplib.Client{} + err = source.StartRecording("rtsp://localhost:8554/mypath", + &description.Session{Medias: []*description.Media{medi}}) + require.NoError(t, err) + defer source.Close() + + var enc *rtph264.Encoder + enc, err = medi.Formats[0].(*format.H264).CreateEncoder() + require.NoError(t, err) + + tracksReceived := make(chan struct{}) + + client := &gohlslib.Client{ + URI: "http://localhost:8888/mypath/index.m3u8", + OnTracks: func(_ []*gohlslib.Track) error { + close(tracksReceived) + return nil + }, + } + err = client.Start() + require.NoError(t, err) + defer client.Close() + + time.Sleep(500 * time.Millisecond) + + for i := range 2 { + var pkts []*rtp.Packet + pkts, err = enc.Encode([][]byte{{5, 2, 3, 4}}) + require.NoError(t, err) + + pkts[0].Timestamp = uint32(i * 90000) + + err = source.WritePacketRTP(medi, pkts[0]) + require.NoError(t, err) + } + + <-tracksReceived + case "webrtc": var u *url.URL u, err = url.Parse("http://localhost:8889/mypath/whip") @@ -1243,6 +1309,9 @@ func TestAPIProtocolKick(t *testing.T) { case "rtmp": pa = "rtmpconns" + case "hls": + pa = "hlssessions" + case "webrtc": pa = "webrtcsessions" @@ -1256,6 +1325,7 @@ func TestAPIProtocolKick(t *testing.T) { } `json:"items"` } httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/"+pa+"/list", nil, &out1) + require.NotEmpty(t, out1.Items) httpRequest(t, hc, http.MethodPost, "http://localhost:9997/v3/"+pa+"/kick/"+out1.Items[0].ID, nil, nil) @@ -1283,6 +1353,7 @@ func TestAPIProtocolKickNotFound(t *testing.T) { "rtsp", "rtsps", "rtmp", + "hls", "webrtc", "srt", } { @@ -1318,6 +1389,9 @@ func TestAPIProtocolKickNotFound(t *testing.T) { case "rtmp": pa = "rtmpconns" + case "hls": + pa = "hlssessions" + case "webrtc": pa = "webrtcsessions" @@ -1341,11 +1415,8 @@ func TestAPIProtocolKickNotFound(t *testing.T) { case "rtsp conns", "rtsps conns", "rtmp", "rtmps", "srt": checkError(t, "connection not found", res.Body) - case "rtsp sessions", "rtsps sessions", "webrtc": + case "rtsp sessions", "rtsps sessions", "hls", "webrtc": checkError(t, "session not found", res.Body) - - case "hls": - checkError(t, "muxer not found", res.Body) } }() }) diff --git a/internal/core/core.go b/internal/core/core.go index d4810d6a..bfc8f837 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -609,6 +609,7 @@ func (p *Core) createResources(initial bool) error { ReadTimeout: p.conf.ReadTimeout, WriteTimeout: p.conf.WriteTimeout, MuxerCloseAfter: p.conf.HLSMuxerCloseAfter, + ExternalCmdPool: p.externalCmdPool, Metrics: p.metrics, PathManager: p.pathManager, Parent: p, diff --git a/internal/core/metrics_test.go b/internal/core/metrics_test.go index ee15af82..1c49565e 100644 --- a/internal/core/metrics_test.go +++ b/internal/core/metrics_test.go @@ -86,6 +86,10 @@ paths_bytes_received 0 paths_bytes_sent 0 paths_readers 0 +# HLS sessions +hls_sessions 0 +hls_sessions_outbound_bytes 0 + # HLS muxers hls_muxers 0 hls_muxers_outbound_bytes 0 diff --git a/internal/core/path.go b/internal/core/path.go index 8d9ae7e9..3a17d424 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -506,7 +506,7 @@ func (pa *path) doDescribe(req defs.PathDescribeReq) { return } - req.Res <- defs.PathDescribeRes{Err: defs.PathNoStreamAvailableError{PathName: pa.name}} + req.Res <- defs.PathDescribeRes{Err: &defs.PathNoStreamAvailableError{PathName: pa.name}} } func (pa *path) doRemovePublisher(req defs.PathRemovePublisherReq) { @@ -598,7 +598,7 @@ func (pa *path) doAddReader(req defs.PathAddReaderReq) { return } - req.Res <- defs.PathAddReaderRes{Err: defs.PathNoStreamAvailableError{PathName: pa.name}} + req.Res <- defs.PathAddReaderRes{Err: &defs.PathNoStreamAvailableError{PathName: pa.name}} } func (pa *path) doRemoveReader(req defs.PathRemoveReaderReq) { @@ -699,11 +699,13 @@ func (pa *path) doAPIPathsGet(req pathAPIPathsGetReq) { return pa.stream.OutboundBytes() }(), Readers: func() []defs.APIPathReader { - ret := make([]defs.APIPathReader, len(pa.readers)) - i := 0 + ret := make([]defs.APIPathReader, 0, len(pa.readers)) + for r := range pa.readers { - ret[i] = *r.APIReaderDescribe() - i++ + desc := *r.APIReaderDescribe() + if desc.Type != defs.APIPathReaderTypeHidden { + ret = append(ret, desc) + } } sort.Slice(ret, func(i, j int) bool { diff --git a/internal/defs/api_hls.go b/internal/defs/api_hls.go index be12489a..241be1a0 100644 --- a/internal/defs/api_hls.go +++ b/internal/defs/api_hls.go @@ -1,13 +1,45 @@ package defs -import "time" +import ( + "time" + + "github.com/google/uuid" +) // APIHLSServer contains methods used by the API and Metrics server. type APIHLSServer interface { + APISessionsList() (*APIHLSSessionList, error) + APISessionsGet(uuid.UUID) (*APIHLSSession, error) + APISessionsKick(uuid.UUID) error APIMuxersList() (*APIHLSMuxerList, error) APIMuxersGet(string) (*APIHLSMuxer, error) } +// APIHLSSessionList is a list of HLS sessions. +type APIHLSSessionList struct { + ItemCount int `json:"itemCount"` + PageCount int `json:"pageCount"` + Items []APIHLSSession `json:"items"` +} + +// APIHLSSession is an HLS session. +type APIHLSSession struct { + ID uuid.UUID `json:"id"` + Created time.Time `json:"created"` + RemoteAddr string `json:"remoteAddr"` + Path string `json:"path"` + Query string `json:"query"` + User string `json:"user"` + OutboundBytes uint64 `json:"outboundBytes"` +} + +// APIHLSMuxerList is a list of HLS muxers. +type APIHLSMuxerList struct { + ItemCount int `json:"itemCount"` + PageCount int `json:"pageCount"` + Items []APIHLSMuxer `json:"items"` +} + // APIHLSMuxer is an HLS muxer. type APIHLSMuxer struct { Path string `json:"path"` @@ -18,10 +50,3 @@ type APIHLSMuxer struct { // deprecated BytesSent uint64 `json:"bytesSent" deprecated:"true"` } - -// APIHLSMuxerList is a list of HLS muxers. -type APIHLSMuxerList struct { - ItemCount int `json:"itemCount"` - PageCount int `json:"pageCount"` - Items []APIHLSMuxer `json:"items"` -} diff --git a/internal/defs/api_path.go b/internal/defs/api_path.go index 830766c1..c3e0ccb0 100644 --- a/internal/defs/api_path.go +++ b/internal/defs/api_path.go @@ -43,16 +43,16 @@ type APIPathReaderType string // reader types. const ( - APIPathReaderTypeHLSMuxer APIPathReaderType = "hlsMuxer" - APIPathReaderTypeRTMPConn APIPathReaderType = "rtmpConn" - APIPathReaderTypeRTMPSConn APIPathReaderType = "rtmpsConn" - APIPathReaderTypeRTSPConn APIPathReaderType = "rtspConn" - APIPathReaderTypeRPICameraSecondary APIPathReaderType = "rpiCameraSecondary" - APIPathReaderTypeRTSPSession APIPathReaderType = "rtspSession" - APIPathReaderTypeRTSPSConn APIPathReaderType = "rtspsConn" - APIPathReaderTypeRTSPSSession APIPathReaderType = "rtspsSession" - APIPathReaderTypeSRTConn APIPathReaderType = "srtConn" - APIPathReaderTypeWebRTCSession APIPathReaderType = "webRTCSession" + APIPathReaderTypeHLSSession APIPathReaderType = "hlsSession" + APIPathReaderTypeRTMPConn APIPathReaderType = "rtmpConn" + APIPathReaderTypeRTMPSConn APIPathReaderType = "rtmpsConn" + APIPathReaderTypeRTSPConn APIPathReaderType = "rtspConn" + APIPathReaderTypeRTSPSession APIPathReaderType = "rtspSession" + APIPathReaderTypeRTSPSConn APIPathReaderType = "rtspsConn" + APIPathReaderTypeRTSPSSession APIPathReaderType = "rtspsSession" + APIPathReaderTypeSRTConn APIPathReaderType = "srtConn" + APIPathReaderTypeWebRTCSession APIPathReaderType = "webRTCSession" + APIPathReaderTypeHidden APIPathReaderType = "hidden" ) // APIPathReader is a reader. diff --git a/internal/defs/path.go b/internal/defs/path.go index 605a8ee0..343cdd46 100644 --- a/internal/defs/path.go +++ b/internal/defs/path.go @@ -16,7 +16,7 @@ type PathNoStreamAvailableError struct { } // Error implements the error interface. -func (e PathNoStreamAvailableError) Error() string { +func (e *PathNoStreamAvailableError) Error() string { return fmt.Sprintf("no stream is available on path '%s'", e.PathName) } diff --git a/internal/linters/go2api/go2api_test.go b/internal/linters/go2api/go2api_test.go index 7e8ee180..e6b161ec 100644 --- a/internal/linters/go2api/go2api_test.go +++ b/internal/linters/go2api/go2api_test.go @@ -175,8 +175,7 @@ func goEnumToApi(rt reflect.Type) (openAPISchema, bool) { case reflect.TypeOf(defs.APIPathReaderType("")): return openAPISchema{Type: "string", Enum: []string{ - "hlsMuxer", - "rpiCameraSecondary", + "hlsSession", "rtmpConn", "rtmpsConn", "rtspConn", @@ -352,6 +351,14 @@ func TestGo2API(t *testing.T) { "HLSMuxerList", defs.APIHLSMuxerList{}, }, + { + "HLSSession", + defs.APIHLSSession{}, + }, + { + "HLSSessionList", + defs.APIHLSSessionList{}, + }, { "Info", defs.APIInfo{}, diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 214f85da..ac63cde5 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -74,6 +74,7 @@ type metricsType string const ( metricsTypePaths metricsType = "paths" + metricsTypeHLSSessions metricsType = "hls_sessions" metricsTypeHLSMuxers metricsType = "hls_muxers" metricsTypeRTSPConns metricsType = "rtsp_conns" metricsTypeRTSPSessions metricsType = "rtsp_sessions" @@ -227,6 +228,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { typ := metricsType(ctx.Query("type")) pathFilter := ctx.Query("path") hlsMuxerFilter := ctx.Query("hls_muxer") + hlsSessionFilter := ctx.Query("hls_session") rtspConnFilter := ctx.Query("rtsp_conn") rtspSessionFilter := ctx.Query("rtsp_session") rtspsConnFilter := ctx.Query("rtsps_conn") @@ -238,6 +240,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { anyFilterActive := pathFilter != "" || hlsMuxerFilter != "" || + hlsSessionFilter != "" || rtspConnFilter != "" || rtspSessionFilter != "" || rtspsConnFilter != "" || @@ -336,6 +339,32 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { } if !interfaceIsEmpty(hlsServer) { + if (typ == "" || typ == metricsTypeHLSSessions) && (!anyFilterActive || hlsSessionFilter != "") { + var data *defs.APIHLSSessionList + data, err := hlsServer.APISessionsList() + if err == nil && len(data.Items) != 0 { + out.WriteString("# HLS sessions\n") + for _, i := range data.Items { + if hlsSessionFilter == "" || hlsSessionFilter == i.ID.String() { + ta := tags(map[string]string{ + "id": i.ID.String(), + "path": i.Path, + "remoteAddr": i.RemoteAddr, + }) + + metric(&out, "hls_sessions", ta, 1) + metric(&out, "hls_sessions_outbound_bytes", ta, int64(i.OutboundBytes)) + } + } + out.WriteString("\n") + } else if hlsSessionFilter == "" { + out.WriteString("# HLS sessions\n") + metric(&out, "hls_sessions", "", 0) + metric(&out, "hls_sessions_outbound_bytes", "", 0) + out.WriteString("\n") + } + } + if (typ == "" || typ == metricsTypeHLSMuxers) && (!anyFilterActive || hlsMuxerFilter != "") { var data *defs.APIHLSMuxerList data, err := hlsServer.APIMuxersList() diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index 376f6640..92ce23eb 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -89,6 +89,28 @@ func (dummyHLSServer) APIMuxersGet(string) (*defs.APIHLSMuxer, error) { panic("unused") } +func (dummyHLSServer) APISessionsList() (*defs.APIHLSSessionList, error) { + return &defs.APIHLSSessionList{ + ItemCount: 1, + PageCount: 1, + Items: []defs.APIHLSSession{{ + ID: uuid.MustParse("18294761-f9d1-4ea9-9a35-fe265b62eb41"), + Created: time.Date(2003, 11, 4, 23, 15, 7, 0, time.UTC), + RemoteAddr: "124.5.5.5:34542", + Path: "mypath", + OutboundBytes: 187, + }}, + }, nil +} + +func (dummyHLSServer) APISessionsGet(uuid.UUID) (*defs.APIHLSSession, error) { + panic("unused") +} + +func (dummyHLSServer) APISessionsKick(uuid.UUID) error { + panic("unused") +} + type dummyRTSPServer struct{} func (dummyRTSPServer) APIConnsList() (*defs.APIRTSPConnsList, error) { @@ -332,6 +354,18 @@ func (emptyHLSServer) APIMuxersGet(string) (*defs.APIHLSMuxer, error) { panic("unused") } +func (emptyHLSServer) APISessionsList() (*defs.APIHLSSessionList, error) { + return &defs.APIHLSSessionList{}, nil +} + +func (emptyHLSServer) APISessionsGet(uuid.UUID) (*defs.APIHLSSession, error) { + panic("unused") +} + +func (emptyHLSServer) APISessionsKick(uuid.UUID) error { + panic("unused") +} + type emptyRTSPServer struct{} func (emptyRTSPServer) APIConnsList() (*defs.APIRTSPConnsList, error) { @@ -478,6 +512,12 @@ func TestMetrics(t *testing.T) { "paths_bytes_received{name=\"mypath\",state=\"ready\"} 123\n"+ "paths_bytes_sent{name=\"mypath\",state=\"ready\"} 456\n"+ "\n"+ + "# HLS sessions\n"+ + "hls_sessions{id=\"18294761-f9d1-4ea9-9a35-fe265b62eb41\",path=\"mypath\","+ + "remoteAddr=\"124.5.5.5:34542\"} 1\n"+ + "hls_sessions_outbound_bytes{id=\"18294761-f9d1-4ea9-9a35-fe265b62eb41\",path=\"mypath\","+ + "remoteAddr=\"124.5.5.5:34542\"} 187\n"+ + "\n"+ "# HLS muxers\n"+ "hls_muxers{name=\"mypath\"} 1\n"+ "hls_muxers_outbound_bytes{name=\"mypath\"} 789\n"+ @@ -835,6 +875,10 @@ func TestZeroMetricsFallback(t *testing.T) { "paths_bytes_sent 0\n"+ "paths_readers 0\n"+ "\n"+ + "# HLS sessions\n"+ + "hls_sessions 0\n"+ + `hls_sessions_outbound_bytes 0`+"\n"+ + "\n"+ "# HLS muxers\n"+ "hls_muxers 0\n"+ "hls_muxers_outbound_bytes 0\n"+ @@ -965,6 +1009,7 @@ func TestFilter(t *testing.T) { for _, ca := range []string{ "path", "hls_muxer", + "hls_session", "rtsp_conn", "rtsp_session", "rtsps_conn", @@ -1007,6 +1052,8 @@ func TestFilter(t *testing.T) { u += "?path=mypath" case "hls_muxer": u += "?hls_muxer=mypath" + case "hls_session": + u += "?hls_session=18294761-f9d1-4ea9-9a35-fe265b62eb41" case "rtsp_conn": u += "?rtsp_conn=18294761-f9d1-4ea9-9a35-fe265b62eb41" case "rtsp_session": @@ -1059,6 +1106,16 @@ func TestFilter(t *testing.T) { `hls_muxers_bytes_sent{name="mypath"} 789`+"\n\n", string(byts)) + case "hls_session": + require.Equal(t, + "# HLS sessions\n"+ + `hls_sessions{id="18294761-f9d1-4ea9-9a35-fe265b62eb41",path="mypath",`+ + `remoteAddr="124.5.5.5:34542"} 1`+"\n"+ + `hls_sessions_outbound_bytes{id="18294761-f9d1-4ea9-9a35-fe265b62eb41",path="mypath",`+ + `remoteAddr="124.5.5.5:34542"} 187`+"\n"+ + "\n", + string(byts)) + case "rtsp_conn": require.Equal(t, "# RTSP connections\n"+ diff --git a/internal/protocols/httpp/handler_origin.go b/internal/protocols/httpp/handler_origin.go index f7f6a8e1..1f7bbc1c 100644 --- a/internal/protocols/httpp/handler_origin.go +++ b/internal/protocols/httpp/handler_origin.go @@ -15,7 +15,10 @@ func isOriginAllowed(origin string, allowOrigins []string) (string, bool) { for _, o := range allowOrigins { if o == "*" { - return o, true + if origin != "" { + return origin, true + } + return "*", true } } diff --git a/internal/protocols/httpp/handler_origin_test.go b/internal/protocols/httpp/handler_origin_test.go index 1f07878d..4e6379f0 100644 --- a/internal/protocols/httpp/handler_origin_test.go +++ b/internal/protocols/httpp/handler_origin_test.go @@ -38,7 +38,7 @@ func TestHandlerOrigin(t *testing.T) { "everything allowed, with origin", "https://example.com", []string{"*"}, - "*", + "https://example.com", }, { "allowed", diff --git a/internal/recorder/recorder_instance.go b/internal/recorder/recorder_instance.go index 2c44ec70..d2ffea2b 100644 --- a/internal/recorder/recorder_instance.go +++ b/internal/recorder/recorder_instance.go @@ -47,8 +47,8 @@ func (ri *recorderInstance) initialize() { ri.format, ) ri.reader = &stream.Reader{ - SkipBytesSent: true, - Parent: ri, + SkipOutboundBytes: true, + Parent: ri, } ri.terminate = make(chan struct{}) diff --git a/internal/servers/hls/http_server.go b/internal/servers/hls/http_server.go index 546ca224..a2ab17c1 100644 --- a/internal/servers/hls/http_server.go +++ b/internal/servers/hls/http_server.go @@ -35,6 +35,12 @@ func mergePathAndQuery(path string, rawQuery string) string { return res } +func isIOS(userAgent string) bool { + return strings.Contains(userAgent, "iPad") || + strings.Contains(userAgent, "iPhone") || + strings.Contains(userAgent, "iPod") +} + type httpServer struct { address string dumpPackets bool @@ -124,6 +130,17 @@ func (s *httpServer) onRequest(ctx *gin.Context) { var dir string var fname string + type contentType int + + const ( + index contentType = iota + multivariantPlaylist + mediaPlaylist + segment + ) + + var contentTyp contentType + switch { case strings.HasSuffix(pa, "/hls.min.js"): ctx.Header("Cache-Control", "max-age=3600") @@ -135,8 +152,16 @@ func (s *httpServer) onRequest(ctx *gin.Context) { case pa == "", pa == "favicon.ico", strings.HasSuffix(pa, "/hls.min.js.map"): return - case strings.HasSuffix(pa, ".m3u8") || - strings.HasSuffix(pa, ".ts") || + case strings.HasSuffix(pa, ".m3u8"): + dir, fname = gopath.Dir(pa), gopath.Base(pa) + + if fname == "index.m3u8" { + contentTyp = multivariantPlaylist + } else { + contentTyp = mediaPlaylist + } + + case strings.HasSuffix(pa, ".ts") || strings.HasSuffix(pa, ".mp4") || strings.HasSuffix(pa, ".mp"): dir, fname = gopath.Dir(pa), gopath.Base(pa) @@ -145,42 +170,169 @@ func (s *httpServer) onRequest(ctx *gin.Context) { fname += "4" } + contentTyp = segment + default: - dir, fname = pa, "" + dir = pa if !strings.HasSuffix(dir, "/") { ctx.Header("Location", mergePathAndQuery(ctx.Request.URL.Path+"/", ctx.Request.URL.RawQuery)) ctx.Writer.WriteHeader(http.StatusMovedPermanently) return } + + dir = dir[:len(dir)-1] + contentTyp = index } - dir = strings.TrimSuffix(dir, "/") - if dir == "" { - return - } + switch contentTyp { + case index: + _, err := s.pathManager.FindPathConf(defs.PathFindPathConfReq{ + AccessRequest: defs.PathAccessRequest{ + Name: dir, + Query: ctx.Request.URL.RawQuery, + Publish: false, + Proto: auth.ProtocolHLS, + Credentials: httpp.Credentials(ctx.Request), + IP: net.ParseIP(ctx.ClientIP()), + }, + }) + if err != nil { + var terr *auth.Error + if errors.As(err, &terr) { + if terr.AskCredentials { + ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`) + s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error")) + return + } + + s.Log(logger.Info, "connection %v failed to authenticate: %v", httpp.RemoteAddr(ctx), terr.Wrapped) + + // wait some seconds to delay brute force attacks + <-time.After(auth.PauseAfterError) - res, err := s.pathManager.FindPathConf(defs.PathFindPathConfReq{ - AccessRequest: defs.PathAccessRequest{ - Name: dir, - Query: ctx.Request.URL.RawQuery, - Publish: false, - Proto: auth.ProtocolHLS, - Credentials: httpp.Credentials(ctx.Request), - IP: net.ParseIP(ctx.ClientIP()), - }, - }) - if err != nil { - var terr *auth.Error - if errors.As(err, &terr) { - if terr.AskCredentials { - ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`) s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error")) return } - s.Log(logger.Info, "connection %v failed to authenticate: %v", httpp.RemoteAddr(ctx), terr.Wrapped) + s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) + return + } + ctx.Header("Cache-Control", "max-age=3600") + ctx.Header("Content-Type", "text/html") + ctx.Writer.WriteHeader(http.StatusOK) + ctx.Writer.Write(hlsIndex) + + case multivariantPlaylist: + if ctx.Request.URL.Query().Get("cookieCheck") != "1" { + http.SetCookie(ctx.Writer, &http.Cookie{ + Name: "cookieCheck", + Value: "1", + }) + + http.SetCookie(ctx.Writer, &http.Cookie{ + Name: "cookieCheck", + Value: "1", + SameSite: http.SameSiteNoneMode, + Secure: true, + Partitioned: true, + HttpOnly: true, + }) + + q := ctx.Request.URL.Query() + q.Set("cookieCheck", "1") + ctx.Request.URL.RawQuery = q.Encode() + ctx.Writer.Header().Set("Location", mergePathAndQuery(ctx.Request.URL.Path, ctx.Request.URL.RawQuery)) + + ctx.Writer.WriteHeader(http.StatusFound) + return + } + + if _, err := ctx.Request.Cookie("cookieCheck"); err != nil && isIOS(ctx.Request.UserAgent()) { + s.writeErrorNoLog(ctx, http.StatusBadRequest, fmt.Errorf("HLS on iOS requires the server to set and read cookies")) + return + } + + q := ctx.Request.URL.Query() + q.Del("cookieCheck") + ctx.Request.URL.RawQuery = q.Encode() + + sx := &session{ + remoteAddr: httpp.RemoteAddr(ctx), + pathName: dir, + externalCmdPool: s.parent.ExternalCmdPool, + pathManager: s.pathManager, + server: s.parent, + } + err := sx.initialize(ctx) + if err != nil { + var terr *auth.Error + if errors.As(err, &terr) { + if terr.AskCredentials { + ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`) + s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error")) + return + } + + s.Log(logger.Info, "connection %v failed to authenticate: %v", httpp.RemoteAddr(ctx), terr.Wrapped) + + // wait some seconds to delay brute force attacks + <-time.After(auth.PauseAfterError) + + s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error")) + return + } + + var terr2 *defs.PathNoStreamAvailableError + if errors.As(err, &terr2) { + s.writeErrorNoLog(ctx, http.StatusNotFound, err) + return + } + + s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) + return + } + + if cookie, err2 := ctx.Request.Cookie("cookieCheck"); err2 == nil && cookie.Value == "1" { + http.SetCookie(ctx.Writer, &http.Cookie{ + Name: sessionCookieName, + Value: sx.secret.String(), + }) + + http.SetCookie(ctx.Writer, &http.Cookie{ + Name: sessionCookieName, + Value: sx.secret.String(), + SameSite: http.SameSiteNoneMode, + Secure: true, + Partitioned: true, + HttpOnly: true, + }) + } else { + q = ctx.Request.URL.Query() + q.Set(sessionQueryParamName, sx.secret.String()) + ctx.Request.URL.RawQuery = q.Encode() + } + + ctx.Writer = &responseWriterCounter{ + ResponseWriter: ctx.Writer, + bytesSent: &sx.bytesSent, + } + + ctx.Request.URL.Path = fname + + err = sx.muxer.handleRequest(ctx) + if err != nil { + s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) + return + } + + default: + muxer, err := s.parent.getMuxer(serverGetMuxerReq{ + path: dir, + create: false, + }) + if err != nil { // wait some seconds to delay brute force attacks <-time.After(auth.PauseAfterError) @@ -188,31 +340,26 @@ func (s *httpServer) onRequest(ctx *gin.Context) { return } - s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) - return - } + sx := muxer.findSession(ctx) + if sx == nil { + // wait some seconds to delay brute force attacks + <-time.After(auth.PauseAfterError) - switch fname { - case "": - ctx.Header("Cache-Control", "max-age=3600") - ctx.Header("Content-Type", "text/html") - ctx.Writer.WriteHeader(http.StatusOK) - ctx.Writer.Write(hlsIndex) - - default: - var mux *muxer - mux, err = s.parent.getMuxer(serverGetMuxerReq{ - path: dir, - remoteAddr: httpp.RemoteAddr(ctx), - query: ctx.Request.URL.RawQuery, - sourceOnDemand: res.Conf.SourceOnDemand, - }) - if err != nil { - ctx.Writer.WriteHeader(http.StatusNotFound) + s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error")) return } + ctx.Writer = &responseWriterCounter{ + ResponseWriter: ctx.Writer, + bytesSent: &sx.bytesSent, + } + ctx.Request.URL.Path = fname - mux.handleRequest(ctx) + + err = muxer.handleRequest(ctx) + if err != nil { + s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) + return + } } } diff --git a/internal/servers/hls/muxer.go b/internal/servers/hls/muxer.go index 5c5b2e90..cb85f2b0 100644 --- a/internal/servers/hls/muxer.go +++ b/internal/servers/hls/muxer.go @@ -4,22 +4,22 @@ import ( "context" "errors" "fmt" - "net/http" "sync" "sync/atomic" "time" + "github.com/bluenviron/gortsplib/v5/pkg/format" "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/hls" "github.com/bluenviron/mediamtx/internal/stream" "github.com/gin-gonic/gin" + "github.com/google/uuid" ) const ( - closeCheckPeriod = 1 * time.Second - recreatePause = 10 * time.Second + recreateInstancePause = 10 * time.Second ) func emptyTimer() *time.Timer { @@ -28,20 +28,9 @@ func emptyTimer() *time.Timer { return t } -type responseWriterWithCounter struct { - http.ResponseWriter - bytesSent *atomic.Uint64 -} - -func (w *responseWriterWithCounter) Write(p []byte) (int, error) { - n, err := w.ResponseWriter.Write(p) - w.bytesSent.Add(uint64(n)) - return n, err -} - -type muxerGetInstanceRes struct { - instance *muxerInstance - cumulatedOutboundFramesDiscarded uint64 +type muxerCloseInstanceReq struct { + instance *muxerInstance + err error } type muxer struct { @@ -67,9 +56,12 @@ type muxer struct { lastRequestTime atomic.Int64 bytesSent atomic.Uint64 - instanceMutex sync.RWMutex + mutex sync.RWMutex instance *muxerInstance cumulatedOutboundFramesDiscarded uint64 + sessionsBySecret map[uuid.UUID]*session + + chCloseInstance chan muxerCloseInstanceReq } func (m *muxer) initialize() { @@ -79,7 +71,8 @@ func (m *muxer) initialize() { m.ctxCancel = ctxCancel m.created = time.Now() m.lastRequestTime.Store(time.Now().UnixNano()) - m.bytesSent.Store(0) + m.sessionsBySecret = make(map[uuid.UUID]*session) + m.chCloseInstance = make(chan muxerCloseInstanceReq) m.Log(logger.Info, "created %s", func() string { if m.remoteAddr == "" { @@ -89,7 +82,7 @@ func (m *muxer) initialize() { }()) // block first request to getInstance() until the first instance is available - m.instanceMutex.Lock() + m.mutex.Lock() m.wg.Add(1) go m.run() @@ -116,9 +109,23 @@ func (m *muxer) run() { m.ctxCancel() - m.parent.closeMuxer(m) + if m.instance != nil { + m.instance.close() + } + + m.mutex.Lock() + + m.instance = nil + + for _, sx := range m.sessionsBySecret { + sx.close2(fmt.Errorf("muxer destroyed")) + } + + m.mutex.Unlock() m.Log(logger.Info, "destroyed: %v", err) + + m.parent.closeMuxer(m) } func (m *muxer) runInner() error { @@ -131,7 +138,7 @@ func (m *muxer) runInner() error { }, }) if err != nil { - m.instanceMutex.Unlock() + m.mutex.Unlock() return err } @@ -142,71 +149,101 @@ func (m *muxer) runInner() error { tmp, err := m.createInstance(res.Stream) if err != nil { if m.remoteAddr != "" || errors.Is(err, hls.ErrNoSupportedCodecs) { - m.instanceMutex.Unlock() + m.mutex.Unlock() return err } - m.Log(logger.Error, err.Error()) + m.Log(logger.Error, "muxer instance crashed: %v", err) } m.instance = tmp - m.instanceMutex.Unlock() + m.mutex.Unlock() - defer func() { - if m.instance != nil { - m.closeInstance() - } - }() - - var instanceError chan error - var recreateTimer *time.Timer + var recreateInstanceTimer *time.Timer if m.instance != nil { - instanceError = m.instance.errorChan() - recreateTimer = emptyTimer() + recreateInstanceTimer = emptyTimer() } else { - instanceError = make(chan error) - recreateTimer = time.NewTimer(recreatePause) + recreateInstanceTimer = time.NewTimer(recreateInstancePause) } + defer func() { + recreateInstanceTimer.Stop() + }() + + sessionCleanupTicker := time.NewTicker(sessionCleanupPeriod) + defer sessionCleanupTicker.Stop() + var activityCheckTimer *time.Timer if m.remoteAddr != "" { - activityCheckTimer = time.NewTimer(closeCheckPeriod) + activityCheckTimer = time.NewTimer(max(time.Duration(m.closeAfter)/3, 1*time.Second)) } else { activityCheckTimer = emptyTimer() } + defer func() { + activityCheckTimer.Stop() + }() + for { select { - case err = <-instanceError: - if m.remoteAddr != "" { - return err + case req := <-m.chCloseInstance: + if m.instance != req.instance { + continue } - m.Log(logger.Error, err.Error()) - m.closeInstance() - instanceError = make(chan error) - recreateTimer = time.NewTimer(recreatePause) + m.mutex.Lock() + m.cumulatedOutboundFramesDiscarded += m.instance.reader.OutboundFramesDiscarded() + m.instance = nil + m.mutex.Unlock() - case <-recreateTimer.C: + if m.remoteAddr != "" { + return req.err + } else { + m.mutex.Lock() + for _, sx := range m.sessionsBySecret { + sx.close2(fmt.Errorf("muxer instance crashed")) + } + m.sessionsBySecret = make(map[uuid.UUID]*session) + m.mutex.Unlock() + + m.Log(logger.Error, "muxer instance crashed: %v", req.err) + } + + recreateInstanceTimer = time.NewTimer(recreateInstancePause) + + case <-recreateInstanceTimer.C: tmp, err = m.createInstance(res.Stream) if err != nil { - m.Log(logger.Error, err.Error()) - recreateTimer = time.NewTimer(recreatePause) - } else { - m.instanceMutex.Lock() - m.instance = tmp - m.instanceMutex.Unlock() - - instanceError = m.instance.errorChan() + m.Log(logger.Error, "muxer instance crashed: %v", err) + recreateInstanceTimer = time.NewTimer(recreateInstancePause) + continue } + m.mutex.Lock() + m.instance = tmp + m.mutex.Unlock() + + case <-sessionCleanupTicker.C: + now := time.Now() + + m.mutex.Lock() + for secret, sx := range m.sessionsBySecret { + lastRequest := time.Unix(0, sx.lastRequestTime.Load()) + + if now.Sub(lastRequest) >= sessionCloseAfter { + delete(m.sessionsBySecret, secret) + sx.close2(fmt.Errorf("inactive")) + } + } + m.mutex.Unlock() + case <-activityCheckTimer.C: t := time.Unix(0, m.lastRequestTime.Load()) if time.Since(t) >= time.Duration(m.closeAfter) { return fmt.Errorf("not used anymore") } - activityCheckTimer = time.NewTimer(closeCheckPeriod) + activityCheckTimer = time.NewTimer(max(time.Duration(m.closeAfter)/3, 1*time.Second)) case <-m.ctx.Done(): return errors.New("terminated") @@ -214,16 +251,6 @@ func (m *muxer) runInner() error { } } -func (m *muxer) closeInstance() { - m.instanceMutex.Lock() - m.cumulatedOutboundFramesDiscarded += m.instance.reader.OutboundFramesDiscarded() - var tmp *muxerInstance - tmp, m.instance = m.instance, nil - m.instanceMutex.Unlock() - - tmp.close() -} - func (m *muxer) createInstance(strm *stream.Stream) (*muxerInstance, error) { mi := &muxerInstance{ variant: m.variant, @@ -233,54 +260,107 @@ func (m *muxer) createInstance(strm *stream.Stream) (*muxerInstance, error) { segmentMaxSize: m.segmentMaxSize, directory: m.directory, pathName: m.pathName, + bytesSent: &m.bytesSent, + wg: m.wg, stream: strm, + server: m.parent, parent: m, } err := mi.initialize() - return mi, err + if err != nil { + return nil, err + } + return mi, nil } -func (m *muxer) getInstance() muxerGetInstanceRes { - m.instanceMutex.RLock() - defer m.instanceMutex.RUnlock() - - return muxerGetInstanceRes{ - instance: m.instance, - cumulatedOutboundFramesDiscarded: m.cumulatedOutboundFramesDiscarded, +func (m *muxer) closeInstance(mi *muxerInstance, err error) { + select { + case m.chCloseInstance <- muxerCloseInstanceReq{instance: mi, err: err}: + case <-m.ctx.Done(): } } // APIReaderDescribe implements reader. func (m *muxer) APIReaderDescribe() *defs.APIPathReader { return &defs.APIPathReader{ - Type: defs.APIPathReaderTypeHLSMuxer, + Type: defs.APIPathReaderTypeHidden, ID: "", } } -func (m *muxer) handleRequest(ctx *gin.Context) { +func (m *muxer) addSession(sx *session) ([]format.Format, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + select { + case <-m.ctx.Done(): + return nil, fmt.Errorf("terminated") + default: + } + + if m.instance == nil { + return nil, fmt.Errorf("muxer instance not available") + } + + m.sessionsBySecret[sx.secret] = sx + return m.instance.reader.Formats(), nil +} + +func (m *muxer) findSession(ctx *gin.Context) *session { + var rawSecret string + if cookie, err := ctx.Request.Cookie(sessionCookieName); err == nil { + rawSecret = cookie.Value + } else { + q := ctx.Request.URL.Query() + rawSecret = q.Get(sessionQueryParamName) + } + + secret, err := uuid.Parse(rawSecret) + if err != nil { + return nil + } + + m.mutex.RLock() + defer m.mutex.RUnlock() + + sx, ok := m.sessionsBySecret[secret] + if !ok { + return nil + } + + if ctx.ClientIP() != sx.ip { + return nil + } + + sx.lastRequestTime.Store(time.Now().UnixNano()) + + return sx +} + +func (m *muxer) handleRequest(ctx *gin.Context) error { m.lastRequestTime.Store(time.Now().UnixNano()) - res := m.getInstance() - if res.instance == nil { - ctx.Writer.WriteHeader(http.StatusNotFound) - return + m.mutex.RLock() + instance := m.instance + m.mutex.RUnlock() + + if instance == nil { + return fmt.Errorf("muxer instance not available") } - w := &responseWriterWithCounter{ - ResponseWriter: ctx.Writer, - bytesSent: &m.bytesSent, - } - - res.instance.handleRequest(w, ctx.Request) + instance.handleRequest(ctx) + return nil } func (m *muxer) apiItem() *defs.APIHLSMuxer { - res := m.getInstance() + m.mutex.RLock() + instance := m.instance + cumulatedOutboundFramesDiscarded := m.cumulatedOutboundFramesDiscarded + m.mutex.RUnlock() - outboundFramesDiscarded := res.cumulatedOutboundFramesDiscarded - if res.instance != nil { - outboundFramesDiscarded += res.instance.reader.OutboundFramesDiscarded() + outboundFramesDiscarded := cumulatedOutboundFramesDiscarded + if instance != nil { + outboundFramesDiscarded += instance.reader.OutboundFramesDiscarded() } return &defs.APIHLSMuxer{ @@ -292,3 +372,52 @@ func (m *muxer) apiItem() *defs.APIHLSMuxer { BytesSent: m.bytesSent.Load(), } } + +func (m *muxer) apiSessionsList() []defs.APIHLSSession { + m.mutex.RLock() + defer m.mutex.RUnlock() + + sessions := make([]defs.APIHLSSession, 0, len(m.sessionsBySecret)) + + for _, sx := range m.sessionsBySecret { + sessions = append(sessions, *sx.apiItem()) + } + + return sessions +} + +func (m *muxer) findSessionByUUID(uuid uuid.UUID) *session { + for _, sx := range m.sessionsBySecret { + if sx.uuid == uuid { + return sx + } + } + return nil +} + +func (m *muxer) apiSessionsGet(uuid uuid.UUID) (*defs.APIHLSSession, bool) { + m.mutex.RLock() + sx := m.findSessionByUUID(uuid) + if sx == nil { + m.mutex.RUnlock() + return nil, false + } + m.mutex.RUnlock() + + return sx.apiItem(), true +} + +func (m *muxer) apiSessionsKick(uuid uuid.UUID) bool { + m.mutex.Lock() + defer m.mutex.Unlock() + + sx := m.findSessionByUUID(uuid) + if sx == nil { + return false + } + + sx.close2(fmt.Errorf("kicked")) + delete(m.sessionsBySecret, sx.secret) + + return true +} diff --git a/internal/servers/hls/muxer_instance.go b/internal/servers/hls/muxer_instance.go index 9af7d5c1..bf291ded 100644 --- a/internal/servers/hls/muxer_instance.go +++ b/internal/servers/hls/muxer_instance.go @@ -1,9 +1,12 @@ package hls import ( - "net/http" + "context" + "fmt" "os" "path/filepath" + "sync" + "sync/atomic" "time" "github.com/bluenviron/gohlslib/v2" @@ -12,8 +15,21 @@ import ( "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/hls" "github.com/bluenviron/mediamtx/internal/stream" + "github.com/gin-gonic/gin" ) +const ( + sessionCookieName = "hlsSession" + sessionQueryParamName = "session" + sessionCloseAfter = 30 * time.Second + sessionCleanupPeriod = sessionCloseAfter / 3 +) + +type instanceParent interface { + logger.Writer + closeInstance(*muxerInstance, error) +} + type muxerInstance struct { variant conf.HLSVariant segmentCount int @@ -22,14 +38,21 @@ type muxerInstance struct { segmentMaxSize conf.StringSize directory string pathName string + bytesSent *atomic.Uint64 + wg *sync.WaitGroup stream *stream.Stream - parent logger.Writer + server logger.Writer + parent instanceParent - hmuxer *gohlslib.Muxer - reader *stream.Reader + ctx context.Context + ctxCancel func() + hmuxer *gohlslib.Muxer + reader *stream.Reader } func (mi *muxerInstance) initialize() error { + mi.Log(logger.Debug, "instance created") + var muxerDirectory string if mi.directory != "" { muxerDirectory = filepath.Join(mi.directory, mi.pathName) @@ -49,8 +72,8 @@ func (mi *muxerInstance) initialize() error { } mi.reader = &stream.Reader{ - SkipBytesSent: true, - Parent: mi, + SkipOutboundBytes: true, + Parent: mi, } err := hls.FromStream(mi.stream.Desc, mi.reader, mi.hmuxer) @@ -68,6 +91,11 @@ func (mi *muxerInstance) initialize() error { mi.stream.AddReader(mi.reader) + mi.ctx, mi.ctxCancel = context.WithCancel(context.Background()) + + mi.wg.Add(1) + go mi.run() + return nil } @@ -77,17 +105,50 @@ func (mi *muxerInstance) Log(level logger.Level, format string, args ...any) { } func (mi *muxerInstance) close() { + mi.ctxCancel() +} + +func (mi *muxerInstance) run() { + defer mi.wg.Done() + + err := mi.runInner() + + mi.ctxCancel() + mi.stream.RemoveReader(mi.reader) + mi.hmuxer.Close() + if mi.hmuxer.Directory != "" { os.Remove(mi.hmuxer.Directory) } + + mi.Log(logger.Debug, "instance destroyed: %v", err) + + mi.parent.closeInstance(mi, err) } -func (mi *muxerInstance) errorChan() chan error { - return mi.reader.Error() +func (mi *muxerInstance) runInner() error { + for { + select { + case <-mi.ctx.Done(): + return fmt.Errorf("terminated") + + case err := <-mi.reader.Error(): + return err + } + } } -func (mi *muxerInstance) handleRequest(w http.ResponseWriter, r *http.Request) { - mi.hmuxer.Handle(w, r) +func (mi *muxerInstance) handleRequest(ctx *gin.Context) { + w := ctx.Writer + + w = &responseWriterNoCache{ResponseWriter: w} + + w = &responseWriterCounter{ + ResponseWriter: w, + bytesSent: mi.bytesSent, + } + + mi.hmuxer.Handle(w, ctx.Request) } diff --git a/internal/servers/hls/response_writer_counter.go b/internal/servers/hls/response_writer_counter.go new file mode 100644 index 00000000..485d2bb1 --- /dev/null +++ b/internal/servers/hls/response_writer_counter.go @@ -0,0 +1,18 @@ +package hls + +import ( + "sync/atomic" + + "github.com/gin-gonic/gin" +) + +type responseWriterCounter struct { + gin.ResponseWriter + bytesSent *atomic.Uint64 +} + +func (w *responseWriterCounter) Write(p []byte) (int, error) { + n, err := w.ResponseWriter.Write(p) + w.bytesSent.Add(uint64(n)) + return n, err +} diff --git a/internal/servers/hls/response_writer_no_cache.go b/internal/servers/hls/response_writer_no_cache.go new file mode 100644 index 00000000..12208043 --- /dev/null +++ b/internal/servers/hls/response_writer_no_cache.go @@ -0,0 +1,19 @@ +package hls + +import ( + "net/http" + + "github.com/gin-gonic/gin" +) + +type responseWriterNoCache struct { + gin.ResponseWriter +} + +func (w *responseWriterNoCache) WriteHeader(statusCode int) { + if statusCode == http.StatusOK { + w.ResponseWriter.Header().Set("Cache-Control", "private, no-cache") + } + + w.ResponseWriter.WriteHeader(statusCode) +} diff --git a/internal/servers/hls/server.go b/internal/servers/hls/server.go index aca58c6f..c845b0e8 100644 --- a/internal/servers/hls/server.go +++ b/internal/servers/hls/server.go @@ -11,12 +11,17 @@ import ( "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" + "github.com/bluenviron/mediamtx/internal/externalcmd" "github.com/bluenviron/mediamtx/internal/logger" + "github.com/google/uuid" ) // ErrMuxerNotFound is returned when a muxer is not found. var ErrMuxerNotFound = errors.New("muxer not found") +// ErrSessionNotFound is returned when a session is not found. +var ErrSessionNotFound = errors.New("session not found") + func interfaceIsEmpty(i any) bool { return reflect.ValueOf(i).Kind() != reflect.Pointer || reflect.ValueOf(i).IsNil() } @@ -28,9 +33,10 @@ type serverGetMuxerRes struct { type serverGetMuxerReq struct { path string - remoteAddr string - query string - sourceOnDemand bool + create bool + remoteAddr string // only if create == true + query string // only if create == true + sourceOnDemand bool // only if create == true res chan serverGetMuxerRes } @@ -53,6 +59,34 @@ type serverAPIMuxersGetReq struct { res chan serverAPIMuxersGetRes } +type serverAPISessionsListRes struct { + data *defs.APIHLSSessionList + err error +} + +type serverAPISessionsListReq struct { + res chan serverAPISessionsListRes +} + +type serverAPISessionsGetRes struct { + data *defs.APIHLSSession + err error +} + +type serverAPISessionsGetReq struct { + uuid uuid.UUID + res chan serverAPISessionsGetRes +} + +type serverAPISessionsKickRes struct { + err error +} + +type serverAPISessionsKickReq struct { + uuid uuid.UUID + res chan serverAPISessionsKickRes +} + type serverMetrics interface { SetHLSServer(defs.APIHLSServer) } @@ -86,6 +120,7 @@ type Server struct { ReadTimeout conf.Duration WriteTimeout conf.Duration MuxerCloseAfter conf.Duration + ExternalCmdPool *externalcmd.Pool Metrics serverMetrics PathManager serverPathManager Parent serverParent @@ -97,12 +132,15 @@ type Server struct { muxers map[string]*muxer // in - chPathReady chan defs.Path - chPathNotReady chan defs.Path - chGetMuxer chan serverGetMuxerReq - chCloseMuxer chan *muxer - chAPIMuxerList chan serverAPIMuxersListReq - chAPIMuxerGet chan serverAPIMuxersGetReq + chPathReady chan defs.Path + chPathNotReady chan defs.Path + chGetMuxer chan serverGetMuxerReq + chCloseMuxer chan *muxer + chAPIMuxerList chan serverAPIMuxersListReq + chAPIMuxerGet chan serverAPIMuxersGetReq + chAPISessionsList chan serverAPISessionsListReq + chAPISessionsGet chan serverAPISessionsGetReq + chAPISessionsKick chan serverAPISessionsKickReq } // Initialize initializes the server. @@ -118,6 +156,9 @@ func (s *Server) Initialize() error { s.chCloseMuxer = make(chan *muxer) s.chAPIMuxerList = make(chan serverAPIMuxersListReq) s.chAPIMuxerGet = make(chan serverAPIMuxersGetReq) + s.chAPISessionsList = make(chan serverAPISessionsListReq) + s.chAPISessionsGet = make(chan serverAPISessionsGetReq) + s.chAPISessionsKick = make(chan serverAPISessionsKickReq) s.httpServer = &httpServer{ address: s.Address, @@ -211,6 +252,8 @@ outer: switch { case ok: req.res <- serverGetMuxerRes{muxer: mux} + case !req.create: + req.res <- serverGetMuxerRes{err: fmt.Errorf("muxer not found")} case s.AlwaysRemux && !req.sourceOnDemand: req.res <- serverGetMuxerRes{err: fmt.Errorf("muxer is waiting to be created")} default: @@ -248,6 +291,43 @@ outer: req.res <- serverAPIMuxersGetRes{data: muxer.apiItem()} + case req := <-s.chAPISessionsList: + data := &defs.APIHLSSessionList{ + Items: []defs.APIHLSSession{}, + } + + for _, muxer := range s.muxers { + data.Items = append(data.Items, muxer.apiSessionsList()...) + } + + sort.Slice(data.Items, func(i, j int) bool { + return data.Items[i].Created.Before(data.Items[j].Created) + }) + + req.res <- serverAPISessionsListRes{data: data} + + case req := <-s.chAPISessionsGet: + for _, muxer := range s.muxers { + session, ok := muxer.apiSessionsGet(req.uuid) + if ok { + req.res <- serverAPISessionsGetRes{data: session} + continue outer + } + } + + req.res <- serverAPISessionsGetRes{err: ErrSessionNotFound} + + case req := <-s.chAPISessionsKick: + for _, muxer := range s.muxers { + ok := muxer.apiSessionsKick(req.uuid) + if ok { + req.res <- serverAPISessionsKickRes{} + continue outer + } + } + + req.res <- serverAPISessionsKickRes{err: ErrSessionNotFound} + case <-s.ctx.Done(): break outer } @@ -349,3 +429,53 @@ func (s *Server) APIMuxersGet(name string) (*defs.APIHLSMuxer, error) { return nil, fmt.Errorf("terminated") } } + +// APISessionsList implements defs.APIHLSServer. +func (s *Server) APISessionsList() (*defs.APIHLSSessionList, error) { + req := serverAPISessionsListReq{ + res: make(chan serverAPISessionsListRes), + } + + select { + case s.chAPISessionsList <- req: + res := <-req.res + return res.data, res.err + + case <-s.ctx.Done(): + return nil, fmt.Errorf("terminated") + } +} + +// APISessionsGet implements defs.APIHLSServer. +func (s *Server) APISessionsGet(uuid uuid.UUID) (*defs.APIHLSSession, error) { + req := serverAPISessionsGetReq{ + uuid: uuid, + res: make(chan serverAPISessionsGetRes), + } + + select { + case s.chAPISessionsGet <- req: + res := <-req.res + return res.data, res.err + + case <-s.ctx.Done(): + return nil, fmt.Errorf("terminated") + } +} + +// APISessionsKick implements defs.APIHLSServer. +func (s *Server) APISessionsKick(uuid uuid.UUID) error { + req := serverAPISessionsKickReq{ + uuid: uuid, + res: make(chan serverAPISessionsKickRes), + } + + select { + case s.chAPISessionsKick <- req: + res := <-req.res + return res.err + + case <-s.ctx.Done(): + return fmt.Errorf("terminated") + } +} diff --git a/internal/servers/hls/server_test.go b/internal/servers/hls/server_test.go index 841461aa..20f64dc9 100644 --- a/internal/servers/hls/server_test.go +++ b/internal/servers/hls/server_test.go @@ -13,6 +13,7 @@ import ( "github.com/bluenviron/gohlslib/v2" "github.com/bluenviron/gohlslib/v2/pkg/codecs" "github.com/bluenviron/gortsplib/v5/pkg/description" + "github.com/bluenviron/gortsplib/v5/pkg/format" "github.com/bluenviron/mediacommon/v2/pkg/codecs/mpeg4audio" "github.com/bluenviron/mediamtx/internal/auth" "github.com/bluenviron/mediamtx/internal/conf" @@ -158,7 +159,7 @@ func TestServerNotFound(t *testing.T) { }, addReaderImpl: func(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { require.Equal(t, "nonexisting", req.AccessRequest.Name) - return nil, fmt.Errorf("not found") + return nil, &defs.PathNoStreamAvailableError{} }, } @@ -253,11 +254,22 @@ func TestServerRead(t *testing.T) { }, addReaderImpl: func(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { require.Equal(t, "teststream", req.AccessRequest.Name) - if ca == "always remux off" { + + switch req.Author.(type) { + case (*muxer): + if ca == "always remux off" { + require.Equal(t, "param=value", req.AccessRequest.Query) + } else { + require.Equal(t, "", req.AccessRequest.Query) + } + + case *session: require.Equal(t, "param=value", req.AccessRequest.Query) - } else { - require.Equal(t, "", req.AccessRequest.Query) + + default: + t.Errorf("should not happen") } + return &defs.PathAddReaderRes{Path: &dummyPath{}, Stream: strm}, nil }, } @@ -579,7 +591,7 @@ func TestAuthError(t *testing.T) { ReadTimeout: conf.Duration(10 * time.Second), WriteTimeout: conf.Duration(10 * time.Second), PathManager: &dummyPathManager{ - findPathConfImpl: func(req defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { + addReaderImpl: func(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { if req.AccessRequest.Credentials.User == "" && req.AccessRequest.Credentials.Pass == "" { return nil, &auth.Error{AskCredentials: true, Wrapped: fmt.Errorf("auth error")} } @@ -625,3 +637,133 @@ func TestAuthError(t *testing.T) { require.Equal(t, 2, n) } + +func TestAuthQueryPreservedAcrossRedirect(t *testing.T) { + s := &Server{ + Address: "127.0.0.1:8888", + Encryption: false, + ServerKey: "", + ServerCert: "", + AlwaysRemux: true, + Variant: conf.HLSVariant(gohlslib.MuxerVariantMPEGTS), + SegmentCount: 7, + SegmentDuration: conf.Duration(1 * time.Second), + PartDuration: conf.Duration(200 * time.Millisecond), + SegmentMaxSize: 50 * 1024 * 1024, + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + PathManager: &dummyPathManager{ + findPathConfImpl: func(_ defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { + return nil, &auth.Error{AskCredentials: true, Wrapped: fmt.Errorf("auth error")} + }, + }, + Parent: test.NilLogger, + } + err := s.Initialize() + require.NoError(t, err) + defer s.Close() + + client := &http.Client{ + CheckRedirect: func(_ *http.Request, _ []*http.Request) error { + return http.ErrUseLastResponse + }, + } + + res, err := client.Get("http://127.0.0.1:8888/stream/index.m3u8?jwt=mytoken") + require.NoError(t, err) + defer res.Body.Close() + + require.Equal(t, http.StatusFound, res.StatusCode) + require.Equal(t, "/stream/index.m3u8?cookieCheck=1&jwt=mytoken", res.Header.Get("Location")) +} + +func TestServerNoSupportedCodecs(t *testing.T) { + for _, ca := range []string{ + "always remux off", + "always remux on", + } { + t.Run(ca, func(t *testing.T) { + desc := &description.Session{Medias: []*description.Media{{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.VP8{}}, + }}} + + strm := &stream.Stream{ + Desc: desc, + WriteQueueSize: 512, + RTPMaxPayloadSize: 1450, + Parent: test.NilLogger, + } + err := strm.Initialize() + require.NoError(t, err) + + pm := &dummyPathManager{ + findPathConfImpl: func(req defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { + require.Equal(t, "teststream", req.AccessRequest.Name) + return &defs.PathFindPathConfRes{Conf: &conf.Path{}, User: req.AccessRequest.Credentials.User}, nil + }, + addReaderImpl: func(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { + require.Equal(t, "teststream", req.AccessRequest.Name) + return &defs.PathAddReaderRes{Path: &dummyPath{}, Stream: strm}, nil + }, + } + + s := &Server{ + Address: "127.0.0.1:8888", + AlwaysRemux: (ca == "always remux on"), + Variant: conf.HLSVariant(gohlslib.MuxerVariantMPEGTS), + SegmentCount: 7, + SegmentDuration: conf.Duration(1 * time.Second), + PartDuration: conf.Duration(200 * time.Millisecond), + SegmentMaxSize: 50 * 1024 * 1024, + TrustedProxies: conf.IPNetworks{}, + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + PathManager: pm, + Parent: test.NilLogger, + } + err = s.Initialize() + require.NoError(t, err) + defer s.Close() + + client := &http.Client{ + CheckRedirect: func(_ *http.Request, _ []*http.Request) error { + return http.ErrUseLastResponse + }, + } + + res, err := client.Get("http://127.0.0.1:8888/teststream/index.m3u8") + require.NoError(t, err) + defer res.Body.Close() + + require.Equal(t, http.StatusFound, res.StatusCode) + require.Equal(t, "/teststream/index.m3u8?cookieCheck=1", res.Header.Get("Location")) + + res, err = client.Get("http://127.0.0.1:8888/teststream/index.m3u8?cookieCheck=1") + require.NoError(t, err) + defer res.Body.Close() + + require.Equal(t, http.StatusInternalServerError, res.StatusCode) + require.Contains(t, res.Header.Get("Content-Type"), "application/json") + + byts, err := io.ReadAll(res.Body) + require.NoError(t, err) + + var payload defs.APIError + err = json.Unmarshal(byts, &payload) + require.NoError(t, err) + + if ca == "always remux off" { + require.Equal(t, defs.APIError{ + Status: defs.APIErrorStatusError, + Error: "terminated", + }, payload) + } else { + require.Equal(t, defs.APIError{ + Status: defs.APIErrorStatusError, + Error: "muxer is waiting to be created", + }, payload) + } + }) + } +} diff --git a/internal/servers/hls/session.go b/internal/servers/hls/session.go new file mode 100644 index 00000000..accf6544 --- /dev/null +++ b/internal/servers/hls/session.go @@ -0,0 +1,168 @@ +package hls + +import ( + "encoding/hex" + "net" + "slices" + "sync/atomic" + "time" + + "github.com/bluenviron/mediamtx/internal/auth" + "github.com/bluenviron/mediamtx/internal/defs" + "github.com/bluenviron/mediamtx/internal/externalcmd" + "github.com/bluenviron/mediamtx/internal/hooks" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/protocols/httpp" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/unit" + "github.com/gin-gonic/gin" + "github.com/google/uuid" +) + +type sessionServer interface { + logger.Writer + getMuxer(serverGetMuxerReq) (*muxer, error) +} + +type session struct { + remoteAddr string + pathName string + externalCmdPool *externalcmd.Pool + pathManager serverPathManager + server sessionServer + + uuid uuid.UUID + secret uuid.UUID + ip string + created time.Time + query string + user string + lastRequestTime atomic.Int64 + bytesSent atomic.Uint64 + path defs.Path + stream *stream.Stream + muxer *muxer + reader *stream.Reader + onUnreadHook func() +} + +func (s *session) initialize(ctx *gin.Context) error { + s.uuid = uuid.New() + s.secret = uuid.New() + s.ip, _, _ = net.SplitHostPort(s.remoteAddr) + s.created = time.Now() + s.query = ctx.Request.URL.RawQuery + s.lastRequestTime.Store(time.Now().UnixNano()) + + res, err := s.pathManager.AddReader(defs.PathAddReaderReq{ + Author: s, + AccessRequest: defs.PathAccessRequest{ + Name: s.pathName, + Query: s.query, + Publish: false, + Proto: auth.ProtocolHLS, + ID: &s.uuid, + Credentials: httpp.Credentials(ctx.Request), + IP: net.ParseIP(ctx.ClientIP()), + }, + }) + if err != nil { + return err + } + + s.path = res.Path + s.stream = res.Stream + s.user = res.User + + muxer, err := s.server.getMuxer(serverGetMuxerReq{ + path: s.pathName, + create: true, + remoteAddr: s.remoteAddr, + query: s.query, + sourceOnDemand: res.Path.SafeConf().SourceOnDemand, + }) + if err != nil { + s.path.RemoveReader(defs.PathRemoveReaderReq{Author: s}) + return err + } + + s.muxer = muxer + + muxerFormats, err := s.muxer.addSession(s) + if err != nil { + s.path.RemoveReader(defs.PathRemoveReaderReq{Author: s}) + return err + } + + s.reader = &stream.Reader{ + Parent: s, + } + + // all of this is needed to allow Stream to increase outbound bytes for every HLS session + for _, medi := range res.Stream.Desc.Medias { + for _, forma := range medi.Formats { + if slices.Contains(muxerFormats, forma) { + s.reader.OnData(medi, forma, func(_ *unit.Unit) error { + return nil + }) + } + } + } + + res.Stream.AddReader(s.reader) + + s.Log(logger.Info, "created by %s, reading from muxer '%s'", s.remoteAddr, s.pathName) + + s.onUnreadHook = hooks.OnRead(hooks.OnReadParams{ + Logger: s, + ExternalCmdPool: s.externalCmdPool, + Conf: res.Path.SafeConf(), + ExternalCmdEnv: res.Path.ExternalCmdEnv(), + Reader: *s.APIReaderDescribe(), + Query: s.query, + }) + + return nil +} + +// called by path or path manager. +// not implemented since closing the Muxer is enough to close every associated session. +func (s *session) Close() { +} + +func (s *session) close2(err error) { + s.stream.RemoveReader(s.reader) + + s.path.RemoveReader(defs.PathRemoveReaderReq{Author: s}) + + s.onUnreadHook() + + s.Log(logger.Info, "closed: %v", err) +} + +// Log implements logger.Writer. +func (s *session) Log(level logger.Level, format string, args ...any) { + id := hex.EncodeToString(s.uuid[:4]) + s.server.Log(level, "[session %v] "+format, append([]any{id}, args...)...) +} + +func (s *session) apiItem() *defs.APIHLSSession { + outboundBytes := s.bytesSent.Load() + + return &defs.APIHLSSession{ + ID: s.uuid, + Created: s.created, + RemoteAddr: s.remoteAddr, + Path: s.pathName, + Query: s.query, + User: s.user, + OutboundBytes: outboundBytes, + } +} + +func (s *session) APIReaderDescribe() *defs.APIPathReader { + return &defs.APIPathReader{ + Type: defs.APIPathReaderTypeHLSSession, + ID: s.uuid.String(), + } +} diff --git a/internal/servers/rtsp/conn.go b/internal/servers/rtsp/conn.go index e4c48520..a189a821 100644 --- a/internal/servers/rtsp/conn.go +++ b/internal/servers/rtsp/conn.go @@ -170,7 +170,7 @@ func (c *conn) onDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx, return res, nil, err2 } - var terr2 defs.PathNoStreamAvailableError + var terr2 *defs.PathNoStreamAvailableError if errors.As(res.Err, &terr2) { return &base.Response{ StatusCode: base.StatusNotFound, diff --git a/internal/servers/rtsp/session.go b/internal/servers/rtsp/session.go index e3bd488b..976a1387 100644 --- a/internal/servers/rtsp/session.go +++ b/internal/servers/rtsp/session.go @@ -287,7 +287,7 @@ func (s *session) onSetup(c *conn, ctx *gortsplib.ServerHandlerOnSetupCtx, return res, nil, err2 } - var terr2 defs.PathNoStreamAvailableError + var terr2 *defs.PathNoStreamAvailableError if errors.As(err, &terr2) { return &base.Response{ StatusCode: base.StatusNotFound, diff --git a/internal/servers/webrtc/server_test.go b/internal/servers/webrtc/server_test.go index d32717eb..0c280ec9 100644 --- a/internal/servers/webrtc/server_test.go +++ b/internal/servers/webrtc/server_test.go @@ -710,7 +710,7 @@ func TestServerReadNotFound(t *testing.T) { return &defs.PathFindPathConfRes{Conf: &conf.Path{}, User: req.AccessRequest.Credentials.User}, nil }, AddReaderImpl: func(_ defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) { - return nil, defs.PathNoStreamAvailableError{} + return nil, &defs.PathNoStreamAvailableError{} }, } diff --git a/internal/servers/webrtc/session.go b/internal/servers/webrtc/session.go index 157d5f02..925d0395 100644 --- a/internal/servers/webrtc/session.go +++ b/internal/servers/webrtc/session.go @@ -279,21 +279,19 @@ func (s *session) runPublish() (int, error) { func (s *session) runRead() (int, error) { ip, _, _ := net.SplitHostPort(s.req.remoteAddr) - req := defs.PathAccessRequest{ - Name: s.req.pathName, - Query: s.req.httpRequest.URL.RawQuery, - Proto: auth.ProtocolWebRTC, - ID: &s.uuid, - Credentials: httpp.Credentials(s.req.httpRequest), - IP: net.ParseIP(ip), - } - res, err := s.pathManager.AddReader(defs.PathAddReaderReq{ - Author: s, - AccessRequest: req, + Author: s, + AccessRequest: defs.PathAccessRequest{ + Name: s.req.pathName, + Query: s.req.httpRequest.URL.RawQuery, + Proto: auth.ProtocolWebRTC, + ID: &s.uuid, + Credentials: httpp.Credentials(s.req.httpRequest), + IP: net.ParseIP(ip), + }, }) if err != nil { - var terr2 defs.PathNoStreamAvailableError + var terr2 *defs.PathNoStreamAvailableError if errors.As(err, &terr2) { return http.StatusNotFound, err } diff --git a/internal/staticsources/rpicamera/source.go b/internal/staticsources/rpicamera/source.go index 20e0d80a..c5b81017 100644 --- a/internal/staticsources/rpicamera/source.go +++ b/internal/staticsources/rpicamera/source.go @@ -95,7 +95,7 @@ func (r *secondaryReader) Close() { // APIReaderDescribe implements reader. func (*secondaryReader) APIReaderDescribe() *defs.APIPathReader { return &defs.APIPathReader{ - Type: defs.APIPathReaderTypeRPICameraSecondary, + Type: defs.APIPathReaderTypeHidden, ID: "", } } @@ -341,7 +341,7 @@ func (s *Source) waitForPrimary( }, }) if err != nil { - var err2 defs.PathNoStreamAvailableError + var err2 *defs.PathNoStreamAvailableError if errors.As(err, &err2) { select { case <-time.After(pauseBetweenErrors): diff --git a/internal/stream/reader.go b/internal/stream/reader.go index fe9d647c..56b7386c 100644 --- a/internal/stream/reader.go +++ b/internal/stream/reader.go @@ -16,8 +16,8 @@ type OnDataFunc func(*unit.Unit) error // Reader is a stream reader. type Reader struct { - SkipBytesSent bool - Parent logger.Writer + SkipOutboundBytes bool + Parent logger.Writer onDatas map[*description.Media]map[format.Format]OnDataFunc queueSize int diff --git a/internal/stream/stream.go b/internal/stream/stream.go index 57d2d353..e4c2527b 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -394,8 +394,8 @@ func (s *Stream) Initialize() error { alwaysAvailable: s.AlwaysAvailable, rtpMaxPayloadSize: s.RTPMaxPayloadSize, replaceNTP: s.ReplaceNTP, - addInboundBytes: s.addInboundBytes, - addOutboundBytes: s.addOutboundBytes, + inboundBytes: &s.inboundBytes, + outboundBytes: &s.outboundBytes, updateLastTime: s.updateLastTime, writeRTSP: s.writeRTSP, inboundFramesInError: s.inboundFramesInError, @@ -574,14 +574,6 @@ func (s *Stream) WaitForReaders() { <-s.hasReaders } -func (s *Stream) addInboundBytes(v uint64) { - s.inboundBytes.Add(v) -} - -func (s *Stream) addOutboundBytes(v uint64) { - s.outboundBytes.Add(v) -} - func (s *Stream) updateLastTime(pts time.Duration) { s.timeMutex.Lock() defer s.timeMutex.Unlock() diff --git a/internal/stream/stream_format.go b/internal/stream/stream_format.go index 522b7dc4..c057989b 100644 --- a/internal/stream/stream_format.go +++ b/internal/stream/stream_format.go @@ -2,6 +2,7 @@ package stream import ( "crypto/rand" + "sync/atomic" "time" "github.com/bluenviron/gortsplib/v5/pkg/description" @@ -44,8 +45,8 @@ type streamFormat struct { rtpMaxPayloadSize int replaceNTP bool inboundFramesInError *errordumper.Dumper - addInboundBytes func(uint64) - addOutboundBytes func(uint64) + inboundBytes *atomic.Uint64 + outboundBytes *atomic.Uint64 updateLastTime func(time.Duration) writeRTSP func(*description.Media, []*rtp.Packet, time.Time) parent logger.Writer diff --git a/internal/stream/stream_media.go b/internal/stream/stream_media.go index c7e4dff0..f3ccb170 100644 --- a/internal/stream/stream_media.go +++ b/internal/stream/stream_media.go @@ -1,6 +1,7 @@ package stream import ( + "sync/atomic" "time" "github.com/bluenviron/gortsplib/v5/pkg/description" @@ -15,8 +16,8 @@ type streamMedia struct { alwaysAvailable bool rtpMaxPayloadSize int replaceNTP bool - addInboundBytes func(uint64) - addOutboundBytes func(uint64) + inboundBytes *atomic.Uint64 + outboundBytes *atomic.Uint64 updateLastTime func(time.Duration) writeRTSP func(*description.Media, []*rtp.Packet, time.Time) inboundFramesInError *errordumper.Dumper @@ -36,8 +37,8 @@ func (sm *streamMedia) initialize() error { rtpMaxPayloadSize: sm.rtpMaxPayloadSize, replaceNTP: sm.replaceNTP, inboundFramesInError: sm.inboundFramesInError, - addInboundBytes: sm.addInboundBytes, - addOutboundBytes: sm.addOutboundBytes, + inboundBytes: sm.inboundBytes, + outboundBytes: sm.outboundBytes, updateLastTime: sm.updateLastTime, writeRTSP: sm.writeRTSP, parent: sm.parent, diff --git a/internal/stream/stream_standard_test.go b/internal/stream/stream_standard_test.go index 7594cd1d..0cfbb53e 100644 --- a/internal/stream/stream_standard_test.go +++ b/internal/stream/stream_standard_test.go @@ -69,7 +69,7 @@ func TestStream(t *testing.T) { require.Equal(t, uint64(14), strm.OutboundBytes()) } -func TestStreamSkipBytesSent(t *testing.T) { +func TestStreamSkipOutboundBytes(t *testing.T) { desc := &description.Session{Medias: []*description.Media{ { Type: description.MediaTypeVideo, @@ -98,7 +98,7 @@ func TestStreamSkipBytesSent(t *testing.T) { require.NoError(t, err) r := &Reader{ - SkipBytesSent: true, + SkipOutboundBytes: true, } recv := make(chan struct{}) diff --git a/internal/stream/sub_stream_format.go b/internal/stream/sub_stream_format.go index 4cd45611..2a9c3e7f 100644 --- a/internal/stream/sub_stream_format.go +++ b/internal/stream/sub_stream_format.go @@ -170,7 +170,7 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { } size := unitSize(u) - ssf.streamFormat.addInboundBytes(size) + ssf.streamFormat.inboundBytes.Add(size) ssf.streamFormat.writeRTSP(ssf.streamFormat.media, u.RTPPackets, u.NTP) @@ -178,8 +178,8 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { csr := sr cOnData := onData sr.push(func() error { - if !csr.SkipBytesSent { - ssf.streamFormat.addOutboundBytes(size) + if !csr.SkipOutboundBytes { + ssf.streamFormat.outboundBytes.Add(size) } return cOnData(u) })