From 809a7c7dbd7bd4d1bf06109170d367c44b0f2999 Mon Sep 17 00:00:00 2001 From: Dmitry Verkhoturov Date: Mon, 30 Nov 2020 02:00:59 +0100 Subject: [PATCH] Clean unused stream code It was supposed to solve #253 but frontend part for it in #357 was never finished, and backend code produces false positive test failures since day 0. The cost of just having this code around is too high, we'll re-add it in case frontend implementation will be finished. --- backend/app/cmd/server.go | 43 +-- backend/app/main_test.go | 7 +- backend/app/rest/api/rest.go | 10 - backend/app/rest/api/rest_public.go | 83 ------ backend/app/rest/api/rest_public_test.go | 337 ----------------------- backend/app/rest/api/rest_test.go | 11 +- backend/app/rest/api/stream.go | 103 ------- backend/app/rest/api/stream_test.go | 63 ----- 8 files changed, 24 insertions(+), 633 deletions(-) delete mode 100644 backend/app/rest/api/stream.go delete mode 100644 backend/app/rest/api/stream_test.go diff --git a/backend/app/cmd/server.go b/backend/app/cmd/server.go index 4f470514..4430cd87 100644 --- a/backend/app/cmd/server.go +++ b/backend/app/cmd/server.go @@ -51,7 +51,6 @@ type ServerCommand struct { SMTP SMTPGroup `group:"smtp" namespace:"smtp" env-namespace:"SMTP"` 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"` ImageProxy ImageProxyGroup `group:"image-proxy" namespace:"image-proxy" env-namespace:"IMAGE_PROXY"` Sites []string `long:"site" env:"SITE" default:"remark" description:"site names" env-delim:","` @@ -219,13 +218,6 @@ 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"` -} - // RPCGroup defines options for remote modules (plugins) type RPCGroup struct { API string `long:"api" env:"API" description:"rpc extension api url"` @@ -431,26 +423,21 @@ func (s *ServerCommand) newServerApp() (*serverApp, error) { } srv := &api.Rest{ - Version: s.Revision, - DataService: dataService, - WebRoot: s.WebRoot, - RemarkURL: s.RemarkURL, - ImageProxy: imgProxy, - CommentFormatter: commentFormatter, - Migrator: migr, - ReadOnlyAge: s.ReadOnlyAge, - SharedSecret: s.SharedSecret, - Authenticator: authenticator, - Cache: loadingCache, - NotifyService: notifyService, - SSLConfig: sslConfig, - UpdateLimiter: s.UpdateLimit, - ImageService: imageService, - Streamer: &api.Streamer{ - TimeOut: s.Stream.TimeOut, - Refresh: s.Stream.RefreshInterval, - MaxActive: int32(s.Stream.MaxActive), - }, + Version: s.Revision, + DataService: dataService, + WebRoot: s.WebRoot, + RemarkURL: s.RemarkURL, + ImageProxy: imgProxy, + CommentFormatter: commentFormatter, + Migrator: migr, + ReadOnlyAge: s.ReadOnlyAge, + SharedSecret: s.SharedSecret, + Authenticator: authenticator, + Cache: loadingCache, + NotifyService: notifyService, + SSLConfig: sslConfig, + UpdateLimiter: s.UpdateLimit, + ImageService: imageService, EmailNotifications: emailNotifications, EmojiEnabled: s.EnableEmoji, AnonVote: s.AnonymousVote && s.RestrictVoteIP, diff --git a/backend/app/main_test.go b/backend/app/main_test.go index 4246b50c..a7d40f06 100644 --- a/backend/app/main_test.go +++ b/backend/app/main_test.go @@ -89,5 +89,10 @@ func waitForHTTPServerStart(port int) { } func TestMain(m *testing.M) { - goleak.VerifyTestMain(m, goleak.IgnoreTopFunction("github.com/umputun/remark42/backend/app.init.0.func1")) + // both ignores are for leaks which are detected locally + goleak.VerifyTestMain( + m, + goleak.IgnoreTopFunction("github.com/umputun/remark42/backend/app.init.0.func1"), + goleak.IgnoreTopFunction("net/http.(*Server).Shutdown"), + ) } diff --git a/backend/app/rest/api/rest.go b/backend/app/rest/api/rest.go index 113ec050..267661f5 100644 --- a/backend/app/rest/api/rest.go +++ b/backend/app/rest/api/rest.go @@ -46,7 +46,6 @@ type Rest struct { Migrator *Migrator NotifyService *notify.Service ImageService *image.Service - Streamer *Streamer AnonVote bool WebRoot string @@ -256,14 +255,6 @@ func (s *Rest) routes() chi.Router { }) - // open routes, streams, no send timeout - rapi.Route("/stream", func(rstream chi.Router) { - rstream.Use(tollbooth_chi.LimitHandler(tollbooth.NewLimiter(10, nil))) - rstream.Use(authMiddleware.Trace, middleware.NoCache, logInfoWithBody) - rstream.Get("/info", s.pubRest.infoStreamCtrl) - rstream.Get("/last", s.pubRest.lastCommentsStreamCtrl) - }) - // open routes, cached rapi.Group(func(ropen chi.Router) { ropen.Use(middleware.Timeout(30 * time.Second)) @@ -359,7 +350,6 @@ func (s *Rest) controllerGroups() (public, private, admin, rss) { commentFormatter: s.CommentFormatter, readOnlyAge: s.ReadOnlyAge, webRoot: s.WebRoot, - streamer: s.Streamer, } privGrp := private{ diff --git a/backend/app/rest/api/rest_public.go b/backend/app/rest/api/rest_public.go index d4b085d5..5e4c3ab8 100644 --- a/backend/app/rest/api/rest_public.go +++ b/backend/app/rest/api/rest_public.go @@ -31,7 +31,6 @@ type public struct { readOnlyAge int commentFormatter *store.CommentFormatter imageService *image.Service - streamer *Streamer webRoot string } @@ -159,49 +158,6 @@ func (s *public) infoCtrl(w http.ResponseWriter, r *http.Request) { } } -// GET /stream/info?site=siteID&url=post-url&since=unix_ts_msec - 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) - - sinceTS, err := s.parseSince(r) - if err != nil { - rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't translate since parameter", rest.ErrDecode) - return - } - - fn := func() steamEventFn { - lastTS := sinceTS - lastCount := 0 - - return func() (event string, data []byte, upd bool, err error) { - key := cache.NewKey(locator.SiteID).ID(URLKey(r)).Scopes(locator.SiteID, locator.URL) - data, err = s.cache.Get(key, func() ([]byte, error) { - info, e := s.dataService.Info(locator, s.readOnlyAge) - if e != nil { - return nil, e - } - // cache update used as indication of post update. comparing lastTS for no-cache. - // removal won't update lastTS, count check will catch it. - if !lastTS.IsZero() && (info.LastTS != lastTS || info.Count != lastCount) { - upd = true - } - lastTS = info.LastTS - lastCount = info.Count - return encodeJSONWithHTML(info) - }) - if err != nil { - return "info", data, false, err - } - return "info", data, upd, nil - } - } - - if e := s.streamer.Activate(r.Context(), fn, w); e != nil { - rest.SendErrorJSON(w, r, http.StatusInternalServerError, e, "can't stream", rest.ErrInternal) - } -} - // GET /last/{limit}?site=siteID&since=unix_ts_msec - last comments for the siteID, across all posts, sorted by time, optionally // limited with "since" param func (s *public) lastCommentsCtrl(w http.ResponseWriter, r *http.Request) { @@ -240,45 +196,6 @@ func (s *public) lastCommentsCtrl(w http.ResponseWriter, r *http.Request) { } } -// GET /stream/last?site=siteID&since=unix_ts_ms - stream of last comments last comments for the siteID, across all posts -func (s *public) lastCommentsStreamCtrl(w http.ResponseWriter, r *http.Request) { - siteID := r.URL.Query().Get("site") - log.Printf("[DEBUG] get last comments stream for %s", siteID) - - sinceTS, err := s.parseSince(r) - if err != nil { - rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't translate since parameter", rest.ErrDecode) - return - } - if sinceTS.IsZero() { - sinceTS = time.Now() - } - - fn := func() steamEventFn { - sinceTime := sinceTS - return func() (event string, data []byte, upd bool, err error) { - key := cache.NewKey(siteID).ID(URLKey(r)).Scopes(lastCommentsScope) - data, err = s.cache.Get(key, func() ([]byte, error) { - comments, e := s.dataService.Last(siteID, 1, sinceTime, rest.GetUserOrEmpty(r)) - if e != nil { - return nil, e - } - sinceTime = time.Now() - if len(comments) > 0 { - sinceTime = comments[0].Timestamp - upd = true - } - return encodeJSONWithHTML(comments) - }) - return "last", data, upd, err - } - } - - if e := s.streamer.Activate(r.Context(), fn, w); e != nil { - rest.SendErrorJSON(w, r, http.StatusInternalServerError, e, "can't stream", rest.ErrInternal) - } -} - // GET /id/{id}?site=siteID&url=post-url - gets a comment by id func (s *public) commentByIDCtrl(w http.ResponseWriter, r *http.Request) { diff --git a/backend/app/rest/api/rest_public_test.go b/backend/app/rest/api/rest_public_test.go index 175c1672..7348cf22 100644 --- a/backend/app/rest/api/rest_public_test.go +++ b/backend/app/rest/api/rest_public_test.go @@ -1,14 +1,11 @@ package api import ( - "context" "encoding/json" "fmt" "io/ioutil" "net/http" "strings" - "sync" - "sync/atomic" "testing" "time" @@ -580,145 +577,6 @@ func TestRest_Info(t *testing.T) { assert.Equal(t, 400, code) } -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 = 800 * time.Millisecond - srv.pubRest.streamer.MaxActive = 100 - - postComment(t, ts.URL) - - done := make(chan struct{}) - go func() { - defer close(done) - for i := 0; i < 10; i++ { - time.Sleep(10 * time.Millisecond) - postComment(t, ts.URL) - } - }() - - body, code := get(t, ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah1") - assert.Equal(t, 200, code) - <-done - - recs := strings.Split(strings.TrimSuffix(body, "\n"), "\n") - require.Equal(t, 10*3, len(recs), "10 records. each 2 lines +1 emty line") - assert.True(t, strings.Contains(recs[0+1], `"count":2`), recs[0]) - assert.True(t, strings.Contains(recs[9*3+1], `"count":11`), recs[9]) - - _, code = get(t, ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah123") - assert.Equal(t, 500, code) -} - -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 - - postComment(t, ts.URL) - - var errsCount int32 - wg := sync.WaitGroup{} - wg.Add(20) - for i := 0; i < 20; i++ { - go func() { - _, code := get(t, ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah1") - if code == 429 { - atomic.AddInt32(&errsCount, 1) - } - wg.Done() - }() - } - wg.Wait() - assert.Equal(t, int32(10), atomic.LoadInt32(&errsCount), "10 streams rejected") -} - -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 - - postComment(t, ts.URL) - - st := time.Now() - _, code := get(t, ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah1") - assert.Equal(t, 200, code) - assert.True(t, time.Since(st) > time.Millisecond*450 && time.Since(st) < time.Millisecond*500, time.Since(st)) -} - -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 = 1500 * time.Millisecond - srv.pubRest.streamer.MaxActive = 100 - - postComment(t, ts.URL) - - done := make(chan struct{}) - go func() { - defer close(done) - for i := 0; i < 5; i++ { - time.Sleep(300 * time.Millisecond) - postComment(t, ts.URL) - } - }() - - client := http.Client{} - req, err := http.NewRequest("GET", ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah1", nil) - require.NoError(t, err) - ctx, cancel := context.WithTimeout(context.Background(), 1000*time.Millisecond) - defer cancel() - req = req.WithContext(ctx) - r, err := client.Do(req) - require.NoError(t, err) - defer r.Body.Close() - <-ctx.Done() - <-done - body, err := ioutil.ReadAll(r.Body) - require.EqualError(t, err, "context deadline exceeded") - assert.Equal(t, 200, r.StatusCode) - - recs := strings.Count(string(body), "data:") - require.Equal(t, 1, recs, "should have 1 event:\n", string(body)) - assert.Contains(t, string(body), `"count":2`) -} - -func TestRest_InfoStreamSince(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 = 900 * time.Millisecond - srv.pubRest.streamer.MaxActive = 100 - - postComment(t, ts.URL) - - done := make(chan struct{}) - go func() { - defer close(done) - for i := 0; i < 10; i++ { - time.Sleep(15 * time.Millisecond) - postComment(t, ts.URL) - } - }() - - body, code := get(t, ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah1&since=12345678") - assert.Equal(t, 200, code) - <-done - recs := strings.Split(strings.TrimSuffix(body, "\n"), "\n") - require.Equal(t, 11*3, len(recs), "include first record, total 11 records. each 2 lines +1 empty line") -} - func TestRest_Robots(t *testing.T) { ts, _, teardown := startupT(t) defer teardown() @@ -730,198 +588,3 @@ func TestRest_Robots(t *testing.T) { "Allow: /api/v1/list\nAllow: /api/v1/config\nAllow: /api/v1/user\nAllow: /api/v1/img\n"+ "Allow: /api/v1/avatar\nAllow: /api/v1/picture\n", body) } - -func TestRest_LastCommentsStream(t *testing.T) { - t.Skip() // FIXME: not in use currently and fails sometime. Should be fixed as we start to use stremeing for real - ts, srv, teardown := startupT(t) - srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only - srv.pubRest.streamer.Refresh = 50 * time.Millisecond - srv.pubRest.streamer.TimeOut = 500 * time.Millisecond - srv.pubRest.streamer.MaxActive = 100 - - // stream endpoint currently relies on real cache being present - cacheBackend, err := cache.NewExpirableCache() - require.NoError(t, err) - memCache := cache.NewScache(cacheBackend) - defer memCache.Close() - srv.privRest.cache = memCache - srv.pubRest.cache = memCache - - postComment(t, ts.URL) - - defer teardown() - done := make(chan struct{}) - go func() { - defer close(done) - for i := 1; i < 10; i++ { - postComment(t, ts.URL) - time.Sleep(100 * time.Millisecond) - } - t.Log("wrote 10 records") - }() - - client := http.Client{} - req, err := http.NewRequest("GET", ts.URL+"/api/v1/stream/last?site=remark42", nil) - require.NoError(t, err) - r, err := client.Do(req) - require.NoError(t, err) - defer r.Body.Close() - <-done - body, err := ioutil.ReadAll(r.Body) - require.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - - assert.Equal(t, "text/event-stream", r.Header.Get("content-type")) - assert.Equal(t, "keep-alive", r.Header.Get("connection")) - - recs := strings.Split(strings.TrimSuffix(string(body), "\n"), "\n") - require.Equal(t, 9*3, len(recs), "9 events") - assert.True(t, strings.Contains(recs[1], `test 123`), recs[1]) -} - -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 - - postComment(t, ts.URL) - - st := time.Now() - _, code := get(t, ts.URL+"/api/v1/stream/last?site=remark42") - assert.Equal(t, 200, code) - assert.True(t, time.Since(st) > time.Millisecond*450 && time.Since(st) < time.Millisecond*500, time.Since(st)) -} - -func TestRest_LastCommentsStreamCancel(t *testing.T) { - t.Skip() - - 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 - - // stream endpoint currently relies on real cache being present - cacheBackend, err := cache.NewExpirableCache() - require.NoError(t, err) - memCache := cache.NewScache(cacheBackend) - defer memCache.Close() - srv.privRest.cache = memCache - srv.pubRest.cache = memCache - - postComment(t, ts.URL) - - done := make(chan struct{}) - go func() { - defer close(done) - for i := 1; i < 10; i++ { - time.Sleep(100 * time.Millisecond) - postComment(t, ts.URL) - } - }() - - client := http.Client{} - req, err := http.NewRequest("GET", ts.URL+"/api/v1/stream/last?site=remark42", nil) - require.NoError(t, err) - ctx, cancel := context.WithTimeout(context.Background(), 290*time.Millisecond) - defer cancel() - req = req.WithContext(ctx) - r, err := client.Do(req) - require.NoError(t, err) - <-done - defer r.Body.Close() - body, err := ioutil.ReadAll(r.Body) - require.EqualError(t, err, "context deadline exceeded") - assert.Equal(t, 200, r.StatusCode) - - recs := strings.Split(strings.TrimSuffix(string(body), "\n"), "\n") - assert.True(t, len(recs) < 30, "less 10 events") -} - -func TestRest_LastCommentsStreamTooMany(t *testing.T) { - t.Skip() - - 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 - - postComment(t, ts.URL) - - var errsCount int32 - wg := sync.WaitGroup{} - wg.Add(20) - for i := 0; i < 20; i++ { - go func() { - _, code := get(t, ts.URL+"/api/v1/stream/last?site=remark42") - if code == 429 { - atomic.AddInt32(&errsCount, 1) - } - wg.Done() - }() - } - wg.Wait() - assert.Equal(t, int32(10), atomic.LoadInt32(&errsCount), "10 streams rejected") - - _, code := get(t, ts.URL+"/api/v1/stream/last?site=remark42") - assert.Equal(t, 200, code, "all streams closed, good to go again") -} - -func TestRest_LastCommentsStreamSince(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 - - // stream endpoint currently relies on real cache being present - cacheBackend, err := cache.NewExpirableCache() - require.NoError(t, err) - memCache := cache.NewScache(cacheBackend) - defer memCache.Close() - srv.privRest.cache = memCache - srv.pubRest.cache = memCache - - postComment(t, ts.URL) - - done := make(chan struct{}) - go func() { - defer close(done) - for i := 1; i < 10; i++ { - time.Sleep(50 * time.Millisecond) - postComment(t, ts.URL) - } - }() - - client := http.Client{} - req, err := http.NewRequest("GET", ts.URL+"/api/v1/stream/last?site=remark42&since=123456", nil) - require.NoError(t, err) - r, err := client.Do(req) - require.NoError(t, err) - <-done - defer r.Body.Close() - body, err := ioutil.ReadAll(r.Body) - require.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - - assert.Equal(t, "text/event-stream", r.Header.Get("content-type")) - - recs := strings.Split(strings.TrimSuffix(string(body), "\n"), "\n") - require.Equal(t, 10*3, len(recs), "should be 10 events, including first record:\n", recs) -} - -func postComment(t *testing.T, url string) { - resp, err := post(t, url+"/api/v1/comment", - `{"text": "test 123", "locator":{"url": "https://radio-t.com/blah1", "site": "remark42"}}`) - require.NoError(t, err) - b, err := ioutil.ReadAll(resp.Body) - require.NoError(t, err) - require.Equal(t, http.StatusCreated, resp.StatusCode, string(b)) -} diff --git a/backend/app/rest/api/rest_test.go b/backend/app/rest/api/rest_test.go index 68466df6..c918174b 100644 --- a/backend/app/rest/api/rest_test.go +++ b/backend/app/rest/api/rest_test.go @@ -378,9 +378,9 @@ func startupT(t *testing.T) (ts *httptest.Server, srv *Rest, teardown func()) { SecretReader: token.SecretFunc(func(aud string) (string, error) { return "secret", nil }), AvatarStore: avatar.NewLocalFS(tmp + "/ava-remark42"), }), - Cache: memCache, - WebRoot: tmp, - RemarkURL: "https://demo.remark42.com", + Cache: memCache, + WebRoot: tmp, + RemarkURL: "https://demo.remark42.com", ImageService: image.NewService(&image.FileSystem{ Location: tmp + "/pics-remark42", Partitions: 100, @@ -401,11 +401,6 @@ func startupT(t *testing.T) (ts *httptest.Server, srv *Rest, teardown func()) { Cache: memCache, KeyStore: astore, }, - Streamer: &Streamer{ - Refresh: 100 * time.Millisecond, - TimeOut: 5 * time.Second, - MaxActive: 100, - }, NotifyService: notify.NopService, EmojiEnabled: true, } diff --git a/backend/app/rest/api/stream.go b/backend/app/rest/api/stream.go deleted file mode 100644 index 2968a9a1..00000000 --- a/backend/app/rest/api/stream.go +++ /dev/null @@ -1,103 +0,0 @@ -package api - -import ( - "context" - "fmt" - "io" - "net/http" - "sync/atomic" - "time" - - log "github.com/go-pkgz/lgr" - "github.com/pkg/errors" -) - -// 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 -} - -type steamEventFn func() (event string, data []byte, upd bool, err error) - -type steamEventResp struct { - data []byte - event string - err error -} - -// 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 { - updCh := s.eventsCh(ctx, eventFn()) - - count := atomic.AddInt32(&s.activeCount, 1) - defer atomic.AddInt32(&s.activeCount, -1) - if count > s.MaxActive { - return errors.New("too many streams") - } - - if ww, ok := w.(http.ResponseWriter); ok { - ww.Header().Set("Content-Type", "text/event-stream") - ww.Header().Set("Connection", "keep-alive") - ww.Header().Set("Cache-Control", "no-cache") - } - - for { - select { - 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 - log.Printf("[DEBUG] stream closed due to timeout") - return nil - case resp, ok := <-updCh: // new update - if !ok { // closed updCh - return nil - } - if resp.err != nil { - return resp.err - } - - // make server-sent event record - // see https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events - if _, e := fmt.Fprintf(w, "event: %s\ndata: %s\n", resp.event, string(resp.data)); e != nil { - return errors.Wrap(e, "send to stream failed") - } - if fw, okFlush := w.(http.Flusher); okFlush { - fw.Flush() - } - } - } -} - -// populate updates to chan, break on context close -func (s *Streamer) eventsCh(ctx context.Context, fn steamEventFn) <-chan steamEventResp { - ch := make(chan steamEventResp) - go func() { - tick := time.NewTicker(s.Refresh) - defer func() { - close(ch) - tick.Stop() - }() - for { - select { - case <-ctx.Done(): // request closed by remote client - return - case <-tick.C: - event, resp, upd, err := fn() - if err != nil { - ch <- steamEventResp{event: event, data: nil, err: errors.Wrap(err, "can't get stream data")} - return - } - if upd { - ch <- steamEventResp{event: event, data: resp, err: nil} - } - } - } - }() - return ch -} diff --git a/backend/app/rest/api/stream_test.go b/backend/app/rest/api/stream_test.go deleted file mode 100644 index dc20ebfe..00000000 --- a/backend/app/rest/api/stream_test.go +++ /dev/null @@ -1,63 +0,0 @@ -package api - -import ( - "bytes" - "context" - "fmt" - "testing" - "time" - - "github.com/stretchr/testify/assert" -) - -func TestStream_Timeout(t *testing.T) { - s := Streamer{ - Refresh: 10 * time.Millisecond, - TimeOut: 100 * time.Millisecond, - MaxActive: 10, - } - - eventFn := func() steamEventFn { - n := 0 - return func() (event string, data []byte, upd bool, err error) { - n++ - if n%2 == 0 || n > 10 { - return "test", nil, false, nil - } - return "test", []byte(fmt.Sprintf("some data %d\n", n)), true, nil - } - } - - buf := bytes.Buffer{} - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - err := s.Activate(ctx, eventFn, &buf) - assert.NoError(t, err) - assert.Equal(t, "event: test\ndata: some data 1\n\nevent: test\ndata: some data 3\n\nevent: test\ndata: some data 5\n\nevent: test\ndata: some data 7\n\nevent: test\ndata: some data 9\n\n", buf.String()) -} - -func TestStream_Cancel(t *testing.T) { - s := Streamer{ - Refresh: 10 * time.Millisecond, - TimeOut: 100 * time.Millisecond, - MaxActive: 10, - } - - eventFn := func() steamEventFn { - n := 0 - return func() (event string, data []byte, upd bool, err error) { - n++ - if n%2 == 0 { - return "test", nil, false, nil - } - return "test", []byte(fmt.Sprintf("some data %d\n", n)), true, nil - } - } - - buf := bytes.Buffer{} - ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) - defer cancel() - err := s.Activate(ctx, eventFn, &buf) - assert.NoError(t, err) - assert.Equal(t, "event: test\ndata: some data 1\n\nevent: test\ndata: some data 3\n\nevent: test\ndata: some data 5\n\nevent: test\ndata: some data 7\n\nevent: test\ndata: some data 9\n\n", buf.String()) -}