Clean unused stream code
It was supposed to solve #253 but frontend part for it in #357 was never finished, and backend code produces false positive test failures since day 0. The cost of just having this code around is too high, we'll re-add it in case frontend implementation will be finished.
This commit is contained in:
Dmitry Verkhoturov
+15
-28
@@ -51,7 +51,6 @@ type ServerCommand struct {
|
||||
SMTP SMTPGroup `group:"smtp" namespace:"smtp" env-namespace:"SMTP"`
|
||||
Image ImageGroup `group:"image" namespace:"image" env-namespace:"IMAGE"`
|
||||
SSL SSLGroup `group:"ssl" namespace:"ssl" env-namespace:"SSL"`
|
||||
Stream StreamGroup `group:"stream" namespace:"stream" env-namespace:"STREAM"`
|
||||
ImageProxy ImageProxyGroup `group:"image-proxy" namespace:"image-proxy" env-namespace:"IMAGE_PROXY"`
|
||||
|
||||
Sites []string `long:"site" env:"SITE" default:"remark" description:"site names" env-delim:","`
|
||||
@@ -219,13 +218,6 @@ type SSLGroup struct {
|
||||
ACMEEmail string `long:"acme-email" env:"ACME_EMAIL" description:"admin email for certificate notifications"`
|
||||
}
|
||||
|
||||
// StreamGroup define options for streaming apis
|
||||
type StreamGroup struct {
|
||||
RefreshInterval time.Duration `long:"refresh" env:"REFRESH" default:"5s" description:"refresh interval for streams"`
|
||||
TimeOut time.Duration `long:"timeout" env:"TIMEOUT" default:"15m" description:"timeout to close streams on inactivity"`
|
||||
MaxActive int `long:"max" env:"MAX" default:"500" description:"max number of parallel streams"`
|
||||
}
|
||||
|
||||
// RPCGroup defines options for remote modules (plugins)
|
||||
type RPCGroup struct {
|
||||
API string `long:"api" env:"API" description:"rpc extension api url"`
|
||||
@@ -431,26 +423,21 @@ func (s *ServerCommand) newServerApp() (*serverApp, error) {
|
||||
}
|
||||
|
||||
srv := &api.Rest{
|
||||
Version: s.Revision,
|
||||
DataService: dataService,
|
||||
WebRoot: s.WebRoot,
|
||||
RemarkURL: s.RemarkURL,
|
||||
ImageProxy: imgProxy,
|
||||
CommentFormatter: commentFormatter,
|
||||
Migrator: migr,
|
||||
ReadOnlyAge: s.ReadOnlyAge,
|
||||
SharedSecret: s.SharedSecret,
|
||||
Authenticator: authenticator,
|
||||
Cache: loadingCache,
|
||||
NotifyService: notifyService,
|
||||
SSLConfig: sslConfig,
|
||||
UpdateLimiter: s.UpdateLimit,
|
||||
ImageService: imageService,
|
||||
Streamer: &api.Streamer{
|
||||
TimeOut: s.Stream.TimeOut,
|
||||
Refresh: s.Stream.RefreshInterval,
|
||||
MaxActive: int32(s.Stream.MaxActive),
|
||||
},
|
||||
Version: s.Revision,
|
||||
DataService: dataService,
|
||||
WebRoot: s.WebRoot,
|
||||
RemarkURL: s.RemarkURL,
|
||||
ImageProxy: imgProxy,
|
||||
CommentFormatter: commentFormatter,
|
||||
Migrator: migr,
|
||||
ReadOnlyAge: s.ReadOnlyAge,
|
||||
SharedSecret: s.SharedSecret,
|
||||
Authenticator: authenticator,
|
||||
Cache: loadingCache,
|
||||
NotifyService: notifyService,
|
||||
SSLConfig: sslConfig,
|
||||
UpdateLimiter: s.UpdateLimit,
|
||||
ImageService: imageService,
|
||||
EmailNotifications: emailNotifications,
|
||||
EmojiEnabled: s.EnableEmoji,
|
||||
AnonVote: s.AnonymousVote && s.RestrictVoteIP,
|
||||
|
||||
@@ -89,5 +89,10 @@ func waitForHTTPServerStart(port int) {
|
||||
}
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
goleak.VerifyTestMain(m, goleak.IgnoreTopFunction("github.com/umputun/remark42/backend/app.init.0.func1"))
|
||||
// both ignores are for leaks which are detected locally
|
||||
goleak.VerifyTestMain(
|
||||
m,
|
||||
goleak.IgnoreTopFunction("github.com/umputun/remark42/backend/app.init.0.func1"),
|
||||
goleak.IgnoreTopFunction("net/http.(*Server).Shutdown"),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -46,7 +46,6 @@ type Rest struct {
|
||||
Migrator *Migrator
|
||||
NotifyService *notify.Service
|
||||
ImageService *image.Service
|
||||
Streamer *Streamer
|
||||
|
||||
AnonVote bool
|
||||
WebRoot string
|
||||
@@ -256,14 +255,6 @@ func (s *Rest) routes() chi.Router {
|
||||
|
||||
})
|
||||
|
||||
// open routes, streams, no send timeout
|
||||
rapi.Route("/stream", func(rstream chi.Router) {
|
||||
rstream.Use(tollbooth_chi.LimitHandler(tollbooth.NewLimiter(10, nil)))
|
||||
rstream.Use(authMiddleware.Trace, middleware.NoCache, logInfoWithBody)
|
||||
rstream.Get("/info", s.pubRest.infoStreamCtrl)
|
||||
rstream.Get("/last", s.pubRest.lastCommentsStreamCtrl)
|
||||
})
|
||||
|
||||
// open routes, cached
|
||||
rapi.Group(func(ropen chi.Router) {
|
||||
ropen.Use(middleware.Timeout(30 * time.Second))
|
||||
@@ -359,7 +350,6 @@ func (s *Rest) controllerGroups() (public, private, admin, rss) {
|
||||
commentFormatter: s.CommentFormatter,
|
||||
readOnlyAge: s.ReadOnlyAge,
|
||||
webRoot: s.WebRoot,
|
||||
streamer: s.Streamer,
|
||||
}
|
||||
|
||||
privGrp := private{
|
||||
|
||||
@@ -31,7 +31,6 @@ type public struct {
|
||||
readOnlyAge int
|
||||
commentFormatter *store.CommentFormatter
|
||||
imageService *image.Service
|
||||
streamer *Streamer
|
||||
webRoot string
|
||||
}
|
||||
|
||||
@@ -159,49 +158,6 @@ func (s *public) infoCtrl(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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, err := s.parseSince(r)
|
||||
if err != nil {
|
||||
rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't translate since parameter", rest.ErrDecode)
|
||||
return
|
||||
}
|
||||
|
||||
fn := func() steamEventFn {
|
||||
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) {
|
||||
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 "info", data, false, err
|
||||
}
|
||||
return "info", data, upd, nil
|
||||
}
|
||||
}
|
||||
|
||||
if e := s.streamer.Activate(r.Context(), fn, w); e != nil {
|
||||
rest.SendErrorJSON(w, r, http.StatusInternalServerError, e, "can't stream", rest.ErrInternal)
|
||||
}
|
||||
}
|
||||
|
||||
// GET /last/{limit}?site=siteID&since=unix_ts_msec - last comments for the siteID, across all posts, sorted by time, optionally
|
||||
// limited with "since" param
|
||||
func (s *public) lastCommentsCtrl(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -240,45 +196,6 @@ func (s *public) lastCommentsCtrl(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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, err := s.parseSince(r)
|
||||
if err != nil {
|
||||
rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't translate since parameter", rest.ErrDecode)
|
||||
return
|
||||
}
|
||||
if sinceTS.IsZero() {
|
||||
sinceTS = time.Now()
|
||||
}
|
||||
|
||||
fn := func() steamEventFn {
|
||||
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) {
|
||||
comments, e := s.dataService.Last(siteID, 1, sinceTime, rest.GetUserOrEmpty(r))
|
||||
if e != nil {
|
||||
return nil, e
|
||||
}
|
||||
sinceTime = time.Now()
|
||||
if len(comments) > 0 {
|
||||
sinceTime = comments[0].Timestamp
|
||||
upd = true
|
||||
}
|
||||
return encodeJSONWithHTML(comments)
|
||||
})
|
||||
return "last", data, upd, err
|
||||
}
|
||||
}
|
||||
|
||||
if e := s.streamer.Activate(r.Context(), fn, w); e != nil {
|
||||
rest.SendErrorJSON(w, r, http.StatusInternalServerError, e, "can't stream", rest.ErrInternal)
|
||||
}
|
||||
}
|
||||
|
||||
// GET /id/{id}?site=siteID&url=post-url - gets a comment by id
|
||||
func (s *public) commentByIDCtrl(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
|
||||
@@ -1,14 +1,11 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -580,145 +577,6 @@ func TestRest_Info(t *testing.T) {
|
||||
assert.Equal(t, 400, code)
|
||||
}
|
||||
|
||||
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 = 800 * time.Millisecond
|
||||
srv.pubRest.streamer.MaxActive = 100
|
||||
|
||||
postComment(t, ts.URL)
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(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=remark42&url=https://radio-t.com/blah1")
|
||||
assert.Equal(t, 200, code)
|
||||
<-done
|
||||
|
||||
recs := strings.Split(strings.TrimSuffix(body, "\n"), "\n")
|
||||
require.Equal(t, 10*3, len(recs), "10 records. each 2 lines +1 emty line")
|
||||
assert.True(t, strings.Contains(recs[0+1], `"count":2`), recs[0])
|
||||
assert.True(t, strings.Contains(recs[9*3+1], `"count":11`), recs[9])
|
||||
|
||||
_, code = get(t, ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah123")
|
||||
assert.Equal(t, 500, code)
|
||||
}
|
||||
|
||||
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
|
||||
|
||||
postComment(t, ts.URL)
|
||||
|
||||
var errsCount int32
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(20)
|
||||
for i := 0; i < 20; i++ {
|
||||
go func() {
|
||||
_, code := get(t, ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah1")
|
||||
if code == 429 {
|
||||
atomic.AddInt32(&errsCount, 1)
|
||||
}
|
||||
wg.Done()
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
assert.Equal(t, int32(10), atomic.LoadInt32(&errsCount), "10 streams rejected")
|
||||
}
|
||||
|
||||
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
|
||||
|
||||
postComment(t, ts.URL)
|
||||
|
||||
st := time.Now()
|
||||
_, code := get(t, ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah1")
|
||||
assert.Equal(t, 200, code)
|
||||
assert.True(t, time.Since(st) > time.Millisecond*450 && time.Since(st) < time.Millisecond*500, time.Since(st))
|
||||
}
|
||||
|
||||
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 = 1500 * time.Millisecond
|
||||
srv.pubRest.streamer.MaxActive = 100
|
||||
|
||||
postComment(t, ts.URL)
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
for i := 0; i < 5; i++ {
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
postComment(t, ts.URL)
|
||||
}
|
||||
}()
|
||||
|
||||
client := http.Client{}
|
||||
req, err := http.NewRequest("GET", ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah1", nil)
|
||||
require.NoError(t, err)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 1000*time.Millisecond)
|
||||
defer cancel()
|
||||
req = req.WithContext(ctx)
|
||||
r, err := client.Do(req)
|
||||
require.NoError(t, err)
|
||||
defer r.Body.Close()
|
||||
<-ctx.Done()
|
||||
<-done
|
||||
body, err := ioutil.ReadAll(r.Body)
|
||||
require.EqualError(t, err, "context deadline exceeded")
|
||||
assert.Equal(t, 200, r.StatusCode)
|
||||
|
||||
recs := strings.Count(string(body), "data:")
|
||||
require.Equal(t, 1, recs, "should have 1 event:\n", string(body))
|
||||
assert.Contains(t, string(body), `"count":2`)
|
||||
}
|
||||
|
||||
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 = 900 * time.Millisecond
|
||||
srv.pubRest.streamer.MaxActive = 100
|
||||
|
||||
postComment(t, ts.URL)
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
for i := 0; i < 10; i++ {
|
||||
time.Sleep(15 * time.Millisecond)
|
||||
postComment(t, ts.URL)
|
||||
}
|
||||
}()
|
||||
|
||||
body, code := get(t, ts.URL+"/api/v1/stream/info?site=remark42&url=https://radio-t.com/blah1&since=12345678")
|
||||
assert.Equal(t, 200, code)
|
||||
<-done
|
||||
recs := strings.Split(strings.TrimSuffix(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()
|
||||
@@ -730,198 +588,3 @@ func TestRest_Robots(t *testing.T) {
|
||||
"Allow: /api/v1/list\nAllow: /api/v1/config\nAllow: /api/v1/user\nAllow: /api/v1/img\n"+
|
||||
"Allow: /api/v1/avatar\nAllow: /api/v1/picture\n", body)
|
||||
}
|
||||
|
||||
func TestRest_LastCommentsStream(t *testing.T) {
|
||||
t.Skip() // FIXME: not in use currently and fails sometime. Should be fixed as we start to use stremeing for real
|
||||
ts, srv, teardown := startupT(t)
|
||||
srv.pubRest.readOnlyAge = 10000000 // make sure we don't hit read-only
|
||||
srv.pubRest.streamer.Refresh = 50 * time.Millisecond
|
||||
srv.pubRest.streamer.TimeOut = 500 * time.Millisecond
|
||||
srv.pubRest.streamer.MaxActive = 100
|
||||
|
||||
// stream endpoint currently relies on real cache being present
|
||||
cacheBackend, err := cache.NewExpirableCache()
|
||||
require.NoError(t, err)
|
||||
memCache := cache.NewScache(cacheBackend)
|
||||
defer memCache.Close()
|
||||
srv.privRest.cache = memCache
|
||||
srv.pubRest.cache = memCache
|
||||
|
||||
postComment(t, ts.URL)
|
||||
|
||||
defer teardown()
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
for i := 1; i < 10; i++ {
|
||||
postComment(t, ts.URL)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
t.Log("wrote 10 records")
|
||||
}()
|
||||
|
||||
client := http.Client{}
|
||||
req, err := http.NewRequest("GET", ts.URL+"/api/v1/stream/last?site=remark42", nil)
|
||||
require.NoError(t, err)
|
||||
r, err := client.Do(req)
|
||||
require.NoError(t, err)
|
||||
defer r.Body.Close()
|
||||
<-done
|
||||
body, err := ioutil.ReadAll(r.Body)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 200, r.StatusCode)
|
||||
|
||||
assert.Equal(t, "text/event-stream", r.Header.Get("content-type"))
|
||||
assert.Equal(t, "keep-alive", r.Header.Get("connection"))
|
||||
|
||||
recs := strings.Split(strings.TrimSuffix(string(body), "\n"), "\n")
|
||||
require.Equal(t, 9*3, len(recs), "9 events")
|
||||
assert.True(t, strings.Contains(recs[1], `test 123`), recs[1])
|
||||
}
|
||||
|
||||
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
|
||||
|
||||
postComment(t, ts.URL)
|
||||
|
||||
st := time.Now()
|
||||
_, code := get(t, ts.URL+"/api/v1/stream/last?site=remark42")
|
||||
assert.Equal(t, 200, code)
|
||||
assert.True(t, time.Since(st) > time.Millisecond*450 && time.Since(st) < time.Millisecond*500, time.Since(st))
|
||||
}
|
||||
|
||||
func TestRest_LastCommentsStreamCancel(t *testing.T) {
|
||||
t.Skip()
|
||||
|
||||
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
|
||||
|
||||
// stream endpoint currently relies on real cache being present
|
||||
cacheBackend, err := cache.NewExpirableCache()
|
||||
require.NoError(t, err)
|
||||
memCache := cache.NewScache(cacheBackend)
|
||||
defer memCache.Close()
|
||||
srv.privRest.cache = memCache
|
||||
srv.pubRest.cache = memCache
|
||||
|
||||
postComment(t, ts.URL)
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(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=remark42", nil)
|
||||
require.NoError(t, err)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 290*time.Millisecond)
|
||||
defer cancel()
|
||||
req = req.WithContext(ctx)
|
||||
r, err := client.Do(req)
|
||||
require.NoError(t, err)
|
||||
<-done
|
||||
defer r.Body.Close()
|
||||
body, err := ioutil.ReadAll(r.Body)
|
||||
require.EqualError(t, err, "context deadline exceeded")
|
||||
assert.Equal(t, 200, r.StatusCode)
|
||||
|
||||
recs := strings.Split(strings.TrimSuffix(string(body), "\n"), "\n")
|
||||
assert.True(t, len(recs) < 30, "less 10 events")
|
||||
}
|
||||
|
||||
func TestRest_LastCommentsStreamTooMany(t *testing.T) {
|
||||
t.Skip()
|
||||
|
||||
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
|
||||
|
||||
postComment(t, ts.URL)
|
||||
|
||||
var errsCount int32
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(20)
|
||||
for i := 0; i < 20; i++ {
|
||||
go func() {
|
||||
_, code := get(t, ts.URL+"/api/v1/stream/last?site=remark42")
|
||||
if code == 429 {
|
||||
atomic.AddInt32(&errsCount, 1)
|
||||
}
|
||||
wg.Done()
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
assert.Equal(t, int32(10), atomic.LoadInt32(&errsCount), "10 streams rejected")
|
||||
|
||||
_, code := get(t, ts.URL+"/api/v1/stream/last?site=remark42")
|
||||
assert.Equal(t, 200, code, "all streams closed, good to go again")
|
||||
}
|
||||
|
||||
func TestRest_LastCommentsStreamSince(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
|
||||
|
||||
// stream endpoint currently relies on real cache being present
|
||||
cacheBackend, err := cache.NewExpirableCache()
|
||||
require.NoError(t, err)
|
||||
memCache := cache.NewScache(cacheBackend)
|
||||
defer memCache.Close()
|
||||
srv.privRest.cache = memCache
|
||||
srv.pubRest.cache = memCache
|
||||
|
||||
postComment(t, ts.URL)
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
for i := 1; i < 10; i++ {
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
postComment(t, ts.URL)
|
||||
}
|
||||
}()
|
||||
|
||||
client := http.Client{}
|
||||
req, err := http.NewRequest("GET", ts.URL+"/api/v1/stream/last?site=remark42&since=123456", nil)
|
||||
require.NoError(t, err)
|
||||
r, err := client.Do(req)
|
||||
require.NoError(t, err)
|
||||
<-done
|
||||
defer r.Body.Close()
|
||||
body, err := ioutil.ReadAll(r.Body)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 200, r.StatusCode)
|
||||
|
||||
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), "should be 10 events, including first record:\n", recs)
|
||||
}
|
||||
|
||||
func postComment(t *testing.T, url string) {
|
||||
resp, err := post(t, url+"/api/v1/comment",
|
||||
`{"text": "test 123", "locator":{"url": "https://radio-t.com/blah1", "site": "remark42"}}`)
|
||||
require.NoError(t, err)
|
||||
b, err := ioutil.ReadAll(resp.Body)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, http.StatusCreated, resp.StatusCode, string(b))
|
||||
}
|
||||
|
||||
@@ -378,9 +378,9 @@ func startupT(t *testing.T) (ts *httptest.Server, srv *Rest, teardown func()) {
|
||||
SecretReader: token.SecretFunc(func(aud string) (string, error) { return "secret", nil }),
|
||||
AvatarStore: avatar.NewLocalFS(tmp + "/ava-remark42"),
|
||||
}),
|
||||
Cache: memCache,
|
||||
WebRoot: tmp,
|
||||
RemarkURL: "https://demo.remark42.com",
|
||||
Cache: memCache,
|
||||
WebRoot: tmp,
|
||||
RemarkURL: "https://demo.remark42.com",
|
||||
ImageService: image.NewService(&image.FileSystem{
|
||||
Location: tmp + "/pics-remark42",
|
||||
Partitions: 100,
|
||||
@@ -401,11 +401,6 @@ func startupT(t *testing.T) (ts *httptest.Server, srv *Rest, teardown func()) {
|
||||
Cache: memCache,
|
||||
KeyStore: astore,
|
||||
},
|
||||
Streamer: &Streamer{
|
||||
Refresh: 100 * time.Millisecond,
|
||||
TimeOut: 5 * time.Second,
|
||||
MaxActive: 100,
|
||||
},
|
||||
NotifyService: notify.NopService,
|
||||
EmojiEnabled: true,
|
||||
}
|
||||
|
||||
@@ -1,103 +0,0 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
log "github.com/go-pkgz/lgr"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
type steamEventFn func() (event string, data []byte, upd bool, err error)
|
||||
|
||||
type steamEventResp struct {
|
||||
data []byte
|
||||
event string
|
||||
err error
|
||||
}
|
||||
|
||||
// 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 {
|
||||
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")
|
||||
}
|
||||
|
||||
if ww, ok := w.(http.ResponseWriter); ok {
|
||||
ww.Header().Set("Content-Type", "text/event-stream")
|
||||
ww.Header().Set("Connection", "keep-alive")
|
||||
ww.Header().Set("Cache-Control", "no-cache")
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
// make server-sent event record
|
||||
// see https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events
|
||||
if _, e := fmt.Fprintf(w, "event: %s\ndata: %s\n", resp.event, string(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:
|
||||
event, resp, upd, err := fn()
|
||||
if err != nil {
|
||||
ch <- steamEventResp{event: event, data: nil, err: errors.Wrap(err, "can't get stream data")}
|
||||
return
|
||||
}
|
||||
if upd {
|
||||
ch <- steamEventResp{event: event, data: resp, err: nil}
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return ch
|
||||
}
|
||||
@@ -1,63 +0,0 @@
|
||||
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() (event string, data []byte, upd bool, err error) {
|
||||
n++
|
||||
if n%2 == 0 || n > 10 {
|
||||
return "test", nil, false, nil
|
||||
}
|
||||
return "test", []byte(fmt.Sprintf("some data %d\n", n)), true, nil
|
||||
}
|
||||
}
|
||||
|
||||
buf := bytes.Buffer{}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
err := s.Activate(ctx, eventFn, &buf)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, "event: test\ndata: some data 1\n\nevent: test\ndata: some data 3\n\nevent: test\ndata: some data 5\n\nevent: test\ndata: some data 7\n\nevent: test\ndata: some data 9\n\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() (event string, data []byte, upd bool, err error) {
|
||||
n++
|
||||
if n%2 == 0 {
|
||||
return "test", nil, false, nil
|
||||
}
|
||||
return "test", []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, "event: test\ndata: some data 1\n\nevent: test\ndata: some data 3\n\nevent: test\ndata: some data 5\n\nevent: test\ndata: some data 7\n\nevent: test\ndata: some data 9\n\n", buf.String())
|
||||
}
|
||||
Reference in New Issue
Block a user