OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

hls: track sessions (#962) (#5683)

Browse Source 
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.
This commit is contained in:
Alessandro Ros
2026-04-25 21:10:34 +02:00
committed by GitHub
parent c4a1ba332a
commit f87d9e659e
40 changed files with 1694 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/openapi.yaml
+152 -2
View File
@@ -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
docs/2-features/21-metrics.md
+6 -1
View File
@@ -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
docs/4-read/13-web-browsers.md
+2
View File
@@ -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
internal/api/api.go
+3
View File
@@ -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) {
internal/api/api_hls.go
+61
View File
@@ -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)
}
internal/api/api_hls_test.go
+218 -2
View File
@@ -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
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")
}
internal/api/api_paths_test.go
+1 -1
View File
@@ -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"},
},
},
internal/auth/error.go
+1 -1
View File
@@ -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()
}
internal/core/api_test.go
+87 -16
View File
@@ -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)
}
}()
})
internal/core/core.go
+1
View File
@@ -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,
internal/core/metrics_test.go
+4
View File
@@ -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
internal/core/path.go
+8 -6
View File
@@ -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 {
internal/defs/api_hls.go
+33 -8
View File
@@ -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"`
}
internal/defs/api_path.go
+2 -2
View File
@@ -43,16 +43,16 @@ type APIPathReaderType string
// reader types.
const (
APIPathReaderTypeHLSMuxer APIPathReaderType = "hlsMuxer"
APIPathReaderTypeHLSSession APIPathReaderType = "hlsSession"
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"
APIPathReaderTypeHidden APIPathReaderType = "hidden"
)
// APIPathReader is a reader.
internal/defs/path.go
+1 -1
View File
@@ -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)
}
internal/linters/go2api/go2api_test.go
+9 -2
View File
@@ -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{},
internal/metrics/metrics.go
+29
View File
@@ -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()
internal/metrics/metrics_test.go
+57
View File
@@ -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"+
internal/protocols/httpp/handler_origin.go
+4 -1
View File
@@ -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
}
}
internal/protocols/httpp/handler_origin_test.go
+1 -1
View File
@@ -38,7 +38,7 @@ func TestHandlerOrigin(t *testing.T) {
"everything allowed, with origin",
"https://example.com",
[]string{"*"},
"*",
"https://example.com",
},
{
"allowed",
internal/recorder/recorder_instance.go
+1 -1
View File
@@ -47,7 +47,7 @@ func (ri *recorderInstance) initialize() {
ri.format,
)
ri.reader = &stream.Reader{
SkipBytesSent: true,
SkipOutboundBytes: true,
Parent: ri,
}
internal/servers/hls/http_server.go
+168 -21
View File
@@ -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,22 +170,24 @@ 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
}
res, err := s.pathManager.FindPathConf(defs.PathFindPathConfReq{
switch contentTyp {
case index:
_, err := s.pathManager.FindPathConf(defs.PathFindPathConfReq{
AccessRequest: defs.PathAccessRequest{
Name: dir,
Query: ctx.Request.URL.RawQuery,
@@ -192,27 +219,147 @@ func (s *httpServer) onRequest(ctx *gin.Context) {
return
}
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,
case multivariantPlaylist:
if ctx.Request.URL.Query().Get("cookieCheck") != "1" {
http.SetCookie(ctx.Writer, &http.Cookie{
Name: "cookieCheck",
Value: "1",
})
if err != nil {
ctx.Writer.WriteHeader(http.StatusNotFound)
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
mux.handleRequest(ctx)
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)
s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error"))
return
}
sx := muxer.findSession(ctx)
if sx == nil {
// wait some seconds to delay brute force attacks
<-time.After(auth.PauseAfterError)
s.writeErrorNoLog(ctx, http.StatusUnauthorized, fmt.Errorf("authentication error"))
return
}
ctx.Writer = &responseWriterCounter{
ResponseWriter: ctx.Writer,
bytesSent: &sx.bytesSent,
}
ctx.Request.URL.Path = fname
err = muxer.handleRequest(ctx)
if err != nil {
s.writeErrorNoLog(ctx, http.StatusInternalServerError, err)
return
}
}
}
internal/servers/hls/muxer.go
+218 -89
View File
@@ -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 {
type muxerCloseInstanceReq struct {
instance *muxerInstance
cumulatedOutboundFramesDiscarded uint64
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()
var recreateInstanceTimer *time.Timer
if m.instance != nil {
recreateInstanceTimer = emptyTimer()
} else {
recreateInstanceTimer = time.NewTimer(recreateInstancePause)
}
defer func() {
if m.instance != nil {
m.closeInstance()
}
recreateInstanceTimer.Stop()
}()
var instanceError chan error
var recreateTimer *time.Timer
if m.instance != nil {
instanceError = m.instance.errorChan()
recreateTimer = emptyTimer()
} else {
instanceError = make(chan error)
recreateTimer = time.NewTimer(recreatePause)
}
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
}
internal/servers/hls/muxer_instance.go
+68 -7
View File
@@ -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
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,7 +72,7 @@ func (mi *muxerInstance) initialize() error {
}
mi.reader = &stream.Reader{
SkipBytesSent: true,
SkipOutboundBytes: true,
Parent: mi,
}
@@ -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)
}
internal/servers/hls/response_writer_counter.go
+18
View File
@@ -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
}
internal/servers/hls/response_writer_no_cache.go
+19
View File
@@ -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)
}
internal/servers/hls/server.go
+133 -3
View File
@@ -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
@@ -103,6 +138,9 @@ type Server struct {
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")
}
}
internal/servers/hls/server_test.go
+144 -2
View File
@@ -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)
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)
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)
}
})
}
}
internal/servers/hls/session.go
+168
View File
@@ -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(),
}
}
internal/servers/rtsp/conn.go
+1 -1
View File
@@ -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,
internal/servers/rtsp/session.go
+1 -1
View File
@@ -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,
internal/servers/webrtc/server_test.go
+1 -1
View File
@@ -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{}
},
}
internal/servers/webrtc/session.go
+5 -7
View File
@@ -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{
res, err := s.pathManager.AddReader(defs.PathAddReaderReq{
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),
}
res, err := s.pathManager.AddReader(defs.PathAddReaderReq{
Author: s,
AccessRequest: req,
},
})
if err != nil {
var terr2 defs.PathNoStreamAvailableError
var terr2 *defs.PathNoStreamAvailableError
if errors.As(err, &terr2) {
return http.StatusNotFound, err
}
internal/staticsources/rpicamera/source.go
+2 -2
View File
@@ -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):
internal/stream/reader.go
+1 -1
View File
@@ -16,7 +16,7 @@ type OnDataFunc func(*unit.Unit) error
// Reader is a stream reader.
type Reader struct {
SkipBytesSent bool
SkipOutboundBytes bool
Parent logger.Writer
onDatas map[*description.Media]map[format.Format]OnDataFunc
internal/stream/stream.go
+2 -10
View File
@@ -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()
internal/stream/stream_format.go
+3 -2
View File
@@ -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
internal/stream/stream_media.go
+5 -4
View File
@@ -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,
internal/stream/stream_standard_test.go
+2 -2
View File
@@ -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{})
internal/stream/sub_stream_format.go
+3 -3
View File
@@ -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)
})