Learn how to schedule an event to be running at specific time
Recently Product
team in my job requires me to have a scheduled event that might or might not run at a specific set of interval.
And the story goes like this:
As a [role] I want [resource] document to be tracked after it was
Assigned
to [role], for every [N] minutes until it wasHandled
So let's do a case study on Go Routine, and Go Channel!
Designing the Scheduler
- It have to be re-usable and shared across all projects
- It have to survive server restart
- It have to be able to re-schedule event
So now, we have these barebone requirements. We can start to write some code.
Event Object
First we have to write Event
object. It is the data that a Scheduler
receive
package scheduler
import (
"time"
)
// Attachment data associated with an event
// Can be anything stored in bytes
type Attachment struct {
Name string
ContentType string
Body []byte
}
// Event which will run on scheduler
type Event struct {
datetime string // RFC3339 please
attachments []Attachment
}
// NewEvent create a new instance of immutable Event
func NewEvent(d string, att []Attachment) *Event {
// we copy the attachment slice to another memory to avoid mutability
cpy := make([]Attachment, len(att))
copy(cpy, att)
return &Event{
datetime: d,
attachments: cpy,
}
}
// Date get event datetime, parsed into RFC3339 format
func (e *Event) Date() (time.Time, error) {
return time.Parse(time.RFC3339, e.datetime)
}
// Attachments returns a copy of attachments slice
// This is done to ensure immutability of event
func (e *Event) Attachments() []Attachment {
cpy := make([]Attachment, len(e.attachments))
copy(cpy, e.attachments)
return cpy
}
Event is designed to be immutable so none of these event can be changed once scheduled.
Each event contains 2 fields
-
datetime
= RFC3339 string representation of a date & time -
attachments
= List of any object associated to an Event
Both of those fields are private and can only be set from the NewEvent
factory.
But both can get fetched from 2 getter methods called Date
dan Attachments
.
The Scheduler
Next in the scheduler, we know that we have to create a method called Schedule
which takes an Event
as its parameter:
package scheduler
// Scheduler ...
type Scheduler struct {}
// Schedule an event
func (s *Scheduler) Schedule(e *Event) error {
// TODO:
}
TODO-1 Validate Event
// Scheduler error collection
var (
ErrEventInPast = errors.New("Event datetime is in the past")
ErrTimeInvalid = errors.New("Datetime format is not in RFC3339")
)
// Schedule an event
func (s *Scheduler) Schedule(e *Event) error {
date, err := e.Date()
if err != nil {
return ErrTimeInvalid
}
now := time.Now()
if date.Unix() <= now.Unix() {
return ErrEventInPast
}
...
}
TODO-2 Use Go routine to wait for event
// Schedule an event
func (s *Scheduler) Schedule(e *Event) error {
...
// fire a go routine
go func(e *Event) {
now := time.Now()
target, _ := e.Date()
waitDuration := target.Sub(now) // compare
select {
case <-time.After(waitDuration):
}
}(e)
}
TODO-3 Add Event Handler
This line below:
select {
case <-time.After(waitDuration):
}
Will block the goroutine for waitDuration
long, and then executes whatever code under the case
But we don't have the event handler handler for now, so it's time to design it.
- The event handler must be supplied by the caller
- Each
Scheduler
can only have 1 handler - Everytime an
Event
is triggered, we'll call the registered handler
The concept really is close with Java
delegates so that's what we'll call it.
// EventHandler delegates
type EventHandler func(*Scheduler, *Event)
// Scheduler ...
type Scheduler struct {
delegate EventHandler
}
// New instance of scheduler
func New(d EventHandler) *Scheduler {
return &Scheduler{
delegate: d,
}
}
Again, we make the Scheduler
immutable so the delegate
can only be instantiated via New
factory function, and can't be changed while it's running.
The EventHandler
delegate accepts 2 parameter, Scheduler
and Event
. This means any delegate
will always receive the correct reference to its Scheduler
and Event
.
And then, we complete our Schedule
method:
// Schedule an event
func (s *Scheduler) Schedule(e *Event) error {
date, err := e.Date()
if err != nil {
return ErrTimeInvalid
}
now := time.Now()
if date.Unix() <= now.Unix() {
return ErrEventInPast
}
// fire a go routine
go func(e *Event) {
now := time.Now()
target, _ := e.Date()
waitDuration := target.Sub(now) // compare
select {
case <-time.After(waitDuration):
s.delegate(s, e)
}
}(e)
return nil
}
Reviewing the Requirements
It looks like we're done with the Scheduler
but let's go back and look at the requirements:
- It have to survive server restart
So, we'll at least have to consider Stop
method which stops all scheduled events, collects it, and report it back to caller:
// Stop all running scheduler and report all pending events
func (s *Scheduler) Stop() (events []*Event) {
// TODO:
}
TODO-1 Stop all scheduled events!
So in the select case
statement in Schedule
method. Not only we have to wait for Event
datetime, we also have to listen to another channel which signals as out Stop
event. So lets call it just that:
// Scheduler ...
type Scheduler struct {
delegate EventHandler
stop chan struct{}
}
// New instance of scheduler
func New(d EventHandler) *Scheduler {
return &Scheduler{
delegate: d,
// initialize stop channel
stop: make(chan struct{}),
}
}
In the Schedule
method we'll listen to the stop
channel:
// Schedule an event
func (s *Scheduler) Schedule(e *Event) error {
...
// fire a go routine
go func(e *Event) {
now := time.Now()
target, _ := e.Date()
waitDuration := target.Sub(now) // compare
select {
case <-time.After(waitDuration):
s.delegate(s, e)
case <-s.stop:
// TODO:
}
}(e)
return nil
}
And then in the Stop
method:
// Stop all running scheduler and report all pending events
func (s *Scheduler) Stop() (events []*Event) {
// close stop channel & it will be broadcasted to all consumer
close(s.stop)
// TODO:
}
TODO-2 Collecting pending events
So far we have orchestrated to stop all running go routine but we haven't collected all the pending events yet.
To do this, we'll once again utilize channel. Let's call it pendings
// Scheduler ...
type Scheduler struct {
delegate EventHandler
stop chan struct{}
pendings chan *Event
}
// New instance of scheduler
func New(d EventHandler) *Scheduler {
return &Scheduler{
delegate: d,
// initialize stop channel
stop: make(chan struct{}),
// initialize buffered event channel
pendings: make(chan *Event, 3),
}
}
pendings
is a buffered channel with length = 3 which means it can hold up to 3 value until producer have to wait for any consumer to fetch a value, freeing up a buffer.
// Schedule an event
func (s *Scheduler) Schedule(e *Event) error {
...
// fire a go routine
go func(e *Event) {
...
select {
...
case <-s.stop:
s.pendings <- e
}
}(e)
return nil
}
And then in the Stop
function we'll have to collect all pending Events
being put in pendings
channel
// Stop all running scheduler and report all pending events
func (s *Scheduler) Stop() (events []*Event) {
close(s.stop)
for e := range s.pendings {
events = append(events, e)
}
return events
}
Looks good? but wait!
for e := range s.pendings
Is iterating over a channel.
BUT NOBODY is CLOSING THE pendings
CHANNEL SO IT WON'T EVER QUIT!
TODO-3 Closing the pendings channel
So how do we know when to close pendings
channel? We'll need to:
- Manually count how many go routine was fired.
- Decreement it everytime the go routine exit. Either it is done handling event, or forcefully stopped.
- Watch the counter to go down to zero, to then close the
pendings
channel
!@#E!T#T!V@$Y
Luckily, we have sync.WaitGroup
to our rescue! It does everything we listed above so let's code right away:
// Scheduler ...
type Scheduler struct {
delegate EventHandler
stop chan struct{}
pendings chan *Event
wg *sync.WaitGroup
}
// New instance of scheduler
func New(d EventHandler) *Scheduler {
return &Scheduler{
delegate: d,
// initialize stop channel
stop: make(chan struct{}),
// initialize buffered event channel
pendings: make(chan *Event, 3),
wg: &sync.WaitGroup{},
}
}
Next, we want to call wg.Add(1)
every time we fire a go routine.
And We want to call wg.Done()
every time a go routine exits.
// Schedule an event
func (s *Scheduler) Schedule(e *Event) error {
...
s.wg.Add(1)
// fire a go routine
go func(e *Event) {
...
defer s.wg.Done()
select {
...
case <-s.stop:
s.pendings <- e
}
}(e)
return nil
}
Lastly! we want to watch and wait for the counter to go down to zero.
// Stop all running scheduler and report all pending events
func (s *Scheduler) Stop() (events []*Event) {
...
go func() {
s.wg.Wait()
close(s.pendings)
}()
...
}
In the Stop
method, we are waiting for wg
counter to drop to zero by calling wg.Wait()
.
And we do it in another go routine so it doesn't block the Stop execution which collects pending events.
Reviewing the Requirements again
- It have to be re-usable and shared across all projects
This is a standalone go package
called scheduler
and can be shared to whoever needed a scheduler
- It have to survive server restart
Actually because scheduler
is a standalone go package
we'll invert the responsibilities of persisting pending events to the user / caller.
All scheduler
cant provide is just a Stop
method that can be called which reports any pending events to the caller.
- It have to be able to re-schedule event
By designing the EventHandler
delegate to accept Scheduler
and Event
as its parameter. We leave the re-schedule implementation to the delegate.
The delegate can just call Scheduler.Schedule
method to re-schedule any event.
Full Code of the Scheduler
package scheduler
import (
"errors"
"sync"
"time"
)
// Scheduler error collection
var (
ErrEventInPast = errors.New("Event datetime is in the past")
ErrTimeInvalid = errors.New("Datetime format is not in RFC3339")
)
// EventHandler delegates
type EventHandler func(*Scheduler, *Event)
// Scheduler ...
type Scheduler struct {
delegate EventHandler
stop chan struct{}
pendings chan *Event
wg *sync.WaitGroup
}
// New instance of scheduler
func New(d EventHandler) *Scheduler {
return &Scheduler{
delegate: d,
// initialize stop channel
stop: make(chan struct{}),
// initialize buffered event channel
pendings: make(chan *Event, 3),
wg: &sync.WaitGroup{},
}
}
// Schedule an event
func (s *Scheduler) Schedule(e *Event) error {
date, err := e.Date()
if err != nil {
return ErrTimeInvalid
}
now := time.Now()
if date.Unix() <= now.Unix() {
return ErrEventInPast
}
s.wg.Add(1)
// fire a go routine
go func(e *Event) {
now := time.Now()
target, _ := e.Date()
waitDuration := target.Sub(now)
defer s.wg.Done()
select {
case <-time.After(waitDuration):
s.delegate(s, e)
case <-s.stop:
s.pendings <- e
}
}(e)
return nil
}
// Stop all running scheduler and report all pending events
func (s *Scheduler) Stop() (events []*Event) {
close(s.stop)
go func() {
s.wg.Wait()
close(s.pendings)
}()
for e := range s.pendings {
events = append(events, e)
}
return events
}
Testing our Scheduler
func Test_Scheduler(t *testing.T) {
one := time.Now().Add(1 * time.Second) // next 1 sec
two := one.Add(1 * time.Second) // next 2 sec
tri := two.Add(1 * time.Second) // next 3 sec
att := []Attachment{
{Name: "Here!"}, // we need this to test immutability
}
sch := New(func(s *Scheduler, e *Event) {
if e.attachments[0].Name == "THERE!" {
t.Error("Name should be Here! not THERE!")
}
})
ev1 := NewEvent(one.Format(time.RFC3339), att)
sch.Schedule(ev1)
ev2 := NewEvent(two.Format(time.RFC3339), att)
sch.Schedule(ev2)
ev3 := NewEvent(tri.Format(time.RFC3339), att)
sch.Schedule(ev3)
// Name changed after all events have been scheduled
att[0].Name = "THERE!"
// sleep for 2 secs to leave only 3rd event as pending
time.Sleep(2 * time.Second)
// stop scheduler and collect the pending events
pendings := sch.Stop()
if len(pendings) != 1 {
t.Error("Pendings should only left the last event")
} else {
last := pendings[0]
if last != ev3 {
t.Error("Pendings[0] should equal to ev3")
}
}
}
And test result shows:
Running tool: go test -timeout 30s -run ^(Test_Scheduler)$
PASS
ok /Go/pkg/scheduler 2.512s
Success: Tests passed.
Hope you guys enjoy this case study ✌️
Top comments (0)