1
0
Fork 0
shezmu/daemon.go

210 lines
5.1 KiB
Go
Raw Normal View History

2016-01-24 16:49:57 +00:00
package shezmu
2015-10-14 00:01:47 +00:00
2015-10-14 00:02:22 +00:00
import (
2015-10-17 00:40:17 +00:00
"errors"
2015-10-14 00:02:22 +00:00
"fmt"
"strings"
"time"
2015-10-17 00:40:17 +00:00
2015-10-17 02:33:46 +00:00
"github.com/juju/ratelimit"
2016-01-24 16:49:57 +00:00
"github.com/localhots/shezmu/caller"
2015-10-14 00:02:22 +00:00
)
2015-10-14 00:01:47 +00:00
// Daemon is the interface that contains a set of methods required to be
// implemented in order to be treated as a daemon.
type Daemon interface {
2015-10-14 00:29:44 +00:00
// Startup implementation should:
2015-10-14 00:01:47 +00:00
//
// func (d *DaemonName) Startup() {
2015-10-14 00:29:44 +00:00
// // 1. Set up a panic handler
2015-10-14 00:01:47 +00:00
// b.HandlePanics(func() {
// log.Error("Oh, crap!")
// })
//
// // 2. If the daemon is doing some IO it is a good idea to limit the
// // rate of its execution
// b.LimitRate(10, 1 * time.Second)
//
// // 3. If the daemon is also a consumer we need to subscribe for
2015-10-14 00:29:44 +00:00
// // topics that would be consumed by the daemon
2015-10-14 00:01:47 +00:00
// b.Subscribe("ProductPriceUpdates", func(p PriceUpdate) {
// log.Printf("Price for %q is now $%.2f", p.Product, p.Amount)
// })
// }
Startup()
// Shutdown implementation should clean up all daemon related stuff:
// close channels, process the last batch of items, etc.
Shutdown()
2016-01-24 22:39:38 +00:00
// String returns the name of a daemon.
String() string
2015-10-14 00:01:47 +00:00
// base is a (hack) function that allows the Daemon interface to reference
// underlying BaseDaemon structure.
base() *BaseDaemon
}
2015-10-14 00:02:22 +00:00
// BaseDaemon is the parent structure for all daemons.
type BaseDaemon struct {
self Daemon
name string
queue chan<- *task
2016-01-17 13:42:05 +00:00
logger Logger
panicHandler PanicHandler
subscriber Subscriber
publisher Publisher
shutdown chan struct{}
limit *ratelimit.Bucket
2015-10-14 00:02:22 +00:00
}
2015-10-18 00:22:07 +00:00
// PanicHandler is a function that handles panics. Duh!
type PanicHandler func(interface{})
2015-10-17 00:40:17 +00:00
var (
errMissingSubscriber = errors.New("subscriber is not set up")
errMissingPublisher = errors.New("publisher is not set up")
2015-10-17 00:40:17 +00:00
)
2015-10-14 00:02:22 +00:00
// Process creates a task and then adds it to processing queue.
func (d *BaseDaemon) Process(a Actor) {
2015-10-17 02:33:46 +00:00
if d.limit != nil {
d.limit.Wait(1)
}
d.tryEnqueue(&task{
daemon: d.self,
actor: a,
createdAt: time.Now(),
})
}
// SystemProcess creates a system task that is restarted in case of failure
// and then adds it to processing queue.
func (d *BaseDaemon) SystemProcess(name string, a Actor) {
d.tryEnqueue(&task{
daemon: d.self,
2015-10-14 00:02:22 +00:00
actor: a,
createdAt: time.Now(),
system: true,
name: name,
})
2015-10-14 00:02:22 +00:00
}
2015-10-17 00:40:17 +00:00
// Subscribe subscriasdsdfsdgdfgdfsg sdgsdfg sdfgs dfgdfgdfg.
func (d *BaseDaemon) Subscribe(topic string, fun interface{}) {
2015-10-24 00:07:47 +00:00
name := fmt.Sprintf("subscription for topic %q", topic)
d.SystemProcess(name, func() {
if d.subscriber == nil {
panic(errMissingSubscriber)
2015-10-17 00:40:17 +00:00
}
stream := d.subscriber.Subscribe(d.String(), topic)
2015-10-17 00:40:17 +00:00
defer stream.Close()
cf, err := caller.New(fun)
if err != nil {
panic(err)
}
for {
select {
case msg := <-stream.Messages():
d.Process(func() { cf.Call(msg) })
case <-d.shutdown:
return
}
}
})
}
// Publish sends a message to the publisher.
func (d *BaseDaemon) Publish(topic string, msg []byte, meta interface{}) {
2015-10-17 00:40:17 +00:00
if d.publisher == nil {
panic(errMissingPublisher)
}
d.publisher.Publish(topic, msg, meta)
2015-10-17 00:40:17 +00:00
}
2015-10-17 02:37:13 +00:00
// LimitRate limits the daemons' processing rate.
2015-10-17 02:33:46 +00:00
func (d *BaseDaemon) LimitRate(times int, per time.Duration) {
rate := float64(time.Second) / float64(per) * float64(times)
2015-10-17 02:37:13 +00:00
if rate <= 0 {
2016-01-17 13:42:05 +00:00
d.Logf("Daemon %s processing rate was limited to %d. Using 1 instead", d.base(), rate)
2015-10-17 02:33:46 +00:00
rate = 1.0
}
2016-01-17 13:42:05 +00:00
d.Logf("Daemon %s processing rate is limited to %.2f ops/s", d.base(), rate)
2015-10-17 02:33:46 +00:00
d.limit = ratelimit.NewBucketWithRate(rate, 1)
}
2015-10-14 00:02:22 +00:00
// HandlePanics sets up a panic handler function for the daemon.
2015-10-18 00:22:07 +00:00
func (d *BaseDaemon) HandlePanics(f PanicHandler) {
d.panicHandler = f
2015-10-14 00:02:22 +00:00
}
// ShutdownRequested returns a channel that is closed the moment daemon shutdown
// is requested.
func (d *BaseDaemon) ShutdownRequested() <-chan struct{} {
return d.shutdown
2015-10-14 00:02:22 +00:00
}
2015-10-15 23:27:03 +00:00
// Continue returns true if daemon should proceed and false if it should stop.
func (d *BaseDaemon) Continue() bool {
select {
case <-d.shutdown:
return false
2015-10-15 23:27:03 +00:00
default:
return true
}
2015-10-14 00:02:22 +00:00
}
2016-01-24 16:49:57 +00:00
// Log logs values using shezmu.Logger.Println function.
2015-10-24 16:25:16 +00:00
func (d *BaseDaemon) Log(v ...interface{}) {
2016-01-17 13:42:05 +00:00
if d.logger != nil {
d.logger.Println(v...)
}
2015-10-24 16:25:16 +00:00
}
2016-01-24 16:49:57 +00:00
// Logf logs values using shezmu.Logger.Printf function.
2015-10-24 16:25:16 +00:00
func (d *BaseDaemon) Logf(format string, v ...interface{}) {
2016-01-17 13:42:05 +00:00
if d.logger != nil {
d.logger.Printf(format, v...)
}
2015-10-24 16:25:16 +00:00
}
// Shutdown is the empty implementation of the daemons' Shutdown function that
// is inherited and used by default.
func (d *BaseDaemon) Shutdown() {}
2015-10-14 00:02:22 +00:00
// String returns the name of the Deamon unerlying struct.
func (d *BaseDaemon) String() string {
if d.name == "" {
d.name = strings.Split(fmt.Sprintf("%T", d.self), ".")[1]
2015-10-14 00:02:22 +00:00
}
return d.name
2015-10-14 00:02:22 +00:00
}
// base is a (hack) function that allows the Daemon interface to reference
// underlying BaseDaemon structure.
func (d *BaseDaemon) base() *BaseDaemon {
return d
2015-10-14 00:02:22 +00:00
}
func (d *BaseDaemon) tryEnqueue(t *task) {
defer func() {
if err := recover(); err != nil {
d.Logf("Failed to enqueue task %q due to process termination", t)
}
}()
d.queue <- t
}
2015-10-23 23:40:20 +00:00
func (d *BaseDaemon) handlePanic(err interface{}) {
if d.panicHandler != nil {
2015-10-18 00:22:07 +00:00
d.panicHandler(err)
2015-10-14 00:02:22 +00:00
}
}