Add Subscriber interface as a replacement for SubscribeFunc

daemon.go

@@ -45,23 +45,23 @@ type Daemon interface {
 
 // BaseDaemon is the parent structure for all daemons.
 type BaseDaemon struct {
-	self          Daemon
+	self         Daemon
-	name          string
+	name         string
-	queue         chan<- *task
+	queue        chan<- *task
-	logger        *log.Logger
+	logger       *log.Logger
-	panicHandler  PanicHandler
+	panicHandler PanicHandler
-	subscribeFunc SubscribeFunc
+	subscriber   Subscriber
-	publisher     Publisher
+	publisher    Publisher
-	shutdown      chan struct{}
+	shutdown     chan struct{}
-	limit         *ratelimit.Bucket
+	limit        *ratelimit.Bucket
 }
 
 // PanicHandler is a function that handles panics. Duh!
 type PanicHandler func(interface{})
 
 var (
-	errMissingSubscriptionFun = errors.New("subscription function is not set up")
+	errMissingSubscriber = errors.New("subscriber is not set up")
-	errMissingPublisher       = errors.New("publisher is not set up")
+	errMissingPublisher  = errors.New("publisher is not set up")
 )
 
 // Process creates a task and then adds it to processing queue.
@@ -92,11 +92,11 @@ func (d *BaseDaemon) SystemProcess(name string, a Actor) {
 func (d *BaseDaemon) Subscribe(topic string, fun interface{}) {
 	name := fmt.Sprintf("subscription for topic %q", topic)
 	d.SystemProcess(name, func() {
-		if d.subscribeFunc == nil {
+		if d.subscriber == nil {
-			panic(errMissingSubscriptionFun)
+			panic(errMissingSubscriber)
 		}
 
-		stream := d.subscribeFunc(d.String(), topic)
+		stream := d.subscriber.Subscribe(d.String(), topic)
 		defer stream.Close()
 
 		cf, err := caller.New(fun)

example/kafka/kafka.go

@@ -18,6 +18,9 @@ type ConsumerState struct {
 	Offset    int64 `json:"offset"`
 }
 
+// Subscriber is a dummy structure that implements satan.Subscriber interface.
+type Subscriber struct{}
+
 // Stream is an implementation of satan.Stremer for Kafka messaging queue.
 type Stream struct {
 	messages chan []byte
@@ -68,7 +71,7 @@ func Shutdown() {
 }
 
 // Subscribe creates a satan.Streamer implementation for Kafka messaging queue.
-func Subscribe(consumer, topic string) satan.Streamer {
+func (s Subscriber) Subscribe(consumer, topic string) satan.Streamer {
 	c, ok := consumers[consumer]
 	if !ok {
 		panic(fmt.Errorf("Consumer %q has no config", consumer))

satan.go

@@ -13,11 +13,11 @@ import (
 
 // Satan is the master daemon.
 type Satan struct {
-	SubscribeFunc SubscribeFunc
+	Subscriber  Subscriber
-	Publisher     Publisher
+	Publisher   Publisher
-	DaemonStats   stats.Publisher
+	DaemonStats stats.Publisher
-	Logger        *log.Logger
+	Logger      *log.Logger
-	NumWorkers    int
+	NumWorkers  int
 
 	daemons      []Daemon
 	queue        chan *task
@@ -32,8 +32,10 @@ type Satan struct {
 // Actor is a function that could be executed by daemon workers.
 type Actor func()
 
-// SubscribeFunc is a function that is used by daemons to subscribe to messages.
+// Subscriber is the interface that is used by daemons to subscribe to messages.
-type SubscribeFunc func(consumer, topic string) Streamer
+type Subscriber interface {
+	Subscribe(consumer, topic string) Streamer
+}
 
 // Streamer is the interface that wraps message consumers. Error handling
 // should be provided by the implementation. Feel free to panic.
@@ -78,7 +80,7 @@ func Summon() *Satan {
 func (s *Satan) AddDaemon(d Daemon) {
 	base := d.base()
 	base.self = d
-	base.subscribeFunc = s.SubscribeFunc
+	base.subscriber = s.Subscriber
 	base.publisher = s.Publisher
 	base.queue = s.queue
 	base.logger = s.Logger