From 20764b7b4e6d623ce79a5962751035663eed4bb5 Mon Sep 17 00:00:00 2001 From: Umputun Date: Wed, 5 Jun 2019 02:14:34 -0500 Subject: [PATCH] enclose streamer in rest's struct --- backend/app/cmd/server.go | 8 ++-- backend/app/rest/api/rest.go | 11 +----- backend/app/rest/api/rest_public.go | 8 ++-- backend/app/rest/api/rest_public_test.go | 49 ++++++++++++------------ backend/app/rest/api/rest_test.go | 5 +++ backend/app/rest/api/stream.go | 22 +++++------ backend/app/rest/api/stream_test.go | 20 +++++----- 7 files changed, 61 insertions(+), 62 deletions(-) diff --git a/backend/app/cmd/server.go b/backend/app/cmd/server.go index b1c4d06a..a13d29e1 100644 --- a/backend/app/cmd/server.go +++ b/backend/app/cmd/server.go @@ -312,9 +312,11 @@ func (s *ServerCommand) newServerApp() (*serverApp, error) { SSLConfig: sslConfig, UpdateLimiter: s.UpdateLimit, ImageService: imageService, - StreamTimeOut: s.Stream.TimeOut, - StreamRefresh: s.Stream.RefreshInterval, - StreamMaxActive: s.Stream.MaxActive, + Streamer: &api.Streamer{ + TimeOut: s.Stream.TimeOut, + Refresh: s.Stream.RefreshInterval, + MaxActive: int32(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 a23d6cfa..47ba1f8d 100644 --- a/backend/app/rest/api/rest.go +++ b/backend/app/rest/api/rest.go @@ -44,6 +44,7 @@ type Rest struct { Migrator *Migrator NotifyService *notify.Service ImageService *image.Service + Streamer *Streamer WebRoot string RemarkURL string @@ -55,10 +56,6 @@ type Rest struct { } UpdateLimiter float64 - StreamTimeOut time.Duration - StreamRefresh time.Duration - StreamMaxActive int - SSLConfig SSLConfig httpsServer *http.Server httpServer *http.Server @@ -337,11 +334,7 @@ func (s *Rest) controllerGroups() (public, private, admin, rss) { commentFormatter: s.CommentFormatter, readOnlyAge: s.ReadOnlyAge, webRoot: s.WebRoot, - streamer: &streamer{ - timeout: s.StreamTimeOut, - refresh: s.StreamRefresh, - maxActive: int32(s.StreamMaxActive), - }, + streamer: s.Streamer, } privGrp := private{ diff --git a/backend/app/rest/api/rest_public.go b/backend/app/rest/api/rest_public.go index 9361229c..7cc5f12c 100644 --- a/backend/app/rest/api/rest_public.go +++ b/backend/app/rest/api/rest_public.go @@ -29,8 +29,8 @@ type public struct { readOnlyAge int commentFormatter *store.CommentFormatter imageService *image.Service + streamer *Streamer webRoot string - streamer *streamer } type pubStore interface { @@ -150,7 +150,7 @@ func (s *public) infoCtrl(w http.ResponseWriter, r *http.Request) { // GET /stream/info?site=siteID&url=post-url - get info stream about the post func (s *public) infoStreamCtrl(w http.ResponseWriter, r *http.Request) { locator := store.Locator{SiteID: r.URL.Query().Get("site"), URL: r.URL.Query().Get("url")} - log.Printf("[DEBUG] start stream for %+v, timeout=%v, refresh=%v", locator, s.streamer.timeout, s.streamer.refresh) + log.Printf("[DEBUG] start stream for %+v, timeout=%v, refresh=%v", locator, s.streamer.TimeOut, s.streamer.Refresh) fn := func() steamEventFn { lastTS := time.Time{} @@ -179,7 +179,7 @@ func (s *public) infoStreamCtrl(w http.ResponseWriter, r *http.Request) { } } - if err := s.streamer.activate(r.Context(), fn, w); err != nil { + if err := s.streamer.Activate(r.Context(), fn, w); err != nil { rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't stream", rest.ErrInternal) } } @@ -252,7 +252,7 @@ func (s *public) lastCommentsStreamCtrl(w http.ResponseWriter, r *http.Request) } } - if err := s.streamer.activate(r.Context(), fn, w); err != nil { + if err := s.streamer.Activate(r.Context(), fn, w); err != nil { rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't stream", rest.ErrInternal) } } diff --git a/backend/app/rest/api/rest_public_test.go b/backend/app/rest/api/rest_public_test.go index 7c4ffabb..74aa6544 100644 --- a/backend/app/rest/api/rest_public_test.go +++ b/backend/app/rest/api/rest_public_test.go @@ -42,7 +42,6 @@ func TestRest_Preview(t *testing.T) { assert.Nil(t, err) assert.Equal(t, "

test 123

\n", string(b)) - resp, err = post(t, ts.URL+"/api/v1/preview", "bad") assert.Nil(t, err) assert.Equal(t, 400, resp.StatusCode) @@ -550,9 +549,9 @@ func TestRest_InfoStream(t *testing.T) { ts, srv, teardown := startupT(t) defer teardown() srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only - srv.pubRest.streamer.refresh = 1 * time.Millisecond - srv.pubRest.streamer.timeout = 300 * time.Millisecond - srv.pubRest.streamer.maxActive = 100 + srv.pubRest.streamer.Refresh = 1 * time.Millisecond + srv.pubRest.streamer.TimeOut = 300 * time.Millisecond + srv.pubRest.streamer.MaxActive = 100 postComment(t, ts.URL) @@ -583,9 +582,9 @@ func TestRest_InfoStreamTooMany(t *testing.T) { ts, srv, teardown := startupT(t) defer teardown() srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only - srv.pubRest.streamer.refresh = 1 * time.Millisecond - srv.pubRest.streamer.timeout = 300 * time.Millisecond - srv.pubRest.streamer.maxActive = 10 + srv.pubRest.streamer.Refresh = 1 * time.Millisecond + srv.pubRest.streamer.TimeOut = 300 * time.Millisecond + srv.pubRest.streamer.MaxActive = 10 postComment(t, ts.URL) @@ -609,9 +608,9 @@ func TestRest_InfoStreamTimeout(t *testing.T) { ts, srv, teardown := startupT(t) defer teardown() srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only - srv.pubRest.streamer.refresh = 10 * time.Millisecond - srv.pubRest.streamer.timeout = 450 * time.Millisecond - srv.pubRest.streamer.maxActive = 100 + srv.pubRest.streamer.Refresh = 10 * time.Millisecond + srv.pubRest.streamer.TimeOut = 450 * time.Millisecond + srv.pubRest.streamer.MaxActive = 100 postComment(t, ts.URL) @@ -625,9 +624,9 @@ func TestRest_InfoStreamCancel(t *testing.T) { ts, srv, teardown := startupT(t) defer teardown() srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only - srv.pubRest.streamer.refresh = 10 * time.Millisecond - srv.pubRest.streamer.timeout = 500 * time.Millisecond - srv.pubRest.streamer.maxActive = 100 + srv.pubRest.streamer.Refresh = 10 * time.Millisecond + srv.pubRest.streamer.TimeOut = 500 * time.Millisecond + srv.pubRest.streamer.MaxActive = 100 postComment(t, ts.URL) @@ -679,9 +678,9 @@ func TestRest_Robots(t *testing.T) { func TestRest_LastCommentsStream(t *testing.T) { ts, srv, teardown := startupT(t) srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only - srv.pubRest.streamer.refresh = 10 * time.Millisecond - srv.pubRest.streamer.timeout = 500 * time.Millisecond - srv.pubRest.streamer.maxActive = 100 + srv.pubRest.streamer.Refresh = 10 * time.Millisecond + srv.pubRest.streamer.TimeOut = 500 * time.Millisecond + srv.pubRest.streamer.MaxActive = 100 postComment(t, ts.URL) @@ -718,9 +717,9 @@ func TestRest_LastCommentsStreamTimeout(t *testing.T) { ts, srv, teardown := startupT(t) defer teardown() srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only - srv.pubRest.streamer.refresh = 10 * time.Millisecond - srv.pubRest.streamer.timeout = 450 * time.Millisecond - srv.pubRest.streamer.maxActive = 100 + srv.pubRest.streamer.Refresh = 10 * time.Millisecond + srv.pubRest.streamer.TimeOut = 450 * time.Millisecond + srv.pubRest.streamer.MaxActive = 100 postComment(t, ts.URL) @@ -733,9 +732,9 @@ func TestRest_LastCommentsStreamTimeout(t *testing.T) { func TestRest_LastCommentsStreamCancel(t *testing.T) { ts, srv, teardown := startupT(t) srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only - srv.pubRest.streamer.refresh = 10 * time.Millisecond - srv.pubRest.streamer.timeout = 500 * time.Millisecond - srv.pubRest.streamer.maxActive = 100 + srv.pubRest.streamer.Refresh = 10 * time.Millisecond + srv.pubRest.streamer.TimeOut = 500 * time.Millisecond + srv.pubRest.streamer.MaxActive = 100 postComment(t, ts.URL) @@ -774,9 +773,9 @@ func TestRest_LastCommentsStreamTooMany(t *testing.T) { ts, srv, teardown := startupT(t) defer teardown() srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only - srv.pubRest.streamer.refresh = 1 * time.Millisecond - srv.pubRest.streamer.timeout = 300 * time.Millisecond - srv.pubRest.streamer.maxActive = 10 + srv.pubRest.streamer.Refresh = 1 * time.Millisecond + srv.pubRest.streamer.TimeOut = 300 * time.Millisecond + srv.pubRest.streamer.MaxActive = 10 postComment(t, ts.URL) diff --git a/backend/app/rest/api/rest_test.go b/backend/app/rest/api/rest_test.go index ed5f045d..a0e8566a 100644 --- a/backend/app/rest/api/rest_test.go +++ b/backend/app/rest/api/rest_test.go @@ -339,6 +339,11 @@ func startupT(t *testing.T) (ts *httptest.Server, srv *Rest, teardown func()) { Cache: &cache.Nop{}, KeyStore: adminStore, }, + Streamer: &Streamer{ + Refresh: 100 * time.Millisecond, + TimeOut: 5 * time.Second, + MaxActive: 100, + }, } srv.ScoreThresholds.Low, srv.ScoreThresholds.Critical = -5, -10 diff --git a/backend/app/rest/api/stream.go b/backend/app/rest/api/stream.go index 278f82a3..18d5ff7f 100644 --- a/backend/app/rest/api/stream.go +++ b/backend/app/rest/api/stream.go @@ -11,11 +11,11 @@ import ( "github.com/pkg/errors" ) -// streamer creates endless stream of \n seprated json records send to remote client -type streamer struct { - timeout time.Duration - refresh time.Duration - maxActive int32 +// Streamer creates endless stream of \n separated json records send to remote client +type Streamer struct { + TimeOut time.Duration + Refresh time.Duration + MaxActive int32 activeCount int32 } @@ -27,15 +27,15 @@ type steamEventResp struct { err error } -// activate starts blocking function streaming update created by eventFn to ResponseWriter +// Activate starts blocking function streaming update created by eventFn to ResponseWriter // canceled on context or inactivity timeout // note: eventFn is a closure needed to allow state management inside eventFn -func (s *streamer) activate(ctx context.Context, eventFn func() steamEventFn, w io.Writer) error { +func (s *Streamer) Activate(ctx context.Context, eventFn func() steamEventFn, w io.Writer) error { updCh := s.eventsCh(ctx, eventFn()) count := atomic.AddInt32(&s.activeCount, 1) defer atomic.AddInt32(&s.activeCount, -1) - if count > s.maxActive { + if count > s.MaxActive { return errors.New("too many streams") } @@ -44,7 +44,7 @@ func (s *streamer) activate(ctx context.Context, eventFn func() steamEventFn, w case <-ctx.Done(): // request closed by remote client log.Printf("[DEBUG] stream closed by remote client, %s", ctx.Err()) return nil - case <-time.After(s.timeout): // request closed by timeout + case <-time.After(s.TimeOut): // request closed by timeout log.Printf("[DEBUG] stream closed due to timeout") return nil case resp, ok := <-updCh: // new update @@ -65,10 +65,10 @@ func (s *streamer) activate(ctx context.Context, eventFn func() steamEventFn, w } // populate updates to chan, break on context close -func (s *streamer) eventsCh(ctx context.Context, fn steamEventFn) <-chan steamEventResp { +func (s *Streamer) eventsCh(ctx context.Context, fn steamEventFn) <-chan steamEventResp { ch := make(chan steamEventResp) go func() { - tick := time.NewTicker(s.refresh) + tick := time.NewTicker(s.Refresh) defer func() { close(ch) tick.Stop() diff --git a/backend/app/rest/api/stream_test.go b/backend/app/rest/api/stream_test.go index c32ca5a4..0a15374f 100644 --- a/backend/app/rest/api/stream_test.go +++ b/backend/app/rest/api/stream_test.go @@ -11,10 +11,10 @@ import ( ) func TestStream_Timeout(t *testing.T) { - s := streamer{ - refresh: 10 * time.Millisecond, - timeout: 100 * time.Millisecond, - maxActive: 10, + s := Streamer{ + Refresh: 10 * time.Millisecond, + TimeOut: 100 * time.Millisecond, + MaxActive: 10, } eventFn := func() steamEventFn { @@ -29,16 +29,16 @@ func TestStream_Timeout(t *testing.T) { } buf := bytes.Buffer{} - err := s.activate(context.Background(), eventFn, &buf) + err := s.Activate(context.Background(), eventFn, &buf) assert.NoError(t, err) assert.Equal(t, "some data 1\nsome data 3\nsome data 5\nsome data 7\nsome data 9\n", buf.String()) } func TestStream_Cancel(t *testing.T) { - s := streamer{ - refresh: 10 * time.Millisecond, - timeout: 100 * time.Millisecond, - maxActive: 10, + s := Streamer{ + Refresh: 10 * time.Millisecond, + TimeOut: 100 * time.Millisecond, + MaxActive: 10, } eventFn := func() steamEventFn { @@ -55,7 +55,7 @@ func TestStream_Cancel(t *testing.T) { buf := bytes.Buffer{} ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) defer cancel() - err := s.activate(ctx, eventFn, &buf) + err := s.Activate(ctx, eventFn, &buf) assert.NoError(t, err) assert.Equal(t, "some data 1\nsome data 3\nsome data 5\nsome data 7\nsome data 9\n", buf.String()) }