diff --git a/pkg/scheduler/service.go b/pkg/scheduler/service.go index c8f4ed2fb..098b3b760 100644 --- a/pkg/scheduler/service.go +++ b/pkg/scheduler/service.go @@ -2,6 +2,7 @@ package scheduler import ( "context" + "sync" "time" "go.uber.org/zap" @@ -17,6 +18,9 @@ type ( interval time.Duration dispatcher dispatcher + // Read & write locking + l *sync.RWMutex + // Simple chan to control if service is running or not ticker *time.Ticker } @@ -61,6 +65,7 @@ func NewService(log *zap.Logger, d dispatcher, interval time.Duration) *service } var svc = &service{ + l: &sync.RWMutex{}, log: log.Named("scheduler"), interval: interval, dispatcher: d, @@ -72,74 +77,102 @@ func NewService(log *zap.Logger, d dispatcher, interval time.Duration) *service // Register all events that should fire on tick (interval) func (svc *service) OnTick(events ...eventbus.Event) { + svc.l.Lock() + defer svc.l.Unlock() 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.l.Lock() svc.ticker = nil + defer svc.l.Unlock() } } // Run starts event scheduler service func (svc *service) Start(ctx context.Context) { + if svc.ticker != nil { svc.log.Debug("already started") return } + svc.l.Lock() svc.ticker = &time.Ticker{} + svc.l.Unlock() go func() { defer sentry.Recover() - nextTick := now().Truncate(svc.interval).Add(svc.interval) + // Calculate how much time we need to wait until next tick + delay := now().Truncate(svc.interval).Add(svc.interval).Sub(now()) svc.log.Debug( "starting", - zap.Time("delay", nextTick), + zap.Duration("delay", delay), zap.Duration("interval", svc.interval), ) // Wait until start of the next interval - time.Sleep(nextTick.Sub(now())) + time.Sleep(delay) + svc.l.Lock() + svc.ticker = time.NewTicker(svc.interval) + svc.l.Unlock() svc.log.Debug("started") - // start with first interval - svc.dispatch(ctx) - svc.ticker = time.NewTicker(svc.interval) - - loop: - for { - select { - 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 + go svc.watch(ctx) }() } +func (svc service) watch(ctx context.Context) { + defer sentry.Recover() + defer func() { + defer svc.log.Debug("stopped") + svc.ticker.Stop() + svc.l.Lock() + svc.ticker = nil + svc.l.Unlock() + }() + + // start with first interval + svc.dispatch(ctx) + + for { + select { + case <-svc.ticker.C: + svc.dispatch(ctx) + + case <-ctx.Done(): + svc.log.Debug("done") + return + } + } +} + func (svc service) Started() (started bool) { + svc.l.RLock() + defer svc.l.RUnlock() + return svc.ticker != nil } func (svc service) dispatch(ctx context.Context) { - for _, ev := range svc.events { - go func(ev eventbus.Event) { - sentry.Recover() - svc.dispatcher.Dispatch(ctx, ev) - }(ev) + svc.l.RLock() + + ee := make([]eventbus.Event, len(svc.events)) + for e := range svc.events { + ee[e] = svc.events[e] + } + + defer svc.l.RUnlock() + + for _, ev := range ee { + go svc.dispatcher.Dispatch(ctx, ev) } } diff --git a/pkg/scheduler/service_test.go b/pkg/scheduler/service_test.go index fcb60d310..bc04fea13 100644 --- a/pkg/scheduler/service_test.go +++ b/pkg/scheduler/service_test.go @@ -11,6 +11,30 @@ import ( "github.com/cortezaproject/corteza-server/pkg/eventbus" ) +type ( + mockEvent struct { + rType string + eType string + match func(name string, op string, values ...string) bool + } +) + +func (e mockEvent) ResourceType() string { + return e.rType +} + +func (e mockEvent) EventType() string { + return e.eType +} + +func (e mockEvent) Match(name string, op string, values ...string) bool { + if e.match == nil { + return true + } + + return e.match(name, op, values...) +} + func TestMainServiceFunctions(t *testing.T) { r := require.New(t) @@ -25,6 +49,7 @@ func TestMainServiceFunctions(t *testing.T) { r.False(gScheduler.Started()) r.Equal(gScheduler, Service()) Service().Start(context.Background()) + Service().OnTick(&mockEvent{}, &mockEvent{}, &mockEvent{}, &mockEvent{}) time.Sleep(actionWait) r.True(gScheduler.Started()) gScheduler.Stop()