CEF/mediamtx
1
0
Fork 0
forked from External/mediamtx
Code Pull requests Activity

hls: decrease number of idle connections kept open after a reconnection (#3128)

Browse source
This commit is contained in:
Alessandro Ros 2024-03-10 11:51:15 +01:00 • committed by GitHub
parent 1989997fe1
commit 24cc62e344
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 123 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

60
internal/api/api_test.go
View file

@ -136,7 +136,9 @@ func TestConfigGlobalGet(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer api.Close() defer api.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var out map[string]interface{} var out map[string]interface{}
httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/config/global/get", nil, &out) httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/config/global/get", nil, &out)
@ -157,7 +159,9 @@ func TestConfigGlobalPatch(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer api.Close() defer api.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
httpRequest(t, hc, http.MethodPatch, "http://myuser:mypass@localhost:9997/v3/config/global/patch", httpRequest(t, hc, http.MethodPatch, "http://myuser:mypass@localhost:9997/v3/config/global/patch",
map[string]interface{}{ map[string]interface{}{
@ -198,7 +202,9 @@ func TestAPIConfigGlobalPatchUnknownField(t *testing.T) { //nolint:dupl
byts, err := json.Marshal(b) byts, err := json.Marshal(b)
require.NoError(t, err) require.NoError(t, err)
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
req, err := http.NewRequest(http.MethodPatch, "http://myuser:mypass@localhost:9997/v3/config/global/patch", req, err := http.NewRequest(http.MethodPatch, "http://myuser:mypass@localhost:9997/v3/config/global/patch",
bytes.NewReader(byts)) bytes.NewReader(byts))
@ -226,7 +232,9 @@ func TestAPIConfigPathDefaultsGet(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer api.Close() defer api.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var out map[string]interface{} var out map[string]interface{}
httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/config/pathdefaults/get", nil, &out) httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/config/pathdefaults/get", nil, &out)
@ -247,7 +255,9 @@ func TestAPIConfigPathDefaultsPatch(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer api.Close() defer api.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
httpRequest(t, hc, http.MethodPatch, "http://myuser:mypass@localhost:9997/v3/config/pathdefaults/patch", httpRequest(t, hc, http.MethodPatch, "http://myuser:mypass@localhost:9997/v3/config/pathdefaults/patch",
map[string]interface{}{ map[string]interface{}{
@ -292,7 +302,9 @@ func TestAPIConfigPathsList(t *testing.T) {
Items []pathConfig `json:"items"` Items []pathConfig `json:"items"`
} }
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var out listRes var out listRes
httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/config/paths/list", nil, &out) httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/config/paths/list", nil, &out)
@ -324,7 +336,9 @@ func TestAPIConfigPathsGet(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer api.Close() defer api.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var out map[string]interface{} var out map[string]interface{}
httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/config/paths/get/my/path", nil, &out) httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/config/paths/get/my/path", nil, &out)
@ -346,7 +360,9 @@ func TestAPIConfigPathsAdd(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer api.Close() defer api.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
httpRequest(t, hc, http.MethodPost, "http://myuser:mypass@localhost:9997/v3/config/paths/add/my/path", httpRequest(t, hc, http.MethodPost, "http://myuser:mypass@localhost:9997/v3/config/paths/add/my/path",
map[string]interface{}{ map[string]interface{}{
@ -385,7 +401,9 @@ func TestAPIConfigPathsAddUnknownField(t *testing.T) { //nolint:dupl
byts, err := json.Marshal(b) byts, err := json.Marshal(b)
require.NoError(t, err) require.NoError(t, err)
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
req, err := http.NewRequest(http.MethodPost, req, err := http.NewRequest(http.MethodPost,
"http://myuser:mypass@localhost:9997/v3/config/paths/add/my/path", bytes.NewReader(byts)) "http://myuser:mypass@localhost:9997/v3/config/paths/add/my/path", bytes.NewReader(byts))
@ -413,7 +431,9 @@ func TestAPIConfigPathsPatch(t *testing.T) { //nolint:dupl
require.NoError(t, err) require.NoError(t, err)
defer api.Close() defer api.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
httpRequest(t, hc, http.MethodPost, "http://myuser:mypass@localhost:9997/v3/config/paths/add/my/path", httpRequest(t, hc, http.MethodPost, "http://myuser:mypass@localhost:9997/v3/config/paths/add/my/path",
map[string]interface{}{ map[string]interface{}{
@ -451,7 +471,9 @@ func TestAPIConfigPathsReplace(t *testing.T) { //nolint:dupl
require.NoError(t, err) require.NoError(t, err)
defer api.Close() defer api.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
httpRequest(t, hc, http.MethodPost, "http://myuser:mypass@localhost:9997/v3/config/paths/add/my/path", httpRequest(t, hc, http.MethodPost, "http://myuser:mypass@localhost:9997/v3/config/paths/add/my/path",
map[string]interface{}{ map[string]interface{}{
@ -489,7 +511,9 @@ func TestAPIConfigPathsDelete(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer api.Close() defer api.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
httpRequest(t, hc, http.MethodPost, "http://myuser:mypass@localhost:9997/v3/config/paths/add/my/path", httpRequest(t, hc, http.MethodPost, "http://myuser:mypass@localhost:9997/v3/config/paths/add/my/path",
map[string]interface{}{ map[string]interface{}{
@ -546,7 +570,9 @@ func TestRecordingsList(t *testing.T) {
err = os.WriteFile(filepath.Join(dir, "mypath2", "2009-11-07_11-22-00-900000.mp4"), []byte(""), 0o644) err = os.WriteFile(filepath.Join(dir, "mypath2", "2009-11-07_11-22-00-900000.mp4"), []byte(""), 0o644)
require.NoError(t, err) require.NoError(t, err)
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var out interface{} var out interface{}
httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/recordings/list", nil, &out) httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/recordings/list", nil, &out)
@ -607,7 +633,9 @@ func TestRecordingsGet(t *testing.T) {
err = os.WriteFile(filepath.Join(dir, "mypath1", "2009-11-07_11-22-00-900000.mp4"), []byte(""), 0o644) err = os.WriteFile(filepath.Join(dir, "mypath1", "2009-11-07_11-22-00-900000.mp4"), []byte(""), 0o644)
require.NoError(t, err) require.NoError(t, err)
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var out interface{} var out interface{}
httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/recordings/get/mypath1", nil, &out) httpRequest(t, hc, http.MethodGet, "http://myuser:mypass@localhost:9997/v3/recordings/get/mypath1", nil, &out)
@ -651,7 +679,9 @@ func TestRecordingsDeleteSegment(t *testing.T) {
err = os.WriteFile(filepath.Join(dir, "mypath1", "2008-11-07_11-22-00-900000.mp4"), []byte(""), 0o644) err = os.WriteFile(filepath.Join(dir, "mypath1", "2008-11-07_11-22-00-900000.mp4"), []byte(""), 0o644)
require.NoError(t, err) require.NoError(t, err)
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
u, err := url.Parse("http://myuser:mypass@localhost:9997/v3/recordings/deletesegment") u, err := url.Parse("http://myuser:mypass@localhost:9997/v3/recordings/deletesegment")
require.NoError(t, err) require.NoError(t, err)

40
internal/core/api_test.go
View file

@ -97,7 +97,9 @@ func TestAPIPathsList(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
media0 := test.UniqueMediaH264() media0 := test.UniqueMediaH264()
@ -155,7 +157,9 @@ func TestAPIPathsList(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
source := gortsplib.Client{TLSConfig: &tls.Config{InsecureSkipVerify: true}} source := gortsplib.Client{TLSConfig: &tls.Config{InsecureSkipVerify: true}}
err = source.StartRecording("rtsps://localhost:8322/mypath", err = source.StartRecording("rtsps://localhost:8322/mypath",
@ -191,7 +195,9 @@ func TestAPIPathsList(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var out pathList var out pathList
httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/paths/list", nil, &out) httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/paths/list", nil, &out)
@ -218,7 +224,9 @@ func TestAPIPathsList(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var out pathList var out pathList
httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/paths/list", nil, &out) httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/paths/list", nil, &out)
@ -245,7 +253,9 @@ func TestAPIPathsList(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var out pathList var out pathList
httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/paths/list", nil, &out) httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/paths/list", nil, &out)
@ -271,7 +281,9 @@ func TestAPIPathsGet(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
for _, ca := range []string{"ok", "ok-nested", "not found"} { for _, ca := range []string{"ok", "ok-nested", "not found"} {
t.Run(ca, func(t *testing.T) { t.Run(ca, func(t *testing.T) {
@ -371,7 +383,9 @@ func TestAPIProtocolListGet(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
medi := test.UniqueMediaH264() medi := test.UniqueMediaH264()
@ -844,7 +858,9 @@ func TestAPIProtocolGetNotFound(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var pa string var pa string
switch ca { switch ca {
@ -934,7 +950,9 @@ func TestAPIProtocolKick(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
medi := test.MediaH264 medi := test.MediaH264
@ -1081,7 +1099,9 @@ func TestAPIProtocolKickNotFound(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
var pa string var pa string
switch ca { switch ca {

10
internal/core/metrics_test.go
View file

@ -65,7 +65,9 @@ func TestMetrics(t *testing.T) {
require.Equal(t, true, ok) require.Equal(t, true, ok)
defer p.Close() defer p.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
t.Run("initial", func(t *testing.T) { t.Run("initial", func(t *testing.T) {
bo := httpPullFile(t, hc, "http://localhost:9998/metrics") bo := httpPullFile(t, hc, "http://localhost:9998/metrics")
@ -165,8 +167,12 @@ webrtc_sessions_bytes_sent 0
su, err := url.Parse("http://localhost:8889/webrtc_path/whip") su, err := url.Parse("http://localhost:8889/webrtc_path/whip")
require.NoError(t, err) require.NoError(t, err)
tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
s := &webrtc.WHIPClient{ s := &webrtc.WHIPClient{
HTTPClient: &http.Client{Transport: &http.Transport{}}, HTTPClient: hc,
URL: su, URL: su,
Log: test.NilLogger{}, Log: test.NilLogger{},
} }

8
internal/core/path_test.go
View file

@ -397,7 +397,9 @@ func TestPathRunOnRead(t *testing.T) {
defer reader.Close() defer reader.Close()
case "webrtc": case "webrtc":
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
u, err := url.Parse("http://localhost:8889/test/whep?query=value") u, err := url.Parse("http://localhost:8889/test/whep?query=value")
require.NoError(t, err) require.NoError(t, err)
@ -543,7 +545,9 @@ func TestPathRecord(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, 1, len(files)) require.Equal(t, 1, len(files))
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
httpRequest(t, hc, http.MethodPatch, "http://localhost:9997/v3/config/paths/patch/all_others", map[string]interface{}{ httpRequest(t, hc, http.MethodPatch, "http://localhost:9997/v3/config/paths/patch/all_others", map[string]interface{}{
"record": false, "record": false,

4
internal/highleveltests/hls_manager_test.go
View file

@ -77,7 +77,9 @@ func TestHLSServerAuth(t *testing.T) {
usr = "testreader2" usr = "testreader2"
} }
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
res, err := hc.Get("http://" + usr + ":testpass@127.0.0.1:8888/teststream/index.m3u8?param=value") res, err := hc.Get("http://" + usr + ":testpass@127.0.0.1:8888/teststream/index.m3u8?param=value")
require.NoError(t, err) require.NoError(t, err)

4
internal/servers/hls/server_test.go
View file

@ -94,7 +94,9 @@ func TestServerNotFound(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer s.Close() defer s.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
func() { func() {
req, err := http.NewRequest(http.MethodGet, "http://myuser:mypass@127.0.0.1:8888/nonexisting/", nil) req, err := http.NewRequest(http.MethodGet, "http://myuser:mypass@127.0.0.1:8888/nonexisting/", nil)

16
internal/servers/webrtc/server_test.go
View file

@ -114,7 +114,9 @@ func TestServerStaticPages(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer s.Close() defer s.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
for _, path := range []string{"/stream", "/stream/publish", "/publish"} { for _, path := range []string{"/stream", "/stream/publish", "/publish"} {
func() { func() {
@ -160,7 +162,9 @@ func TestServerPublish(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer s.Close() defer s.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
// preflight requests must always work, without authentication // preflight requests must always work, without authentication
func() { func() {
@ -285,7 +289,9 @@ func TestServerRead(t *testing.T) {
u, err := url.Parse(ur) u, err := url.Parse(ur)
require.NoError(t, err) require.NoError(t, err)
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
wc := &webrtc.WHIPClient{ wc := &webrtc.WHIPClient{
HTTPClient: hc, HTTPClient: hc,
@ -370,7 +376,9 @@ func TestServerReadNotFound(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer s.Close() defer s.Close()
hc := &http.Client{Transport: &http.Transport{}} tr := &http.Transport{}
defer tr.CloseIdleConnections()
hc := &http.Client{Transport: tr}
iceServers, err := webrtc.WHIPOptionsICEServers(context.Background(), hc, iceServers, err := webrtc.WHIPOptionsICEServers(context.Background(), hc,
"http://myuser:mypass@localhost:8886/nonexisting/whep") "http://myuser:mypass@localhost:8886/nonexisting/whep")

11
internal/staticsources/hls/source.go
View file

@ -42,14 +42,17 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
decodeErrLogger := logger.NewLimitedLogger(s) decodeErrLogger := logger.NewLimitedLogger(s)
tr := &http.Transport{
TLSClientConfig: tls.ConfigForFingerprint(params.Conf.SourceFingerprint),
}
defer tr.CloseIdleConnections()
var c *gohlslib.Client var c *gohlslib.Client
c = &gohlslib.Client{ c = &gohlslib.Client{
URI: s.ResolvedSource, URI: s.ResolvedSource,
HTTPClient: &http.Client{ HTTPClient: &http.Client{
Timeout: time.Duration(s.ReadTimeout), Timeout: time.Duration(s.ReadTimeout),
Transport: &http.Transport{ Transport: tr,
TLSClientConfig: tls.ConfigForFingerprint(params.Conf.SourceFingerprint),
},
}, },
OnDownloadPrimaryPlaylist: func(u string) { OnDownloadPrimaryPlaylist: func(u string) {
s.Log(logger.Debug, "downloading primary playlist %v", u) s.Log(logger.Debug, "downloading primary playlist %v", u)

18
internal/staticsources/webrtc/source.go
View file

@ -40,18 +40,18 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
u.Scheme = strings.ReplaceAll(u.Scheme, "whep", "http") u.Scheme = strings.ReplaceAll(u.Scheme, "whep", "http")
hc := &http.Client{ tr := &http.Transport{
Timeout: time.Duration(s.ReadTimeout), TLSClientConfig: tls.ConfigForFingerprint(params.Conf.SourceFingerprint),
Transport: &http.Transport{
TLSClientConfig: tls.ConfigForFingerprint(params.Conf.SourceFingerprint),
},
} }
defer hc.CloseIdleConnections() defer tr.CloseIdleConnections()
client := webrtc.WHIPClient{ client := webrtc.WHIPClient{
HTTPClient: hc, HTTPClient: &http.Client{
URL: u, Timeout: time.Duration(s.ReadTimeout),
Log: s, Transport: tr,
},
URL: u,
Log: s,
} }
tracks, err := client.Read(params.Context) tracks, err := client.Read(params.Context)