Close heartbeats in dispatcher when unknown task type (#57)

This will unblock the queue in case the task type is not handled by anything.

Author: rdner

pkg/queue/handlers/dispatcher.go

@@ -44,6 +44,7 @@ func (h *dispatchHandler) Process(ctx context.Context, task queue.Task, heartbea
 	handler, ok := h.handlers[task.Type]
 	if !ok {
 		logrus.Error("there is no handler for this task type")
+		close(heartbeats)
 		return ErrNoHandlerFound
 	}
 	return handler.Process(ctx, task, heartbeats)

pkg/queue/handlers/dispatcher_test.go

@@ -43,13 +43,19 @@ func TestDispatcherProcess(t *testing.T) {
 		require.Equal(t, "invalid", err.Error())
 	})
 
-	t.Run("returns ErrNoHandlerFound when there is no handler", func(t *testing.T) {
+	t.Run("returns ErrNoHandlerFound when there is no handler and closes heartbeats", 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)
+		heartbeats := make(chan queue.Progress)
+		err := h.Process(ctx, task, heartbeats)
+		<-heartbeats
+		require.PanicsWithError(t, "send on closed channel", func() {
+			heartbeats <- queue.Progress{}
+		})
 		require.Error(t, err)
 		require.Equal(t, ErrNoHandlerFound.Error(), err.Error())
+
 	})
 }