mirrors/remark42
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

generalize stream api, move to separate struct & file

Browse Source
This commit is contained in:
Umputun
2019-06-04 22:06:33 -05:00
parent e1bd0e5104
commit fe46509e8e
5 changed files with 231 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/rest/api/rest.go
+5 -3
View File
@@ -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{
backend/app/rest/api/rest_public.go
+48 -150
View File
@@ -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
}
backend/app/rest/api/rest_public_test.go
+25 -25
View File
@@ -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)
backend/app/rest/api/stream.go
+92
View File
@@ -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
}
backend/app/rest/api/stream_test.go
+61
View File
@@ -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())
}