externalize all stream params

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

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{

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()

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) {