DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Improve scheduler pkg, add tests

Browse Source
This commit is contained in:
Denis Arh
2019-12-29 10:30:28 +01:00
parent 57990135a5
commit 052f4f9c3e
4 changed files with 239 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/scheduler/helpers.go
+27 -18
View File
@@ -8,11 +8,19 @@ import (
)
// OnInterval parses all given strings as crontab expressions (ii) and returns true if any of them matches current time
func OnInterval(ee ...string) bool {
func OnInterval(ii ...string) bool {
match, err := onInterval(now(), ii...)
if err != nil {
sentry.CaptureException(err)
}
return match
}
func onInterval(now time.Time, ii ...string) (bool, error) {
var (
// This function will likely to be called exactly on a minute (00 sec) or a few milliseconds after
// Round it up to the smallest unit that cronexpr package supports
currTime = now().Truncate(time.Second)
currTime = now.Truncate(time.Second)
// For cron expression reference we need to subtract 1ns
// this will cause Next() fn to include next nanosecond (if it matches)
@@ -20,41 +28,42 @@ func OnInterval(ee ...string) bool {
)
// At least one of the given expressions should match
for _, e := range ee {
exp, err := cronexpr.Parse(e)
for _, i := range ii {
exp, err := cronexpr.Parse(i)
if err != nil {
sentry.CaptureException(err)
return false
return false, err
}
if currTime.Equal(exp.Next(cronRef)) {
return true
}
return currTime.Equal(exp.Next(cronRef)), nil
}
return false
return false, nil
}
// OnTimestamp parses all given strings as RFC3339 timestamps and returns true if any of them matches current time
func OnTimestamp(tt ...string) bool {
match, err := onTimestamp(now(), tt...)
if err != nil {
sentry.CaptureException(err)
}
return match
}
func onTimestamp(now time.Time, tt ...string) (bool, error) {
var (
// This function will likely to be called exactly on a minute (00 sec) or a few milliseconds after
// Round it up to the smallest unit that cronexpr package supports
currTime = now().Truncate(time.Second)
currTime = now.Truncate(time.Second)
)
for _, t := range tt {
ts, err := time.Parse(time.RFC3339, t)
if err != nil {
sentry.CaptureException(err)
return false
return false, err
}
if currTime.Equal(ts.Round(time.Second)) {
return true
}
return currTime.Equal(ts.Round(time.Second)), nil
}
return false
return false, nil
}
pkg/scheduler/helpers_test.go
+141
View File
@@ -0,0 +1,141 @@
package scheduler
import (
"testing"
"time"
"github.com/stretchr/testify/assert"
)
func TestOnInterval(t *testing.T) {
cases := []struct {
name string
now string
ii []string
match bool
err bool
}{
{
"empty",
time.Now().Format(time.RFC3339),
[]string{},
false,
false,
},
{
"nil",
time.Now().Format(time.RFC3339),
nil,
false,
false,
},
{
"next minute",
"2019-10-10T10:11:00Z",
[]string{"* * * * *"},
true,
false,
},
{
"every 5 minutes",
"2019-10-10T10:15:00Z",
[]string{"*/5 * * * *"},
true,
false,
},
{
"not full minute",
"2019-10-10T10:15:50Z",
[]string{"* * * * *"},
false,
false,
},
{
"invalid format",
"2019-10-10T10:15:50Z",
[]string{":P"},
false,
true,
},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
now, err := time.Parse(time.RFC3339, c.now)
assert.NoError(t, err)
match, err := onInterval(now, c.ii...)
if !c.err {
assert.NoError(t, err)
}
assert.Equal(t, c.match, match)
})
}
// touch exported func
OnInterval("* * * * *")
OnInterval(":P")
}
func TestOnTimestamp(t *testing.T) {
cases := []struct {
name string
now string
ii []string
match bool
err bool
}{
{
"empty",
time.Now().Format(time.RFC3339),
[]string{},
false,
false,
},
{
"nil",
time.Now().Format(time.RFC3339),
nil,
false,
false,
},
{
"must match",
"2019-10-10T10:11:00Z",
[]string{"2019-10-10T10:11:00Z"},
true,
false,
},
{
"must not match",
"2019-10-10T10:15:00Z",
[]string{"2020-10-10T10:15:00Z"},
false,
false,
},
{
"invalid format",
"2019-10-10T10:15:00Z",
[]string{":P"},
false,
true,
},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
now, err := time.Parse(time.RFC3339, c.now)
assert.NoError(t, err)
match, err := onTimestamp(now, c.ii...)
if !c.err {
assert.NoError(t, err)
}
assert.Equal(t, c.match, match)
})
}
// touch exported func
OnTimestamp("2019-10-10T10:15:00Z")
OnTimestamp(":P")
}
pkg/scheduler/service.go
+38 -18
View File
@@ -18,7 +18,7 @@ type (
dispatcher dispatcher
// Simple chan to control if service is running or not
active chan bool
ticker *time.Ticker
}
dispatcher interface {
@@ -28,6 +28,9 @@ type (
const (
defaultInterval = time.Minute
// There should not be more than 2 per each service (<no of services> * 2 [interval + timestamp])
maxEvents = 16
)
var (
@@ -40,8 +43,8 @@ var (
// Setup configures global scheduling service
func Setup(log *zap.Logger, d dispatcher, interval time.Duration) {
if gScheduler != nil {
// shut it down
gScheduler.active <- false
// shut down current global scheduler
gScheduler.Stop()
}
gScheduler = NewService(log, d, interval)
@@ -61,6 +64,7 @@ func NewService(log *zap.Logger, d dispatcher, interval time.Duration) *service
log: log.Named("scheduler"),
interval: interval,
dispatcher: d,
events: make([]eventbus.Event, 0, maxEvents),
}
return svc
@@ -71,19 +75,31 @@ func (svc *service) OnTick(events ...eventbus.Event) {
svc.events = append(svc.events, events...)
}
func (svc *service) Stop() {
if svc.ticker == nil {
svc.log.Debug("already stopped")
} else {
svc.log.Debug("stopping")
svc.ticker.Stop()
svc.ticker = nil
}
}
// Run starts event scheduler service
func (svc service) Start(ctx context.Context) {
if svc.active != nil {
func (svc *service) Start(ctx context.Context) {
if svc.ticker != nil {
svc.log.Debug("already started")
return
}
svc.active = make(chan bool, 1)
svc.ticker = &time.Ticker{}
go func() {
defer sentry.Recover()
nextTick := now().Truncate(svc.interval).Add(svc.interval)
svc.log.Info(
svc.log.Debug(
"starting",
zap.Time("delay", nextTick),
zap.Duration("interval", svc.interval),
@@ -91,30 +107,34 @@ func (svc service) Start(ctx context.Context) {
// Wait until start of the next interval
time.Sleep(nextTick.Sub(now()))
svc.log.Info("started")
svc.log.Debug("started")
// start with first interval
svc.dispatch(ctx)
ticker := time.NewTicker(svc.interval)
defer ticker.Stop()
defer svc.log.Info("stopped")
svc.ticker = time.NewTicker(svc.interval)
loop:
for {
select {
case <-svc.active:
svc.log.Info("unactivated")
return
case <-ctx.Done():
svc.log.Info("done")
return
case <-ticker.C:
case <-svc.ticker.C:
svc.dispatch(ctx)
case <-ctx.Done():
svc.log.Debug("done")
break loop
}
}
defer svc.log.Debug("stopped")
svc.ticker.Stop()
svc.ticker = nil
}()
}
func (svc service) Started() (started bool) {
return svc.ticker != nil
}
func (svc service) dispatch(ctx context.Context) {
for _, ev := range svc.events {
go func(ev eventbus.Event) {
pkg/scheduler/service_test.go
+33
View File
@@ -0,0 +1,33 @@
package scheduler
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
"github.com/cortezaproject/corteza-server/pkg/eventbus"
)
func TestMainServiceFunctions(t *testing.T) {
r := require.New(t)
const (
loopInterval = time.Millisecond
actionWait = loopInterval * 10
)
r.Nil(gScheduler)
Setup(zap.NewNop(), eventbus.New(), time.Second)
r.NotNil(gScheduler)
r.False(gScheduler.Started())
r.Equal(gScheduler, Service())
Service().Start(context.Background())
time.Sleep(actionWait)
r.True(gScheduler.Started())
gScheduler.Stop()
time.Sleep(actionWait)
r.False(gScheduler.Started())
}