Detailed Explanation of Handler

In this page, I will explain the design of the Handler interface.

Handler Interface

The Handler you provide to the server is the core of your asynchronous task handling logic. The responsibility of the Handler is to accept a task and process it while considering the context. If processing fails, it should report any errors for later task retry.

The interface is defined as follows:

type Handler interface {
    ProcessTask(context.Context, *Task) error
}

This is a simple interface, succinctly describing the responsibilities of a Handler.

Implementation of the Interface

There are various ways to implement this handler interface.

Here’s an example of defining your own struct type to handle tasks:

type MyTaskHandler struct {
   // ... fields
}

// Implement the ProcessTask method
func (h *MyTaskHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
   // ... task handling logic
}

You can even define a function to satisfy the interface, thanks to the HandlerFunc adapter type.

func myHandler(ctx context.Context, t *asynq.Task) error {
    // ... task handling logic
}

// h satisfies the Handler interface
h := asynq.HandlerFunc(myHandler)

In most cases, you may need to check the input task’s Type and handle it accordingly.

func (h *MyTaskHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
   switch t.Type() {
   case "type1":
      // handle type1
   case "type2":
      // handle type2
   case "typeN":
      // handle typeN
   default:
      return fmt.Errorf("unexpected task type: %q", t.Type())
   }
}

As you can see, a handler can be composed of many different handlers. Each case in the example above can be handled by a dedicated handler. This is where the ServeMux type comes into play.

Using ServeMux

Note: You do not necessarily need to use the ServeMux type to implement a handler, but it can be very useful in many cases.
By using ServeMux, you can register multiple handlers. It will match the type of each task with the registered patterns and call the handler corresponding to the pattern closest to the task type name.

mux := asynq.NewServeMux()
mux.Handle("email:welcome", welcomeEmailHandler) // Register handler
mux.Handle("email:reminder", reminderEmailHandler)
mux.Handle("email:", defaultEmailHandler) // Default handler for other task types starting with "email:"

Using Middleware

If you need to execute some code before and/or after handling requests, you can achieve this with middleware. Middleware is a function that takes a Handler and returns a Handler.
Below is an example of middleware that logs the start and end of task processing.

func loggingMiddleware(h asynq.Handler) asynq.Handler {
    return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
        start := time.Now()
        log.Printf("Processing started for %q", t.Type())
        err := h.ProcessTask(ctx, t)
        if err != nil {
            return err
        }
        log.Printf("Processing completed for %q: elapsed time = %v", t.Type(), time.Since(start))
        return nil
    })
}

Now you can use this middleware to “wrap” your handler.

myHandler = loggingMiddleware(myHandler)

Additionally, if you are using ServeMux, you can provide middleware like this.

mux := NewServeMux()
mux.Use(loggingMiddleware)

Grouping Middleware

If you have a scenario where you want to apply a middleware to a group of tasks, you can achieve this by combining multiple ServeMux instances. One restriction is that each group of tasks needs to have the same prefix in their type names.

Example:
If you have tasks for handling orders and tasks for handling products, and you want to apply shared logic to all “product” tasks and another shared logic to all “order” tasks, you can do it like this:

productHandlers := asynq.NewServeMux()
productHandlers.Use(productMiddleware) // Apply shared logic to all product tasks
productHandlers.HandleFunc("product:update", productUpdateTaskHandler)
// ... Register other "product" task handlers

orderHandlers := asynq.NewServeMux()
orderHandler.Use(orderMiddleware) // Apply shared logic to all order tasks
orderHandlers.HandleFunc("order:refund", orderRefundTaskHandler)
// ... Register other "order" task handlers

// Top-level handler
mux := asynq.NewServeMux()
mux.Use(someGlobalMiddleware) // Apply shared logic to all tasks
mux.Handle("product:", productHandlers)
mux.Handle("order:", orderHandlers)

if err := srv.Run(mux); err != nil {
    log.Fatal(err)
}