Add dispatch task handler (#42)

This handler allows to register other handlers and dispatch based on
type of the task in the queue.

Author: rdner

pkg/queue/handlers/dispatcher.go

@@ -0,0 +1,50 @@
+package handlers
+
+import (
+	"context"
+
+	"github.com/contiamo/go-base/pkg/queue"
+	"github.com/contiamo/go-base/pkg/queue/workers"
+	"github.com/contiamo/go-base/pkg/tracing"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+var (
+	// ErrNoHandlerFound occurs when dispatcher can'f find a registered handler for a task type
+	ErrNoHandlerFound = errors.New("no handler found")
+)
+
+// NewDispatchHandler creates a task handler that will dispatch tasks to other handlers
+func NewDispatchHandler(handlers map[queue.TaskType]workers.TaskHandler) workers.TaskHandler {
+	return &dispatchHandler{
+		Tracer:   tracing.NewTracer("handlers", "DispatchHandler"),
+		handlers: handlers,
+	}
+}
+
+type dispatchHandler struct {
+	tracing.Tracer
+	handlers map[queue.TaskType]workers.TaskHandler
+}
+
+func (h *dispatchHandler) Process(ctx context.Context, task queue.Task, heartbeats chan<- queue.Progress) (err error) {
+	span, ctx := h.StartSpan(ctx, "Process")
+	defer func() {
+		h.FinishSpan(span, err)
+	}()
+	span.SetTag("task.id", task.ID)
+	span.SetTag("task.queue", task.Queue)
+	span.SetTag("task.type", task.Type)
+	span.SetTag("task.spec", string(task.Spec))
+
+	logrus := logrus.WithField("type", task.Type)
+
+	logrus.Debug("dispatching task...")
+	handler, ok := h.handlers[task.Type]
+	if !ok {
+		logrus.Error("there is no handler for this task type")
+		return ErrNoHandlerFound
+	}
+	return handler.Process(ctx, task, heartbeats)
+}

pkg/queue/handlers/dispatcher_test.go

@@ -0,0 +1,55 @@
+package handlers
+
+import (
+	"context"
+	"errors"
+	"io/ioutil"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/contiamo/go-base/pkg/queue"
+	"github.com/contiamo/go-base/pkg/queue/workers"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/require"
+)
+
+type errHandler struct {
+	t       *testing.T
+	expTask queue.Task
+}
+
+func (h errHandler) Process(ctx context.Context, task queue.Task, heartbeats chan<- queue.Progress) (err error) {
+	require.Equal(h.t, h.expTask, task)
+	return errors.New("invalid")
+}
+
+func TestDispatcherProcess(t *testing.T) {
+	logrus.SetOutput(ioutil.Discard)
+	defer logrus.SetOutput(os.Stdout)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	t.Run("propagates the underlying handler's error", func(t *testing.T) {
+		task := queue.Task{TaskBase: queue.TaskBase{Type: "test"}}
+
+		h := NewDispatchHandler(map[queue.TaskType]workers.TaskHandler{
+			"test": errHandler{t: t, expTask: task},
+		})
+
+		err := h.Process(ctx, task, nil)
+		require.Error(t, err)
+		require.Equal(t, "invalid", err.Error())
+	})
+
+	t.Run("returns ErrNoHandlerFound when there is no handler", func(t *testing.T) {
+		task := queue.Task{TaskBase: queue.TaskBase{Type: "test"}}
+
+		h := NewDispatchHandler(map[queue.TaskType]workers.TaskHandler{})
+
+		err := h.Process(ctx, task, nil)
+		require.Error(t, err)
+		require.Equal(t, ErrNoHandlerFound.Error(), err.Error())
+	})
+}