From 445aec860e7192919ee6f5c567a1b90d92649a51 Mon Sep 17 00:00:00 2001 From: Umputun Date: Tue, 18 Jun 2019 12:14:27 -0500 Subject: [PATCH] add support of "since" param (unix msec) #336 #253 --- backend/app/rest/api/rest_public.go | 31 +++++++++-- backend/app/rest/api/rest_public_test.go | 68 ++++++++++++++++++++++++ 2 files changed, 95 insertions(+), 4 deletions(-) diff --git a/backend/app/rest/api/rest_public.go b/backend/app/rest/api/rest_public.go index 8de7808d..ab61f920 100644 --- a/backend/app/rest/api/rest_public.go +++ b/backend/app/rest/api/rest_public.go @@ -147,14 +147,26 @@ func (s *public) infoCtrl(w http.ResponseWriter, r *http.Request) { } } -// GET /stream/info?site=siteID&url=post-url - get info stream about the post +// 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 := time.Time{} + since := r.URL.Query().Get("since") + if since != "" { + unixTS, e := strconv.ParseInt(since, 10, 64) + if e != nil { + rest.SendErrorJSON(w, r, http.StatusBadRequest, e, "can't translate since parameter", rest.ErrDecode) + return + } + sinceTs = time.Unix(unixTS/1000, 1000000*(unixTS%1000)) // since param in msec timestamp + } + fn := func() steamEventFn { - lastTS := time.Time{} + 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) { @@ -227,13 +239,24 @@ func (s *public) lastCommentsCtrl(w http.ResponseWriter, r *http.Request) { } } -// GET /stream/last?site=siteID& - stream of last comments last comments for the siteID, across all posts +// 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 := time.Time{} + since := r.URL.Query().Get("since") + if since != "" { + unixTS, e := strconv.ParseInt(since, 10, 64) + if e != nil { + rest.SendErrorJSON(w, r, http.StatusBadRequest, e, "can't translate since parameter", rest.ErrDecode) + return + } + sinceTs = time.Unix(unixTS/1000, 1000000*(unixTS%1000)) // since param in msec timestamp + } + fn := func() steamEventFn { - sinceTime := time.Now() + 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) { diff --git a/backend/app/rest/api/rest_public_test.go b/backend/app/rest/api/rest_public_test.go index 6ec3006c..03aa69e5 100644 --- a/backend/app/rest/api/rest_public_test.go +++ b/backend/app/rest/api/rest_public_test.go @@ -665,6 +665,35 @@ func TestRest_InfoStreamCancel(t *testing.T) { assert.True(t, strings.Contains(recs[1*3+1], `"count":3`), recs[1]) } +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 = 500 * time.Millisecond + srv.pubRest.streamer.MaxActive = 100 + + postComment(t, ts.URL) + + wg := sync.WaitGroup{} + wg.Add(1) + go func() { + defer wg.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=radio-t&url=https://radio-t.com/blah1&since=12345678") + assert.Equal(t, 200, code) + wg.Wait() + + t.Logf(string(body)) + recs := strings.Split(strings.TrimSuffix(string(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() @@ -801,6 +830,45 @@ func TestRest_LastCommentsStreamTooMany(t *testing.T) { assert.Equal(t, 200, code, "all streams closed, good to go again") } +func TestRest_LastCommentsStreamSince(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 + + postComment(t, ts.URL) + + defer teardown() + wg := sync.WaitGroup{} + wg.Add(1) + go func() { + defer wg.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=radio-t&since=123456", nil) + require.Nil(t, err) + r, err := client.Do(req) + require.Nil(t, err) + defer r.Body.Close() + body, err := ioutil.ReadAll(r.Body) + require.Nil(t, err) + assert.Equal(t, 200, r.StatusCode) + + wg.Wait() + t.Logf("headers: %+v", r.Header) + 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), "10 events, includes first record") + t.Logf("%v", recs) +} + func postComment(t *testing.T, url string) { resp, e := post(t, url+"/api/v1/comment", `{"text": "test 123", "locator":{"url": "https://radio-t.com/blah1", "site": "radio-t"}}`)