From 0f2298d1930f3b2538b622ac92e3333cf9c972d2 Mon Sep 17 00:00:00 2001 From: Umputun Date: Mon, 3 Jun 2019 22:48:28 -0500 Subject: [PATCH] externalize all stream params --- backend/app/cmd/server.go | 13 +++++++++++-- backend/app/rest/api/rest.go | 7 ++++--- backend/app/rest/api/rest_public.go | 2 +- backend/app/rest/api/rest_public_test.go | 9 +++++++++ 4 files changed, 25 insertions(+), 6 deletions(-) diff --git a/backend/app/cmd/server.go b/backend/app/cmd/server.go index 72313356..b1c4d06a 100644 --- a/backend/app/cmd/server.go +++ b/backend/app/cmd/server.go @@ -46,6 +46,7 @@ type ServerCommand struct { Notify NotifyGroup `group:"notify" namespace:"notify" env-namespace:"NOTIFY"` Image ImageGroup `group:"image" namespace:"image" env-namespace:"IMAGE"` SSL SSLGroup `group:"ssl" namespace:"ssl" env-namespace:"SSL"` + Stream StreamGroup `group:"stream" namespace:"stream" env-namespace:"STREAM"` Sites []string `long:"site" env:"SITE" default:"remark" description:"site names" env-delim:","` AdminPasswd string `long:"admin-passwd" env:"ADMIN_PASSWD" default:"" description:"admin basic auth password"` @@ -170,6 +171,13 @@ type SSLGroup struct { ACMEEmail string `long:"acme-email" env:"ACME_EMAIL" description:"admin email for certificate notifications"` } +// StreamGroup define options for streaming apis +type StreamGroup struct { + RefreshInterval time.Duration `long:"refresh" env:"REFRESH" default:"5s" description:"refresh interval for streams"` + TimeOut time.Duration `long:"timeout" env:"TIMEOUT" default:"15m" description:"timeout to close streams on inactivity"` + MaxActive int `long:"max" env:"MAX" default:"500" description:"max number of parallel streams"` +} + // serverApp holds all active objects type serverApp struct { *ServerCommand @@ -304,8 +312,9 @@ func (s *ServerCommand) newServerApp() (*serverApp, error) { SSLConfig: sslConfig, UpdateLimiter: s.UpdateLimit, ImageService: imageService, - StreamTimeOut: time.Minute * 15, - StreamRefresh: time.Second * 5, + StreamTimeOut: s.Stream.TimeOut, + StreamRefresh: s.Stream.RefreshInterval, + StreamMaxActive: s.Stream.MaxActive, } srv.ScoreThresholds.Low, srv.ScoreThresholds.Critical = s.LowScore, s.CriticalScore diff --git a/backend/app/rest/api/rest.go b/backend/app/rest/api/rest.go index a56d180f..5029ba86 100644 --- a/backend/app/rest/api/rest.go +++ b/backend/app/rest/api/rest.go @@ -55,8 +55,9 @@ type Rest struct { } UpdateLimiter float64 - StreamTimeOut time.Duration - StreamRefresh time.Duration + StreamTimeOut time.Duration + StreamRefresh time.Duration + StreamMaxActive int SSLConfig SSLConfig httpsServer *http.Server @@ -338,7 +339,7 @@ func (s *Rest) controllerGroups() (public, private, admin, rss) { webRoot: s.WebRoot, streamTimeOut: s.StreamTimeOut, streamRefresh: s.StreamRefresh, - maxActiveStreams: 500, + maxActiveStreams: int32(s.StreamMaxActive), } privGrp := private{ diff --git a/backend/app/rest/api/rest_public.go b/backend/app/rest/api/rest_public.go index 388886d4..8e9611dc 100644 --- a/backend/app/rest/api/rest_public.go +++ b/backend/app/rest/api/rest_public.go @@ -562,7 +562,7 @@ func (s *public) eventsCh(ctx context.Context, fn eventFn) <-chan eventResp { for { select { case <-ctx.Done(): // request closed by remote client - log.Printf("[DEBUG] info stream closed by remote client, %v", ctx.Err()) + log.Printf("[DEBUG] stream closed by remote client, %v", ctx.Err()) return case <-tick.C: resp, upd, err := fn() diff --git a/backend/app/rest/api/rest_public_test.go b/backend/app/rest/api/rest_public_test.go index 2cd4ed22..82c3c965 100644 --- a/backend/app/rest/api/rest_public_test.go +++ b/backend/app/rest/api/rest_public_test.go @@ -533,6 +533,7 @@ func TestRest_InfoStream(t *testing.T) { srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only srv.pubRest.streamRefresh = 1 * time.Millisecond srv.pubRest.streamTimeOut = 300 * time.Millisecond + srv.pubRest.maxActiveStreams = 100 postComment(t, ts.URL) @@ -588,6 +589,7 @@ func TestRest_InfoStreamTimeout(t *testing.T) { srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only srv.pubRest.streamRefresh = 10 * time.Millisecond srv.pubRest.streamTimeOut = 450 * time.Millisecond + srv.pubRest.maxActiveStreams = 100 postComment(t, ts.URL) @@ -602,6 +604,7 @@ func TestRest_InfoStreamCancel(t *testing.T) { srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only srv.pubRest.streamRefresh = 10 * time.Millisecond srv.pubRest.streamTimeOut = 500 * time.Millisecond + srv.pubRest.maxActiveStreams = 100 postComment(t, ts.URL) @@ -653,6 +656,7 @@ func TestRest_LastCommentsStream(t *testing.T) { srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only srv.pubRest.streamRefresh = 10 * time.Millisecond srv.pubRest.streamTimeOut = 500 * time.Millisecond + srv.pubRest.maxActiveStreams = 100 postComment(t, ts.URL) @@ -691,6 +695,7 @@ func TestRest_LastCommentsStreamTimeout(t *testing.T) { srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only srv.pubRest.streamRefresh = 10 * time.Millisecond srv.pubRest.streamTimeOut = 450 * time.Millisecond + srv.pubRest.maxActiveStreams = 100 postComment(t, ts.URL) @@ -705,6 +710,7 @@ func TestRest_LastCommentsStreamCancel(t *testing.T) { srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only srv.pubRest.streamRefresh = 10 * time.Millisecond srv.pubRest.streamTimeOut = 500 * time.Millisecond + srv.pubRest.maxActiveStreams = 100 postComment(t, ts.URL) @@ -763,6 +769,9 @@ func TestRest_LastCommentsStreamTooMany(t *testing.T) { } wg.Wait() assert.Equal(t, int32(10), atomic.LoadInt32(&errsCount), "10 streams rejected") + + _, code := get(t, ts.URL+"/api/v1/stream/last?site=radio-t") + assert.Equal(t, 200, code, "all streams closed, good to go again") } func postComment(t *testing.T, url string) {