add support of "since" param (unix msec) #336 #253

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

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"}}`)