2015-10-14 01:11:29 +00:00
|
|
|
package satan
|
|
|
|
|
|
|
|
import (
|
|
|
|
"log"
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
)
|
|
|
|
|
|
|
|
// Satan is the master daemon.
|
|
|
|
type Satan struct {
|
2015-10-17 00:41:21 +00:00
|
|
|
SubscribeFunc SubscribeFunc
|
|
|
|
Publisher Publisher
|
|
|
|
|
2015-10-14 01:11:29 +00:00
|
|
|
daemons []Daemon
|
|
|
|
queue chan *task
|
2015-10-17 00:41:21 +00:00
|
|
|
shutdown chan struct{}
|
2015-10-14 01:11:29 +00:00
|
|
|
wg sync.WaitGroup
|
|
|
|
latency *statistics
|
|
|
|
}
|
|
|
|
|
2015-10-15 23:07:04 +00:00
|
|
|
// Actor is a function that could be executed by daemon workers.
|
|
|
|
type Actor func()
|
|
|
|
|
2015-10-17 00:41:21 +00:00
|
|
|
// SubscribeFunc is a function that is used by daemons to subscribe to messages.
|
|
|
|
type SubscribeFunc func(consumer, topic string) Streamer
|
|
|
|
|
|
|
|
// Streamer is the interface that wraps message consumers. Error handling
|
|
|
|
// should be provided by the implementation. Feel free to panic.
|
|
|
|
type Streamer interface {
|
|
|
|
Messages() <-chan []byte
|
|
|
|
Close()
|
|
|
|
}
|
|
|
|
|
|
|
|
// Publisher is the interface that wraps message publishers. Error handling
|
|
|
|
// should be provided by the implementation. Feel free to panic.
|
|
|
|
type Publisher interface {
|
|
|
|
Publish(msg []byte)
|
|
|
|
Close()
|
|
|
|
}
|
|
|
|
|
2015-10-14 01:11:29 +00:00
|
|
|
type task struct {
|
|
|
|
daemon Daemon
|
|
|
|
actor Actor
|
|
|
|
createdAt time.Time
|
2015-10-15 23:07:04 +00:00
|
|
|
system bool
|
2015-10-14 01:11:29 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
const (
|
|
|
|
defaultNumWorkers = 10
|
|
|
|
)
|
|
|
|
|
|
|
|
// Summon creates a new instance of Satan.
|
|
|
|
func Summon() *Satan {
|
|
|
|
return &Satan{
|
|
|
|
queue: make(chan *task),
|
|
|
|
latency: newStatistics(),
|
|
|
|
shutdown: make(chan struct{}),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// AddDaemon adds a new daemon.
|
|
|
|
func (s *Satan) AddDaemon(d Daemon) {
|
2015-10-17 00:41:21 +00:00
|
|
|
base := d.base()
|
|
|
|
base.self = d
|
|
|
|
base.subscribeFunc = s.SubscribeFunc
|
|
|
|
base.publisher = s.Publisher
|
|
|
|
base.queue = s.queue
|
|
|
|
base.shutdown = make(chan struct{})
|
|
|
|
base.stats = newStatistics()
|
2015-10-14 01:11:29 +00:00
|
|
|
|
2015-10-17 00:41:21 +00:00
|
|
|
go d.Startup()
|
2015-10-14 01:11:29 +00:00
|
|
|
s.daemons = append(s.daemons, d)
|
|
|
|
}
|
|
|
|
|
|
|
|
// StartDaemons starts all registered daemons.
|
|
|
|
func (s *Satan) StartDaemons() {
|
|
|
|
s.wg.Add(defaultNumWorkers)
|
|
|
|
for i := 0; i < defaultNumWorkers; i++ {
|
|
|
|
go func(i int) {
|
|
|
|
s.runWorker(i)
|
|
|
|
s.wg.Done()
|
|
|
|
}(i)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// StopDaemons stops all running daemons.
|
|
|
|
func (s *Satan) StopDaemons() {
|
|
|
|
for _, d := range s.daemons {
|
|
|
|
close(d.base().shutdown)
|
|
|
|
d.Shutdown()
|
|
|
|
|
|
|
|
log.Printf("%s daemon performace statistics:\n%s\n",
|
|
|
|
d.base(), d.base().stats.snapshot())
|
|
|
|
}
|
2015-10-15 23:07:04 +00:00
|
|
|
close(s.shutdown)
|
2015-10-14 01:11:29 +00:00
|
|
|
s.wg.Wait()
|
2015-10-15 23:07:04 +00:00
|
|
|
close(s.queue)
|
2015-10-14 01:11:29 +00:00
|
|
|
|
|
|
|
log.Printf("Task processing latency statistics:\n%s\n", s.latency.snapshot())
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *Satan) runWorker(i int) {
|
|
|
|
log.Printf("Starting worker #%d", i+1)
|
|
|
|
defer log.Printf("Worker #%d has stopped", i+1)
|
|
|
|
|
2015-10-15 23:28:40 +00:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case t := <-s.queue:
|
|
|
|
dur := time.Now().UnixNano() - t.createdAt.UnixNano()
|
|
|
|
s.latency.add(time.Duration(dur))
|
|
|
|
|
|
|
|
// log.Printf("Daemon #%d got some job to do!", i+1)
|
|
|
|
s.processTask(t)
|
|
|
|
default:
|
|
|
|
select {
|
|
|
|
case <-s.shutdown:
|
|
|
|
return
|
|
|
|
default:
|
|
|
|
}
|
|
|
|
}
|
2015-10-14 01:11:29 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *Satan) processTask(t *task) {
|
|
|
|
defer t.daemon.base().handlePanic()
|
|
|
|
start := time.Now()
|
|
|
|
|
2015-10-17 01:31:47 +00:00
|
|
|
t.actor() // <--- THE ACTION HAPPENS HERE
|
2015-10-14 01:11:29 +00:00
|
|
|
|
|
|
|
dur := time.Now().UnixNano() - start.UnixNano()
|
|
|
|
t.daemon.base().stats.add(time.Duration(dur))
|
|
|
|
}
|