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

enclose streamer in rest's struct

Browse Source
This commit is contained in:
Umputun
2019-06-05 02:14:34 -05:00
parent a8b2dee85c
commit 20764b7b4e
7 changed files with 61 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/cmd/server.go
+5 -3
View File
@@ -312,9 +312,11 @@ func (s *ServerCommand) newServerApp() (*serverApp, error) {
SSLConfig: sslConfig,
UpdateLimiter: s.UpdateLimit,
ImageService: imageService,
StreamTimeOut: s.Stream.TimeOut,
StreamRefresh: s.Stream.RefreshInterval,
StreamMaxActive: s.Stream.MaxActive,
Streamer: &api.Streamer{
TimeOut: s.Stream.TimeOut,
Refresh: s.Stream.RefreshInterval,
MaxActive: int32(s.Stream.MaxActive),
},
}
srv.ScoreThresholds.Low, srv.ScoreThresholds.Critical = s.LowScore, s.CriticalScore
backend/app/rest/api/rest.go
+2 -9
View File
@@ -44,6 +44,7 @@ type Rest struct {
Migrator *Migrator
NotifyService *notify.Service
ImageService *image.Service
Streamer *Streamer
WebRoot string
RemarkURL string
@@ -55,10 +56,6 @@ type Rest struct {
}
UpdateLimiter float64
StreamTimeOut time.Duration
StreamRefresh time.Duration
StreamMaxActive int
SSLConfig SSLConfig
httpsServer *http.Server
httpServer *http.Server
@@ -337,11 +334,7 @@ func (s *Rest) controllerGroups() (public, private, admin, rss) {
commentFormatter: s.CommentFormatter,
readOnlyAge: s.ReadOnlyAge,
webRoot: s.WebRoot,
streamer: &streamer{
timeout: s.StreamTimeOut,
refresh: s.StreamRefresh,
maxActive: int32(s.StreamMaxActive),
},
streamer: s.Streamer,
}
privGrp := private{
backend/app/rest/api/rest_public.go
+4 -4
View File
@@ -29,8 +29,8 @@ type public struct {
readOnlyAge int
commentFormatter *store.CommentFormatter
imageService *image.Service
streamer *Streamer
webRoot string
streamer *streamer
}
type pubStore interface {
@@ -150,7 +150,7 @@ 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.streamer.timeout, s.streamer.refresh)
log.Printf("[DEBUG] start stream for %+v, timeout=%v, refresh=%v", locator, s.streamer.TimeOut, s.streamer.Refresh)
fn := func() steamEventFn {
lastTS := time.Time{}
@@ -179,7 +179,7 @@ func (s *public) infoStreamCtrl(w http.ResponseWriter, r *http.Request) {
}
}
if err := s.streamer.activate(r.Context(), fn, w); err != nil {
if err := s.streamer.Activate(r.Context(), fn, w); err != nil {
rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't stream", rest.ErrInternal)
}
}
@@ -252,7 +252,7 @@ func (s *public) lastCommentsStreamCtrl(w http.ResponseWriter, r *http.Request)
}
}
if err := s.streamer.activate(r.Context(), fn, w); err != nil {
if err := s.streamer.Activate(r.Context(), fn, w); err != nil {
rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't stream", rest.ErrInternal)
}
}
backend/app/rest/api/rest_public_test.go
+24 -25
View File
@@ -42,7 +42,6 @@ func TestRest_Preview(t *testing.T) {
assert.Nil(t, err)
assert.Equal(t, "<p>test 123</p>\n", string(b))
resp, err = post(t, ts.URL+"/api/v1/preview", "bad")
assert.Nil(t, err)
assert.Equal(t, 400, resp.StatusCode)
@@ -550,9 +549,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.streamer.refresh = 1 * time.Millisecond
srv.pubRest.streamer.timeout = 300 * time.Millisecond
srv.pubRest.streamer.maxActive = 100
srv.pubRest.streamer.Refresh = 1 * time.Millisecond
srv.pubRest.streamer.TimeOut = 300 * time.Millisecond
srv.pubRest.streamer.MaxActive = 100
postComment(t, ts.URL)
@@ -583,9 +582,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.streamer.refresh = 1 * time.Millisecond
srv.pubRest.streamer.timeout = 300 * time.Millisecond
srv.pubRest.streamer.maxActive = 10
srv.pubRest.streamer.Refresh = 1 * time.Millisecond
srv.pubRest.streamer.TimeOut = 300 * time.Millisecond
srv.pubRest.streamer.MaxActive = 10
postComment(t, ts.URL)
@@ -609,9 +608,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.streamer.refresh = 10 * time.Millisecond
srv.pubRest.streamer.timeout = 450 * time.Millisecond
srv.pubRest.streamer.maxActive = 100
srv.pubRest.streamer.Refresh = 10 * time.Millisecond
srv.pubRest.streamer.TimeOut = 450 * time.Millisecond
srv.pubRest.streamer.MaxActive = 100
postComment(t, ts.URL)
@@ -625,9 +624,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.streamer.refresh = 10 * time.Millisecond
srv.pubRest.streamer.timeout = 500 * time.Millisecond
srv.pubRest.streamer.maxActive = 100
srv.pubRest.streamer.Refresh = 10 * time.Millisecond
srv.pubRest.streamer.TimeOut = 500 * time.Millisecond
srv.pubRest.streamer.MaxActive = 100
postComment(t, ts.URL)
@@ -679,9 +678,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.streamer.refresh = 10 * time.Millisecond
srv.pubRest.streamer.timeout = 500 * time.Millisecond
srv.pubRest.streamer.maxActive = 100
srv.pubRest.streamer.Refresh = 10 * time.Millisecond
srv.pubRest.streamer.TimeOut = 500 * time.Millisecond
srv.pubRest.streamer.MaxActive = 100
postComment(t, ts.URL)
@@ -718,9 +717,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.streamer.refresh = 10 * time.Millisecond
srv.pubRest.streamer.timeout = 450 * time.Millisecond
srv.pubRest.streamer.maxActive = 100
srv.pubRest.streamer.Refresh = 10 * time.Millisecond
srv.pubRest.streamer.TimeOut = 450 * time.Millisecond
srv.pubRest.streamer.MaxActive = 100
postComment(t, ts.URL)
@@ -733,9 +732,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.streamer.refresh = 10 * time.Millisecond
srv.pubRest.streamer.timeout = 500 * time.Millisecond
srv.pubRest.streamer.maxActive = 100
srv.pubRest.streamer.Refresh = 10 * time.Millisecond
srv.pubRest.streamer.TimeOut = 500 * time.Millisecond
srv.pubRest.streamer.MaxActive = 100
postComment(t, ts.URL)
@@ -774,9 +773,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.streamer.refresh = 1 * time.Millisecond
srv.pubRest.streamer.timeout = 300 * time.Millisecond
srv.pubRest.streamer.maxActive = 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/rest_test.go
+5
View File
@@ -339,6 +339,11 @@ func startupT(t *testing.T) (ts *httptest.Server, srv *Rest, teardown func()) {
Cache: &cache.Nop{},
KeyStore: adminStore,
},
Streamer: &Streamer{
Refresh: 100 * time.Millisecond,
TimeOut: 5 * time.Second,
MaxActive: 100,
},
}
srv.ScoreThresholds.Low, srv.ScoreThresholds.Critical = -5, -10
backend/app/rest/api/stream.go
+11 -11
View File
@@ -11,11 +11,11 @@ import (
"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
// 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
}
@@ -27,15 +27,15 @@ type steamEventResp struct {
err error
}
// activate starts blocking function streaming update created by eventFn to ResponseWriter
// 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 {
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 {
if count > s.MaxActive {
return errors.New("too many streams")
}
@@ -44,7 +44,7 @@ func (s *streamer) activate(ctx context.Context, eventFn func() steamEventFn, w
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
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
@@ -65,10 +65,10 @@ func (s *streamer) activate(ctx context.Context, eventFn func() steamEventFn, w
}
// populate updates to chan, break on context close
func (s *streamer) eventsCh(ctx context.Context, fn steamEventFn) <-chan steamEventResp {
func (s *Streamer) eventsCh(ctx context.Context, fn steamEventFn) <-chan steamEventResp {
ch := make(chan steamEventResp)
go func() {
tick := time.NewTicker(s.refresh)
tick := time.NewTicker(s.Refresh)
defer func() {
close(ch)
tick.Stop()
backend/app/rest/api/stream_test.go
+10 -10
View File
@@ -11,10 +11,10 @@ import (
)
func TestStream_Timeout(t *testing.T) {
s := streamer{
refresh: 10 * time.Millisecond,
timeout: 100 * time.Millisecond,
maxActive: 10,
s := Streamer{
Refresh: 10 * time.Millisecond,
TimeOut: 100 * time.Millisecond,
MaxActive: 10,
}
eventFn := func() steamEventFn {
@@ -29,16 +29,16 @@ func TestStream_Timeout(t *testing.T) {
}
buf := bytes.Buffer{}
err := s.activate(context.Background(), eventFn, &buf)
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,
s := Streamer{
Refresh: 10 * time.Millisecond,
TimeOut: 100 * time.Millisecond,
MaxActive: 10,
}
eventFn := func() steamEventFn {
@@ -55,7 +55,7 @@ func TestStream_Cancel(t *testing.T) {
buf := bytes.Buffer{}
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
err := s.activate(ctx, eventFn, &buf)
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())
}