Update the broker interface

broker/broker.go

@@ -4,28 +4,35 @@ type Broker interface {
 	Address() string
 	Connect() error
 	Disconnect() error
-	Init() error
+	Init(...Option) error
-	Publish(string, *Message) error
+	Publish(string, *Message, ...PublishOption) error
-	Subscribe(string, Handler) (Subscriber, error)
+	Subscribe(string, Handler, ...SubscribeOption) (Subscriber, error)
 	String() string
 }
 
-type Handler func(*Message)
+// Handler is used to process messages via a subscription of a topic.
+// The handler is passed a publication interface which contains the
+// message and optional Ack method to acknowledge receipt of the message.
+type Handler func(Publication) error
 
 type Message struct {
 	Header map[string]string
 	Body   []byte
 }
 
+// Publication is given to a subscription handler for processing
+type Publication interface {
+	Topic() string
+	Message() *Message
+	Ack() error
+}
+
 type Subscriber interface {
+	Config() SubscribeOptions
 	Topic() string
 	Unsubscribe() error
 }
 
-type options struct{}
-
-type Option func(*options)
-
 var (
 	DefaultBroker Broker = newHttpBroker([]string{})
 )
@@ -34,8 +41,8 @@ func NewBroker(addrs []string, opt ...Option) Broker {
 	return newHttpBroker(addrs, opt...)
 }
 
-func Init() error {
+func Init(opts ...Option) error {
-	return DefaultBroker.Init()
+	return DefaultBroker.Init(opts...)
 }
 
 func Connect() error {
@@ -46,12 +53,12 @@ func Disconnect() error {
 	return DefaultBroker.Disconnect()
 }
 
-func Publish(topic string, msg *Message) error {
+func Publish(topic string, msg *Message, opts ...PublishOption) error {
-	return DefaultBroker.Publish(topic, msg)
+	return DefaultBroker.Publish(topic, msg, opts...)
 }
 
-func Subscribe(topic string, handler Handler) (Subscriber, error) {
+func Subscribe(topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
-	return DefaultBroker.Subscribe(topic, handler)
+	return DefaultBroker.Subscribe(topic, handler, opts...)
 }
 
 func String() string {

broker/http_broker.go

@@ -17,6 +17,10 @@ import (
 	"github.com/pborman/uuid"
 )
 
+// HTTP Broker is a placeholder for actual message brokers.
+// This should not really be used in production but useful
+// in developer where you want zero dependencies.
+
 type httpBroker struct {
 	id          string
 	address     string
@@ -29,6 +33,7 @@ type httpBroker struct {
 }
 
 type httpSubscriber struct {
+	opts  SubscribeOptions
 	id    string
 	topic string
 	ch    chan *httpSubscriber
@@ -36,6 +41,11 @@ type httpSubscriber struct {
 	svc   *registry.Service
 }
 
+type httpPublication struct {
+	m *Message
+	t string
+}
+
 var (
 	DefaultSubPath = "/_sub"
 )
@@ -55,6 +65,22 @@ func newHttpBroker(addrs []string, opt ...Option) Broker {
 	}
 }
 
+func (h *httpPublication) Ack() error {
+	return nil
+}
+
+func (h *httpPublication) Message() *Message {
+	return h.m
+}
+
+func (h *httpPublication) Topic() string {
+	return h.t
+}
+
+func (h *httpSubscriber) Config() SubscribeOptions {
+	return h.opts
+}
+
 func (h *httpSubscriber) Topic() string {
 	return h.topic
 }
@@ -150,9 +176,10 @@ func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	p := &httpPublication{m: m, t: topic}
 	h.RLock()
 	for _, subscriber := range h.subscribers[topic] {
-		subscriber.fn(m)
+		subscriber.fn(p)
 	}
 	h.RUnlock()
 }
@@ -169,7 +196,7 @@ func (h *httpBroker) Disconnect() error {
 	return h.stop()
 }
 
-func (h *httpBroker) Init() error {
+func (h *httpBroker) Init(opts ...Option) error {
 	if len(h.id) == 0 {
 		h.id = "broker-" + uuid.NewUUID().String()
 	}
@@ -178,7 +205,7 @@ func (h *httpBroker) Init() error {
 	return nil
 }
 
-func (h *httpBroker) Publish(topic string, msg *Message) error {
+func (h *httpBroker) Publish(topic string, msg *Message, opts ...PublishOption) error {
 	s, err := registry.GetService("topic:" + topic)
 	if err != nil {
 		return err
@@ -201,7 +228,9 @@ func (h *httpBroker) Publish(topic string, msg *Message) error {
 	return nil
 }
 
-func (h *httpBroker) Subscribe(topic string, handler Handler) (Subscriber, error) {
+func (h *httpBroker) Subscribe(topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+	opt := newSubscribeOptions(opts...)
+
 	// parse address for host, port
 	parts := strings.Split(h.Address(), ":")
 	host := strings.Join(parts[:len(parts)-1], ":")
@@ -220,6 +249,7 @@ func (h *httpBroker) Subscribe(topic string, handler Handler) (Subscriber, error
 	}
 
 	subscriber := &httpSubscriber{
+		opts:  opt,
 		id:    uuid.NewUUID().String(),
 		topic: topic,
 		ch:    h.unsubscribe,

broker/options.go

@@ -0,0 +1,47 @@
+package broker
+
+type Options struct{}
+
+type PublishOptions struct{}
+
+type SubscribeOptions struct {
+	// AutoAck defaults to true
+	AutoAck bool
+	// NumHandlers defaults to 1
+	NumHandlers int
+}
+
+type Option func(*Options)
+
+type PublishOption func(*PublishOptions)
+
+type SubscribeOption func(*SubscribeOptions)
+
+// DisableAutoAck will disable auto acking of messages
+// after they have been handled.
+func DisableAutoAck() SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.AutoAck = false
+	}
+}
+
+// NumHandlers sets the number of concurrent handlers to create
+// for a subscriber.
+func NumHandlers(i int) SubscribeOption {
+	return func(o *SubscribeOptions) {
+		o.NumHandlers = i
+	}
+}
+
+func newSubscribeOptions(opts ...SubscribeOption) SubscribeOptions {
+	opt := SubscribeOptions{
+		AutoAck:     true,
+		NumHandlers: 1,
+	}
+
+	for _, o := range opts {
+		o(&opt)
+	}
+
+	return opt
+}