From fe46509e8e10018946ad7e79378423c190d4876a Mon Sep 17 00:00:00 2001 From: Umputun Date: Tue, 4 Jun 2019 22:06:33 -0500 Subject: [PATCH] generalize stream api, move to separate struct & file --- backend/app/rest/api/rest.go | 8 +- backend/app/rest/api/rest_public.go | 198 ++++++----------------- backend/app/rest/api/rest_public_test.go | 50 +++--- backend/app/rest/api/stream.go | 92 +++++++++++ backend/app/rest/api/stream_test.go | 61 +++++++ 5 files changed, 231 insertions(+), 178 deletions(-) create mode 100644 backend/app/rest/api/stream.go create mode 100644 backend/app/rest/api/stream_test.go diff --git a/backend/app/rest/api/rest.go b/backend/app/rest/api/rest.go index 5029ba86..a23d6cfa 100644 --- a/backend/app/rest/api/rest.go +++ b/backend/app/rest/api/rest.go @@ -337,9 +337,11 @@ func (s *Rest) controllerGroups() (public, private, admin, rss) { commentFormatter: s.CommentFormatter, readOnlyAge: s.ReadOnlyAge, webRoot: s.WebRoot, - streamTimeOut: s.StreamTimeOut, - streamRefresh: s.StreamRefresh, - maxActiveStreams: int32(s.StreamMaxActive), + streamer: &streamer{ + timeout: s.StreamTimeOut, + refresh: s.StreamRefresh, + maxActive: int32(s.StreamMaxActive), + }, } privGrp := private{ diff --git a/backend/app/rest/api/rest_public.go b/backend/app/rest/api/rest_public.go index 53d68533..9361229c 100644 --- a/backend/app/rest/api/rest_public.go +++ b/backend/app/rest/api/rest_public.go @@ -1,7 +1,6 @@ package api import ( - "context" "crypto/sha1" // nolint "encoding/base64" "io" @@ -10,7 +9,6 @@ import ( "path" "strconv" "strings" - "sync/atomic" "time" "github.com/go-chi/chi" @@ -18,7 +16,6 @@ import ( log "github.com/go-pkgz/lgr" R "github.com/go-pkgz/rest" "github.com/go-pkgz/rest/cache" - "github.com/pkg/errors" "github.com/umputun/remark/backend/app/rest" "github.com/umputun/remark/backend/app/store" @@ -33,11 +30,7 @@ type public struct { commentFormatter *store.CommentFormatter imageService *image.Service webRoot string - streamTimeOut time.Duration - streamRefresh time.Duration - maxActiveStreams int32 - - activeStreamsCount int32 + streamer *streamer } type pubStore interface { @@ -157,66 +150,37 @@ 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.streamTimeOut, s.streamRefresh) + log.Printf("[DEBUG] start stream for %+v, timeout=%v, refresh=%v", locator, s.streamer.timeout, s.streamer.refresh) - lastTS := time.Time{} - lastCount := 0 - info := func() (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 + fn := func() steamEventFn { + lastTS := time.Time{} + lastCount := 0 + return func() (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 data, false, err } - // 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 data, false, err + + return data, upd, nil } - - return data, upd, nil } - count := atomic.AddInt32(&s.activeStreamsCount, 1) - defer atomic.AddInt32(&s.activeStreamsCount, -1) - if count > s.maxActiveStreams { - rest.SendErrorJSON(w, r, http.StatusTooManyRequests, errors.New("too many streams"), - "can't open new stream", rest.ErrActionRejected) - return - } - - updCh := s.eventsCh(r.Context(), info) - - for { - select { - case <-r.Context().Done(): // request closed by remote client - return - case <-time.After(s.streamTimeOut): // request closed by timeout - log.Printf("[DEBUG] info stream closed due to timeout") - return - case resp, ok := <-updCh: // new update - if !ok { // closed - return - } - if resp.err != nil { - rest.SendErrorJSON(w, r, http.StatusBadRequest, resp.err, "can't get post info", rest.ErrPostNotFound) - return - } - if _, e := w.Write(resp.data); e != nil { - log.Printf("[WARN] failed to send info stream, %v", e) - return - } - if fw, okFlush := w.(http.Flusher); okFlush { - fw.Flush() - } - } + if err := s.streamer.activate(r.Context(), fn, w); err != nil { + rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't stream", rest.ErrInternal) } } @@ -268,59 +232,29 @@ 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) - sinceTime := time.Now() - - info := func() (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 - } - if len(comments) > 0 { - sinceTime = comments[0].Timestamp - upd = true - } - sinceTime = time.Now() - return encodeJSONWithHTML(comments) - }) - return data, upd, err - } - - updCh := s.eventsCh(r.Context(), info) - - count := atomic.AddInt32(&s.activeStreamsCount, 1) - defer atomic.AddInt32(&s.activeStreamsCount, -1) - if count > s.maxActiveStreams { - rest.SendErrorJSON(w, r, http.StatusTooManyRequests, errors.New("too many streams"), - "can't open new stream", rest.ErrActionRejected) - return - } - - for { - select { - case <-r.Context().Done(): // request closed by remote client - return - case <-time.After(s.streamTimeOut): // request closed by timeout - log.Printf("[DEBUG] last comments stream closed due to timeout") - return - case resp, ok := <-updCh: // new update - if !ok { // closed - return - } - if resp.err != nil { - rest.SendErrorJSON(w, r, http.StatusInternalServerError, resp.err, "can't get last comments", rest.ErrInternal) - return - } - if _, e := w.Write(resp.data); e != nil { - log.Printf("[WARN] failed to send last comments stream, %v", e) - return - } - if fw, okFlush := w.(http.Flusher); okFlush { - fw.Flush() - } + fn := func() steamEventFn { + sinceTime := time.Now() + return func() (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 + } + if len(comments) > 0 { + sinceTime = comments[0].Timestamp + upd = true + } + sinceTime = time.Now() + return encodeJSONWithHTML(comments) + }) + return data, upd, err } } + + if err := s.streamer.activate(r.Context(), fn, w); err != nil { + rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't stream", rest.ErrInternal) + } } // GET /id/{id}?site=siteID&url=post-url - gets a comment by id @@ -544,39 +478,3 @@ func (s *public) applyView(comments []store.Comment, view string) []store.Commen } return comments } - -type eventFn func() (data []byte, upd bool, err error) - -type eventResp struct { - data []byte - err error -} - -// populate updates to chan, break on context close -func (s *public) eventsCh(ctx context.Context, fn eventFn) <-chan eventResp { - ch := make(chan eventResp) - go func() { - tick := time.NewTicker(s.streamRefresh) - defer func() { - close(ch) - tick.Stop() - }() - for { - select { - case <-ctx.Done(): // request closed by remote client - log.Printf("[DEBUG] stream closed by remote client, %v", ctx.Err()) - return - case <-tick.C: - resp, upd, err := fn() - if err != nil { - ch <- eventResp{data: nil, err: errors.Wrap(err, "can't get stream data")} - return - } - if upd { - ch <- eventResp{data: resp, err: nil} - } - } - } - }() - return ch -} diff --git a/backend/app/rest/api/rest_public_test.go b/backend/app/rest/api/rest_public_test.go index 0120d847..ecb7048e 100644 --- a/backend/app/rest/api/rest_public_test.go +++ b/backend/app/rest/api/rest_public_test.go @@ -532,9 +532,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.streamRefresh = 1 * time.Millisecond - srv.pubRest.streamTimeOut = 300 * time.Millisecond - srv.pubRest.maxActiveStreams = 100 + srv.pubRest.streamer.refresh = 1 * time.Millisecond + srv.pubRest.streamer.timeout = 300 * time.Millisecond + srv.pubRest.streamer.maxActive = 100 postComment(t, ts.URL) @@ -562,9 +562,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.streamRefresh = 1 * time.Millisecond - srv.pubRest.streamTimeOut = 300 * time.Millisecond - srv.pubRest.maxActiveStreams = 10 + srv.pubRest.streamer.refresh = 1 * time.Millisecond + srv.pubRest.streamer.timeout = 300 * time.Millisecond + srv.pubRest.streamer.maxActive = 10 postComment(t, ts.URL) @@ -588,9 +588,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.streamRefresh = 10 * time.Millisecond - srv.pubRest.streamTimeOut = 450 * time.Millisecond - srv.pubRest.maxActiveStreams = 100 + srv.pubRest.streamer.refresh = 10 * time.Millisecond + srv.pubRest.streamer.timeout = 450 * time.Millisecond + srv.pubRest.streamer.maxActive = 100 postComment(t, ts.URL) @@ -604,9 +604,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.streamRefresh = 10 * time.Millisecond - srv.pubRest.streamTimeOut = 500 * time.Millisecond - srv.pubRest.maxActiveStreams = 100 + srv.pubRest.streamer.refresh = 10 * time.Millisecond + srv.pubRest.streamer.timeout = 500 * time.Millisecond + srv.pubRest.streamer.maxActive = 100 postComment(t, ts.URL) @@ -658,9 +658,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.streamRefresh = 10 * time.Millisecond - srv.pubRest.streamTimeOut = 500 * time.Millisecond - srv.pubRest.maxActiveStreams = 100 + srv.pubRest.streamer.refresh = 10 * time.Millisecond + srv.pubRest.streamer.timeout = 500 * time.Millisecond + srv.pubRest.streamer.maxActive = 100 postComment(t, ts.URL) @@ -697,9 +697,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.streamRefresh = 10 * time.Millisecond - srv.pubRest.streamTimeOut = 450 * time.Millisecond - srv.pubRest.maxActiveStreams = 100 + srv.pubRest.streamer.refresh = 10 * time.Millisecond + srv.pubRest.streamer.timeout = 450 * time.Millisecond + srv.pubRest.streamer.maxActive = 100 postComment(t, ts.URL) @@ -712,9 +712,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.streamRefresh = 10 * time.Millisecond - srv.pubRest.streamTimeOut = 500 * time.Millisecond - srv.pubRest.maxActiveStreams = 100 + srv.pubRest.streamer.refresh = 10 * time.Millisecond + srv.pubRest.streamer.timeout = 500 * time.Millisecond + srv.pubRest.streamer.maxActive = 100 postComment(t, ts.URL) @@ -732,7 +732,7 @@ func TestRest_LastCommentsStreamCancel(t *testing.T) { client := http.Client{} req, err := http.NewRequest("GET", ts.URL+"/api/v1/stream/last?site=radio-t", nil) require.Nil(t, err) - ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond) + ctx, cancel := context.WithTimeout(context.Background(), 290*time.Millisecond) defer cancel() req = req.WithContext(ctx) r, err := client.Do(req) @@ -753,9 +753,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.streamRefresh = 1 * time.Millisecond - srv.pubRest.streamTimeOut = 300 * time.Millisecond - srv.pubRest.maxActiveStreams = 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/stream.go b/backend/app/rest/api/stream.go new file mode 100644 index 00000000..067d702c --- /dev/null +++ b/backend/app/rest/api/stream.go @@ -0,0 +1,92 @@ +package api + +import ( + "context" + "io" + "net/http" + "sync/atomic" + "time" + + log "github.com/go-pkgz/lgr" + "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 + + activeCount int32 +} + +type steamEventFn func() (data []byte, upd bool, err error) + +type steamEventResp struct { + data []byte + err error +} + +// activate starts blocking function streaming update created by eventFn to ResponseWriter +// canceled on context or inactivity timeout +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") + } + + 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 + } + if _, e := w.Write(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: + resp, upd, err := fn() + if err != nil { + ch <- steamEventResp{data: nil, err: errors.Wrap(err, "can't get stream data")} + return + } + if upd { + ch <- steamEventResp{data: resp, err: nil} + } + } + } + }() + return ch +} diff --git a/backend/app/rest/api/stream_test.go b/backend/app/rest/api/stream_test.go new file mode 100644 index 00000000..c32ca5a4 --- /dev/null +++ b/backend/app/rest/api/stream_test.go @@ -0,0 +1,61 @@ +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() (data []byte, upd bool, err error) { + n++ + if n%2 == 0 || n > 10 { + return nil, false, nil + } + return []byte(fmt.Sprintf("some data %d\n", n)), true, nil + } + } + + buf := bytes.Buffer{} + 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, + } + + eventFn := func() steamEventFn { + n := 0 + return func() (data []byte, upd bool, err error) { + n++ + if n%2 == 0 { + return nil, false, nil + } + return []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, "some data 1\nsome data 3\nsome data 5\nsome data 7\nsome data 9\n", buf.String()) +}