Extend mq package (#155)

mq/consumer.go

@@ -17,9 +17,9 @@ const headerRemainingRetries = "x-remaining-retries"
 type consumer struct {
 	client *Client
 
-	queue   Queue
-	fn      func(Message) error
-	options *ConsumerOptions
+	queue            Queue
+	messageProcessor MessageProcessor
+	options          *ConsumerOptions
 
 	messages <-chan amqp.Delivery
 	stopChan chan struct{}
@@ -30,6 +30,18 @@ type Consumer interface {
 	Reconnect(ctx context.Context) error
 }
 
+type MessageProcessor interface {
+	Process(Message) error
+}
+
+// MessageProcessorFunc is an adapter to allow to use
+// an ordinary functions as mq MessageProcessor.
+type MessageProcessorFunc func(message Message) error
+
+func (f MessageProcessorFunc) Process(m Message) error {
+	return f(m)
+}
+
 func (c *consumer) Start(ctx context.Context) error {
 	c.stopChan = make(chan struct{})
 
@@ -117,7 +129,7 @@ func (c *consumer) process(queueName string, body []byte) error {
 	}
 
 	defer metric.Duration(metric.Start())
-	err := c.fn(body)
+	err := c.messageProcessor.Process(body)
 
 	if err != nil {
 		metric.Failure()

mq/mq.go

@@ -95,12 +95,12 @@ func (c *Client) InitExchange(name ExchangeName) Exchange {
 	}
 }
 
-func (c *Client) InitConsumer(queueName QueueName, options *ConsumerOptions, fn func(message Message) error) Consumer {
+func (c *Client) InitConsumer(queueName QueueName, options *ConsumerOptions, processor MessageProcessor) Consumer {
 	return &consumer{
-		client:  c,
-		queue:   c.InitQueue(queueName),
-		fn:      fn,
-		options: options,
+		client:           c,
+		queue:            c.InitQueue(queueName),
+		messageProcessor: processor,
+		options:          options,
 	}
 }