v0.7.0

Changes:

- Added Unique option for clients
- Require go v.13 or above (to use new errors wrapping functions)
- Fixed adding queue key to all-queues set (asynq:queues) when scheduling.

.travis.yml

@@ -2,9 +2,7 @@ language: go
 go_import_path: github.com/hibiken/asynq
 git:
   depth: 1
-env:
-  - GO111MODULE=on # go modules are the default
-go: [1.12.x, 1.13.x, 1.14.x]
+go: [1.13.x, 1.14.x]
 script:
   - go test -race -v -coverprofile=coverage.txt -covermode=atomic ./...
 services:

CHANGELOG.md

@@ -7,6 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.7.0] - 2020-03-22
+
+### Changed
+
+- Support Go v1.13+, dropped support for go v1.12
+
+### Added
+
+- `Unique` option was added to allow client to enqueue a task only if it's unique within a certain time period.
+
+## [0.6.2] - 2020-03-15
+
+### Added
+
+- `Use` method was added to `ServeMux` to apply middlewares to all handlers.
+
 ## [0.6.1] - 2020-03-12
 
 ### Added

README.md

@@ -22,61 +22,133 @@ First, make sure you are running a Redis server locally.
 $ redis-server
 ```
 
-To create and schedule tasks, use `Client` and provide a task and when to enqueue the task.  
+Next, write a package that encapslates task creation and task handling.
+
+```go
+package tasks
+
+import (
+    "fmt"
+
+    "github.com/hibiken/asynq"
+)
+
+// A list of background task types.
+const (
+    EmailDelivery   = "email:deliver"
+    ImageProcessing = "image:process"
+)
+
+// Write function NewXXXTask to create a task.
+
+func NewEmailDeliveryTask(userID int, tmplID string) *asynq.Task {
+    payload := map[string]interface{}{"user_id": userID, "template_id": tmplID}
+    return asynq.NewTask(EmailDelivery, payload)
+}
+
+func NewImageProcessingTask(src, dst string) *asynq.Task {
+    payload := map[string]interface{}{"src": src, "dst": dst}
+    return asynq.NewTask(ImageProcessing, payload)
+}
+
+// Write function HandleXXXTask to handle the given task.
+// NOTE: It satisfies the asynq.HandlerFunc interface.
+
+func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
+    userID, err := t.Payload.GetInt("user_id")
+    if err != nil {
+        return err
+    }
+    tmplID, err := t.Payload.GetString("template_id")
+    if err != nil {
+        return err
+    }
+    fmt.Printf("Send Email to User: user_id = %d, template_id = %s\n", userID, tmplID)
+    // Email delivery logic ...
+    return nil
+}
+
+func HandleImageProcessingTask(ctx context.Context, t *asynq.Task) error {
+    src, err := t.Payload.GetString("src")
+    if err != nil {
+        return err
+    }
+    dst, err := t.Payload.GetString("dst")
+    if err != nil {
+        return err
+    }
+    fmt.Printf("Process image: src = %s, dst = %s\n", src, dst)
+    // Image processing logic ...
+    return nil
+}
+```
+
+In your web application code, import the above package and use [`Client`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Client) to enqueue tasks to the task queue.  
 A task will be processed by a background worker as soon as the task gets enqueued.  
 Scheduled tasks will be stored in Redis and will be enqueued at the specified time.
 
 ```go
+package main
+
+import (
+    "time"
+
+    "github.com/hibiken/asynq"
+    "your/app/package/tasks"
+)
+
+const redisAddr = "127.0.0.1:6379"
+
 func main() {
-    r := &asynq.RedisClientOpt{
-        Addr: "127.0.0.1:6379",
+    r := &asynq.RedisClientOpt{Addr: redisAddr}
+    c := asynq.NewClient(r)
+
+    // Example 1: Enqueue task to be processed immediately.
+
+    t := tasks.NewEmailDeliveryTask(42, "some:template:id")
+    err := c.Enqueue(t)
+    if err != nil {
+        log.Fatal("could not enqueue task: %v", err)
     }
 
-    client := asynq.NewClient(r)
 
-    // Create a task with task type and payload.
-    t1 := asynq.NewTask("email:signup", map[string]interface{}{"user_id": 42})
+    // Example 2: Schedule task to be processed in the future.
 
-    t2 := asynq.NewTask("email:reminder", map[string]interface{}{"user_id": 42})
+    t = tasks.NewEmailDeliveryTask(42, "other:template:id")
+    err = c.EnqueueIn(24*time.Hour, t)
+    if err != nil {
+        log.Fatal("could not schedule task: %v", err)
+    }
 
-    // Enqueue immediately.
-    err := client.Enqueue(t1)
 
-    // Enqueue 24 hrs later.
-    err = client.EnqueueIn(24*time.Hour, t2)
+    // Example 3: Pass options to tune task processing behavior.
+    // Options include MaxRetry, Queue, Timeout, Deadline, etc.
 
-    // Enqueue at specific time.
-    err = client.EnqueueAt(time.Date(2020, time.March, 6, 10, 0, 0, 0, time.UTC), t2)
-
-    // Pass vararg options to specify processing behavior for the given task.
-    //
-    // MaxRetry specifies the max number of retry if the task fails (Default is 25).
-    // Queue specifies which queue to enqueue this task to (Default is "default" queue).
-    // Timeout specifies the the task timeout (Default is no timeout).
-    err = client.Enqueue(t1, asynq.MaxRetry(10), asynq.Queue("critical"), asynq.Timeout(time.Minute))
+    t = tasks.NewImageProcessingTask("some/blobstore/url", "other/blobstore/url")
+    err = c.Enqueue(t, asynq.MaxRetry(10), asynq.Queue("critical"), asynq.Timeout(time.Minute))
+    if err != nil {
+        log.Fatal("could not enqueue task: %v", err)
+    }
 }
 ```
 
-To start the background workers, use `Background` and provide your `Handler` to process the tasks.
+Next, create a binary to process these tasks in the background.  
+To start the background workers, use [`Background`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Background) and provide your [`Handler`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Handler) to process the tasks.
 
-`Handler` is an interface with one method `ProcessTask` with the following signature.
+You can optionally use [`ServeMux`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#ServeMux) to create a handler, just as you would with [`"net/http"`](https://golang.org/pkg/net/http/) Handler.
 
 ```go
-// ProcessTask should return nil if the processing of a task is successful.
-//
-// If ProcessTask return a non-nil error or panics, the task will be retried after delay.
-type Handler interface {
-    ProcessTask(context.Context, *asynq.Task) error
-}
-```
+package main
 
-You can optionally use `ServeMux` to create a handler, just as you would with `"net/http"` Handler.
+import (
+    "github.com/hibiken/asynq"
+    "your/app/package/tasks"
+)
+
+const redisAddr = "127.0.0.1:6379"
 
-```go
 func main() {
-    r := &asynq.RedisClientOpt{
-        Addr: "127.0.0.1:6379",
-    }
+    r := &asynq.RedisClientOpt{Addr: redisAddr}
 
     bg := asynq.NewBackground(r, &asynq.Config{
         // Specify how many concurrent workers to use
@@ -92,23 +164,12 @@ func main() {
 
     // mux maps a type to a handler
     mux := asynq.NewServeMux()
-    mux.HandleFunc("email:signup", signupEmailHandler)
-    mux.HandleFunc("email:reminder", reminderEmailHandler)
+    mux.HandleFunc(tasks.EmailDelivery, tasks.HandleEmailDeliveryTask)
+    mux.HandleFunc(tasks.ImageProcessing, tasks.HandleImageProcessingTask)
     // ...register other handlers...
 
     bg.Run(mux)
 }
-
-// function with the same signature as the ProcessTask method for the Handler interface.
-func signupEmailHandler(ctx context.Context, t *asynq.Task) error {
-    id, err := t.Payload.GetInt("user_id")
-    if err != nil {
-        return err
-    }
-    fmt.Printf("Send welcome email to user %d\n", id)
-    // ...your email sending logic...
-    return nil
-}
 ```
 
 For a more detailed walk-through of the library, see our [Getting Started Guide](https://github.com/hibiken/asynq/wiki/Getting-Started).
@@ -144,7 +205,7 @@ go get -u github.com/hibiken/asynq/tools/asynqmon
 | Dependency                 | Version |
 | -------------------------- | ------- |
 | [Redis](https://redis.io/) | v2.8+   |
-| [Go](https://golang.org/)  | v1.12+  |
+| [Go](https://golang.org/)  | v1.13+  |
 
 ## Contributing
 

client.go

@@ -5,6 +5,9 @@
 package asynq
 
 import (
+	"errors"
+	"fmt"
+	"sort"
 	"strings"
 	"time"
 
@@ -38,6 +41,7 @@ type (
 	queueOption    string
 	timeoutOption  time.Duration
 	deadlineOption time.Time
+	uniqueOption   time.Duration
 )
 
 // MaxRetry returns an option to specify the max number of times
@@ -70,11 +74,30 @@ func Deadline(t time.Time) Option {
 	return deadlineOption(t)
 }
 
+// Unique returns an option to enqueue a task only if the given task is unique.
+// Task enqueued with this option is guaranteed to be unique within the given ttl.
+// Once the task gets processed successfully or once the TTL has expired, another task with the same uniqueness may be enqueued.
+// ErrDuplicateTask error is returned when enqueueing a duplicate task.
+//
+// Uniqueness of a task is based on the following properties:
+//     - Task Type
+//     - Task Payload
+//     - Queue Name
+func Unique(ttl time.Duration) Option {
+	return uniqueOption(ttl)
+}
+
+// ErrDuplicateTask indicates that the given task could not be enqueued since it's a duplicate of another task.
+//
+// ErrDuplicateTask error only applies to tasks enqueued with a Unique option.
+var ErrDuplicateTask = errors.New("task already exists")
+
 type option struct {
 	retry     int
 	queue     string
 	timeout   time.Duration
 	deadline  time.Time
+	uniqueTTL time.Duration
 }
 
 func composeOptions(opts ...Option) option {
@@ -94,6 +117,8 @@ func composeOptions(opts ...Option) option {
 			res.timeout = time.Duration(opt)
 		case deadlineOption:
 			res.deadline = time.Time(opt)
+		case uniqueOption:
+			res.uniqueTTL = time.Duration(opt)
 		default:
 			// ignore unexpected option
 		}
@@ -101,6 +126,39 @@ func composeOptions(opts ...Option) option {
 	return res
 }
 
+// uniqueKey computes the redis key used for the given task.
+// It returns an empty string if ttl is zero.
+func uniqueKey(t *Task, ttl time.Duration, qname string) string {
+	if ttl == 0 {
+		return ""
+	}
+	return fmt.Sprintf("%s:%s:%s", t.Type, serializePayload(t.Payload.data), qname)
+}
+
+func serializePayload(payload map[string]interface{}) string {
+	if payload == nil {
+		return "nil"
+	}
+	type entry struct {
+		k string
+		v interface{}
+	}
+	var es []entry
+	for k, v := range payload {
+		es = append(es, entry{k, v})
+	}
+	// sort entries by key
+	sort.Slice(es, func(i, j int) bool { return es[i].k < es[j].k })
+	var b strings.Builder
+	for _, e := range es {
+		if b.Len() > 0 {
+			b.WriteString(",")
+		}
+		b.WriteString(fmt.Sprintf("%s=%v", e.k, e.v))
+	}
+	return b.String()
+}
+
 const (
 	// Max retry count by default
 	defaultMaxRetry = 25
@@ -122,8 +180,18 @@ func (c *Client) EnqueueAt(t time.Time, task *Task, opts ...Option) error {
 		Retry:     opt.retry,
 		Timeout:   opt.timeout.String(),
 		Deadline:  opt.deadline.Format(time.RFC3339),
+		UniqueKey: uniqueKey(task, opt.uniqueTTL, opt.queue),
 	}
-	return c.enqueue(msg, t)
+	var err error
+	if time.Now().After(t) {
+		err = c.enqueue(msg, opt.uniqueTTL)
+	} else {
+		err = c.schedule(msg, t, opt.uniqueTTL)
+	}
+	if err == rdb.ErrDuplicateTask {
+		return fmt.Errorf("%w", ErrDuplicateTask)
+	}
+	return err
 }
 
 // Enqueue enqueues task to be processed immediately.
@@ -146,9 +214,17 @@ func (c *Client) EnqueueIn(d time.Duration, task *Task, opts ...Option) error {
 	return c.EnqueueAt(time.Now().Add(d), task, opts...)
 }
 
-func (c *Client) enqueue(msg *base.TaskMessage, t time.Time) error {
-	if time.Now().After(t) {
+func (c *Client) enqueue(msg *base.TaskMessage, uniqueTTL time.Duration) error {
+	if uniqueTTL > 0 {
+		return c.rdb.EnqueueUnique(msg, uniqueTTL)
+	}
 	return c.rdb.Enqueue(msg)
 }
+
+func (c *Client) schedule(msg *base.TaskMessage, t time.Time, uniqueTTL time.Duration) error {
+	if uniqueTTL > 0 {
+		ttl := t.Add(uniqueTTL).Sub(time.Now())
+		return c.rdb.ScheduleUnique(msg, t, ttl)
+	}
 	return c.rdb.Schedule(msg, t)
 }

client_test.go

@@ -5,10 +5,12 @@
 package asynq
 
 import (
+	"errors"
 	"testing"
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	h "github.com/hibiken/asynq/internal/asynqtest"
 	"github.com/hibiken/asynq/internal/base"
 )
@@ -44,8 +46,8 @@ func TestClientEnqueueAt(t *testing.T) {
 			processAt: now,
 			opts:      []Option{},
 			wantEnqueued: map[string][]*base.TaskMessage{
-				"default": []*base.TaskMessage{
-					&base.TaskMessage{
+				"default": {
+					{
 						Type:     task.Type,
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
@@ -129,8 +131,8 @@ func TestClientEnqueue(t *testing.T) {
 				MaxRetry(3),
 			},
 			wantEnqueued: map[string][]*base.TaskMessage{
-				"default": []*base.TaskMessage{
-					&base.TaskMessage{
+				"default": {
+					{
 						Type:     task.Type,
 						Payload:  task.Payload.data,
 						Retry:    3,
@@ -148,8 +150,8 @@ func TestClientEnqueue(t *testing.T) {
 				MaxRetry(-2),
 			},
 			wantEnqueued: map[string][]*base.TaskMessage{
-				"default": []*base.TaskMessage{
-					&base.TaskMessage{
+				"default": {
+					{
 						Type:     task.Type,
 						Payload:  task.Payload.data,
 						Retry:    0, // Retry count should be set to zero
@@ -168,8 +170,8 @@ func TestClientEnqueue(t *testing.T) {
 				MaxRetry(10),
 			},
 			wantEnqueued: map[string][]*base.TaskMessage{
-				"default": []*base.TaskMessage{
-					&base.TaskMessage{
+				"default": {
+					{
 						Type:     task.Type,
 						Payload:  task.Payload.data,
 						Retry:    10, // Last option takes precedence
@@ -187,8 +189,8 @@ func TestClientEnqueue(t *testing.T) {
 				Queue("custom"),
 			},
 			wantEnqueued: map[string][]*base.TaskMessage{
-				"custom": []*base.TaskMessage{
-					&base.TaskMessage{
+				"custom": {
+					{
 						Type:     task.Type,
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
@@ -206,8 +208,8 @@ func TestClientEnqueue(t *testing.T) {
 				Queue("HIGH"),
 			},
 			wantEnqueued: map[string][]*base.TaskMessage{
-				"high": []*base.TaskMessage{
-					&base.TaskMessage{
+				"high": {
+					{
 						Type:     task.Type,
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
@@ -225,8 +227,8 @@ func TestClientEnqueue(t *testing.T) {
 				Timeout(20 * time.Second),
 			},
 			wantEnqueued: map[string][]*base.TaskMessage{
-				"default": []*base.TaskMessage{
-					&base.TaskMessage{
+				"default": {
+					{
 						Type:     task.Type,
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
@@ -244,8 +246,8 @@ func TestClientEnqueue(t *testing.T) {
 				Deadline(time.Date(2020, time.June, 24, 0, 0, 0, 0, time.UTC)),
 			},
 			wantEnqueued: map[string][]*base.TaskMessage{
-				"default": []*base.TaskMessage{
-					&base.TaskMessage{
+				"default": {
+					{
 						Type:     task.Type,
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
@@ -324,8 +326,8 @@ func TestClientEnqueueIn(t *testing.T) {
 			delay: 0,
 			opts:  []Option{},
 			wantEnqueued: map[string][]*base.TaskMessage{
-				"default": []*base.TaskMessage{
-					&base.TaskMessage{
+				"default": {
+					{
 						Type:     task.Type,
 						Payload:  task.Payload.data,
 						Retry:    defaultMaxRetry,
@@ -361,3 +363,210 @@ func TestClientEnqueueIn(t *testing.T) {
 		}
 	}
 }
+
+func TestUniqueKey(t *testing.T) {
+	tests := []struct {
+		desc  string
+		task  *Task
+		ttl   time.Duration
+		qname string
+		want  string
+	}{
+		{
+			"with zero TTL",
+			NewTask("email:send", map[string]interface{}{"a": 123, "b": "hello", "c": true}),
+			0,
+			"default",
+			"",
+		},
+		{
+			"with primitive types",
+			NewTask("email:send", map[string]interface{}{"a": 123, "b": "hello", "c": true}),
+			10 * time.Minute,
+			"default",
+			"email:send:a=123,b=hello,c=true:default",
+		},
+		{
+			"with unsorted keys",
+			NewTask("email:send", map[string]interface{}{"b": "hello", "c": true, "a": 123}),
+			10 * time.Minute,
+			"default",
+			"email:send:a=123,b=hello,c=true:default",
+		},
+		{
+			"with composite types",
+			NewTask("email:send",
+				map[string]interface{}{
+					"address": map[string]string{"line": "123 Main St", "city": "Boston", "state": "MA"},
+					"names":   []string{"bob", "mike", "rob"}}),
+			10 * time.Minute,
+			"default",
+			"email:send:address=map[city:Boston line:123 Main St state:MA],names=[bob mike rob]:default",
+		},
+		{
+			"with complex types",
+			NewTask("email:send",
+				map[string]interface{}{
+					"time":     time.Date(2020, time.July, 28, 0, 0, 0, 0, time.UTC),
+					"duration": time.Hour}),
+			10 * time.Minute,
+			"default",
+			"email:send:duration=1h0m0s,time=2020-07-28 00:00:00 +0000 UTC:default",
+		},
+		{
+			"with nil payload",
+			NewTask("reindex", nil),
+			10 * time.Minute,
+			"default",
+			"reindex:nil:default",
+		},
+	}
+
+	for _, tc := range tests {
+		got := uniqueKey(tc.task, tc.ttl, tc.qname)
+		if got != tc.want {
+			t.Errorf("%s: uniqueKey(%v, %v, %q) = %q, want %q", tc.desc, tc.task, tc.ttl, tc.qname, got, tc.want)
+		}
+	}
+}
+
+func TestEnqueueUnique(t *testing.T) {
+	r := setup(t)
+	c := NewClient(RedisClientOpt{
+		Addr: redisAddr,
+		DB:   redisDB,
+	})
+
+	tests := []struct {
+		task *Task
+		ttl  time.Duration
+	}{
+		{
+			NewTask("email", map[string]interface{}{"user_id": 123}),
+			time.Hour,
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r) // clean up db before each test case.
+
+		// Enqueue the task first. It should succeed.
+		err := c.Enqueue(tc.task, Unique(tc.ttl))
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		gotTTL := r.TTL(uniqueKey(tc.task, tc.ttl, base.DefaultQueueName)).Val()
+		if !cmp.Equal(tc.ttl.Seconds(), gotTTL.Seconds(), cmpopts.EquateApprox(0, 1)) {
+			t.Errorf("TTL = %v, want %v", gotTTL, tc.ttl)
+			continue
+		}
+
+		// Enqueue the task again. It should fail.
+		err = c.Enqueue(tc.task, Unique(tc.ttl))
+		if err == nil {
+			t.Errorf("Enqueueing %+v did not return an error", tc.task)
+			continue
+		}
+		if !errors.Is(err, ErrDuplicateTask) {
+			t.Errorf("Enqueueing %+v returned an error that is not ErrDuplicateTask", tc.task)
+			continue
+		}
+	}
+}
+
+func TestEnqueueInUnique(t *testing.T) {
+	r := setup(t)
+	c := NewClient(RedisClientOpt{
+		Addr: redisAddr,
+		DB:   redisDB,
+	})
+
+	tests := []struct {
+		task *Task
+		d    time.Duration
+		ttl  time.Duration
+	}{
+		{
+			NewTask("reindex", nil),
+			time.Hour,
+			10 * time.Minute,
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r) // clean up db before each test case.
+
+		// Enqueue the task first. It should succeed.
+		err := c.EnqueueIn(tc.d, tc.task, Unique(tc.ttl))
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		gotTTL := r.TTL(uniqueKey(tc.task, tc.ttl, base.DefaultQueueName)).Val()
+		wantTTL := time.Duration(tc.ttl.Seconds()+tc.d.Seconds()) * time.Second
+		if !cmp.Equal(wantTTL.Seconds(), gotTTL.Seconds(), cmpopts.EquateApprox(0, 1)) {
+			t.Errorf("TTL = %v, want %v", gotTTL, wantTTL)
+			continue
+		}
+
+		// Enqueue the task again. It should fail.
+		err = c.EnqueueIn(tc.d, tc.task, Unique(tc.ttl))
+		if err == nil {
+			t.Errorf("Enqueueing %+v did not return an error", tc.task)
+			continue
+		}
+		if !errors.Is(err, ErrDuplicateTask) {
+			t.Errorf("Enqueueing %+v returned an error that is not ErrDuplicateTask", tc.task)
+			continue
+		}
+	}
+}
+
+func TestEnqueueAtUnique(t *testing.T) {
+	r := setup(t)
+	c := NewClient(RedisClientOpt{
+		Addr: redisAddr,
+		DB:   redisDB,
+	})
+
+	tests := []struct {
+		task *Task
+		at   time.Time
+		ttl  time.Duration
+	}{
+		{
+			NewTask("reindex", nil),
+			time.Now().Add(time.Hour),
+			10 * time.Minute,
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r) // clean up db before each test case.
+
+		// Enqueue the task first. It should succeed.
+		err := c.EnqueueAt(tc.at, tc.task, Unique(tc.ttl))
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		gotTTL := r.TTL(uniqueKey(tc.task, tc.ttl, base.DefaultQueueName)).Val()
+		wantTTL := tc.at.Add(tc.ttl).Sub(time.Now())
+		if !cmp.Equal(wantTTL.Seconds(), gotTTL.Seconds(), cmpopts.EquateApprox(0, 1)) {
+			t.Errorf("TTL = %v, want %v", gotTTL, wantTTL)
+			continue
+		}
+
+		// Enqueue the task again. It should fail.
+		err = c.EnqueueAt(tc.at, tc.task, Unique(tc.ttl))
+		if err == nil {
+			t.Errorf("Enqueueing %+v did not return an error", tc.task)
+			continue
+		}
+		if !errors.Is(err, ErrDuplicateTask) {
+			t.Errorf("Enqueueing %+v returned an error that is not ErrDuplicateTask", tc.task)
+			continue
+		}
+	}
+}

internal/base/base.go

@@ -97,6 +97,11 @@ type TaskMessage struct {
 	//
 	// time.Time's zero value means no deadline.
 	Deadline string
+
+	// UniqueKey holds the redis key used for uniqueness lock for this task.
+	//
+	// Empty string indicates that no uniqueness lock was used.
+	UniqueKey string
 }
 
 // ProcessState holds process level information.

internal/base/base_test.go

@@ -110,9 +110,9 @@ func TestProcessStateConcurrentAccess(t *testing.T) {
 	var wg sync.WaitGroup
 	started := time.Now()
 	msgs := []*TaskMessage{
-		&TaskMessage{ID: xid.New(), Type: "type1", Payload: map[string]interface{}{"user_id": 42}},
-		&TaskMessage{ID: xid.New(), Type: "type2"},
-		&TaskMessage{ID: xid.New(), Type: "type3"},
+		{ID: xid.New(), Type: "type1", Payload: map[string]interface{}{"user_id": 42}},
+		{ID: xid.New(), Type: "type2"},
+		{ID: xid.New(), Type: "type3"},
 	}
 
 	// Simulate hearbeater calling SetStatus and SetStarted.

internal/log/log.go

@@ -11,36 +11,45 @@ import (
 	"os"
 )
 
+// NewLogger creates and returns a new instance of Logger.
 func NewLogger(out io.Writer) *Logger {
 	return &Logger{
 		stdlog.New(out, "", stdlog.Ldate|stdlog.Ltime|stdlog.Lmicroseconds|stdlog.LUTC),
 	}
 }
 
+// Logger is a wrapper object around log.Logger from the standard library.
+// It supports logging at various log levels.
 type Logger struct {
 	*stdlog.Logger
 }
 
+// Debug logs a message at Debug level.
 func (l *Logger) Debug(format string, args ...interface{}) {
 	format = "DEBUG: " + format
 	l.Printf(format, args...)
 }
 
+// Info logs a message at Info level.
 func (l *Logger) Info(format string, args ...interface{}) {
 	format = "INFO: " + format
 	l.Printf(format, args...)
 }
 
+// Warn logs a message at Warning level.
 func (l *Logger) Warn(format string, args ...interface{}) {
 	format = "WARN: " + format
 	l.Printf(format, args...)
 }
 
+// Error logs a message at Error level.
 func (l *Logger) Error(format string, args ...interface{}) {
 	format = "ERROR: " + format
 	l.Printf(format, args...)
 }
 
+// Fatal logs a message at Fatal level
+// and process will exit with status set to 1.
 func (l *Logger) Fatal(format string, args ...interface{}) {
 	format = "FATAL: " + format
 	l.Printf(format, args...)

internal/rdb/rdb.go

@@ -22,6 +22,9 @@ var (
 
 	// ErrTaskNotFound indicates that a task that matches the given identifier was not found.
 	ErrTaskNotFound = errors.New("could not find a task")
+
+	// ErrDuplicateTask indicates that another task with the same unique key holds the uniqueness lock.
+	ErrDuplicateTask = errors.New("task already exists")
 )
 
 const statsTTL = 90 * 24 * time.Hour // 90 days
@@ -59,6 +62,46 @@ func (r *RDB) Enqueue(msg *base.TaskMessage) error {
 	return enqueueCmd.Run(r.client, []string{key, base.AllQueues}, bytes).Err()
 }
 
+// KEYS[1] -> unique key in the form <type>:<payload>:<qname>
+// KEYS[2] -> asynq:queues:<qname>
+// KEYS[2] -> asynq:queues
+// ARGV[1] -> task ID
+// ARGV[2] -> uniqueness lock TTL
+// ARGV[3] -> task message data
+var enqueueUniqueCmd = redis.NewScript(`
+local ok = redis.call("SET", KEYS[1], ARGV[1], "NX", "EX", ARGV[2])
+if not ok then
+  return 0
+end
+redis.call("LPUSH", KEYS[2], ARGV[3])
+redis.call("SADD", KEYS[3], KEYS[2])
+return 1
+`)
+
+// EnqueueUnique inserts the given task if the task's uniqueness lock can be acquired.
+// It returns ErrDuplicateTask if the lock cannot be acquired.
+func (r *RDB) EnqueueUnique(msg *base.TaskMessage, ttl time.Duration) error {
+	bytes, err := json.Marshal(msg)
+	if err != nil {
+		return err
+	}
+	key := base.QueueKey(msg.Queue)
+	res, err := enqueueUniqueCmd.Run(r.client,
+		[]string{msg.UniqueKey, key, base.AllQueues},
+		msg.ID.String(), int(ttl.Seconds()), bytes).Result()
+	if err != nil {
+		return err
+	}
+	n, ok := res.(int64)
+	if !ok {
+		return fmt.Errorf("could not cast %v to int64", res)
+	}
+	if n == 0 {
+		return ErrDuplicateTask
+	}
+	return nil
+}
+
 // Dequeue queries given queues in order and pops a task message if there is one and returns it.
 // If all queues are empty, ErrNoProcessableTask error is returned.
 func (r *RDB) Dequeue(qnames ...string) (*base.TaskMessage, error) {
@@ -118,8 +161,10 @@ func (r *RDB) dequeue(queues ...string) (data string, err error) {
 
 // KEYS[1] -> asynq:in_progress
 // KEYS[2] -> asynq:processed:<yyyy-mm-dd>
+// KEYS[3] -> unique key in the format <type>:<payload>:<qname>
 // ARGV[1] -> base.TaskMessage value
 // ARGV[2] -> stats expiration timestamp
+// ARGV[3] -> task ID
 // Note: LREM count ZERO means "remove all elements equal to val"
 var doneCmd = redis.NewScript(`
 redis.call("LREM", KEYS[1], 0, ARGV[1]) 
@@ -127,10 +172,14 @@ local n = redis.call("INCR", KEYS[2])
 if tonumber(n) == 1 then
 	redis.call("EXPIREAT", KEYS[2], ARGV[2])
 end
+if string.len(KEYS[3]) > 0 and redis.call("GET", KEYS[3]) == ARGV[3] then
+  redis.call("DEL", KEYS[3])
+end
 return redis.status_reply("OK")
 `)
 
 // Done removes the task from in-progress queue to mark the task as done.
+// It removes a uniqueness lock acquired by the task, if any.
 func (r *RDB) Done(msg *base.TaskMessage) error {
 	bytes, err := json.Marshal(msg)
 	if err != nil {
@@ -140,8 +189,8 @@ func (r *RDB) Done(msg *base.TaskMessage) error {
 	processedKey := base.ProcessedKey(now)
 	expireAt := now.Add(statsTTL)
 	return doneCmd.Run(r.client,
-		[]string{base.InProgressQueue, processedKey},
-		bytes, expireAt.Unix()).Err()
+		[]string{base.InProgressQueue, processedKey, msg.UniqueKey},
+		bytes, expireAt.Unix(), msg.ID.String()).Err()
 }
 
 // KEYS[1] -> asynq:in_progress
@@ -164,15 +213,71 @@ func (r *RDB) Requeue(msg *base.TaskMessage) error {
 		string(bytes)).Err()
 }
 
+// KEYS[1] -> asynq:scheduled
+// KEYS[2] -> asynq:queues
+// ARGV[1] -> score (process_at timestamp)
+// ARGV[2] -> task message
+// ARGV[3] -> queue key
+var scheduleCmd = redis.NewScript(`
+redis.call("ZADD", KEYS[1], ARGV[1], ARGV[2])
+redis.call("SADD", KEYS[2], ARGV[3])
+return 1
+`)
+
 // Schedule adds the task to the backlog queue to be processed in the future.
 func (r *RDB) Schedule(msg *base.TaskMessage, processAt time.Time) error {
 	bytes, err := json.Marshal(msg)
 	if err != nil {
 		return err
 	}
+	qkey := base.QueueKey(msg.Queue)
 	score := float64(processAt.Unix())
-	return r.client.ZAdd(base.ScheduledQueue,
-		&redis.Z{Member: string(bytes), Score: score}).Err()
+	return scheduleCmd.Run(r.client,
+		[]string{base.ScheduledQueue, base.AllQueues},
+		score, bytes, qkey).Err()
+}
+
+// KEYS[1] -> unique key in the format <type>:<payload>:<qname>
+// KEYS[2] -> asynq:scheduled
+// KEYS[3] -> asynq:queues
+// ARGV[1] -> task ID
+// ARGV[2] -> uniqueness lock TTL
+// ARGV[3] -> score (process_at timestamp)
+// ARGV[4] -> task message
+// ARGV[5] -> queue key
+var scheduleUniqueCmd = redis.NewScript(`
+local ok = redis.call("SET", KEYS[1], ARGV[1], "NX", "EX", ARGV[2])
+if not ok then
+  return 0
+end
+redis.call("ZADD", KEYS[2], ARGV[3], ARGV[4])
+redis.call("SADD", KEYS[3], ARGV[5])
+return 1
+`)
+
+// ScheduleUnique adds the task to the backlog queue to be processed in the future if the uniqueness lock can be acquired.
+// It returns ErrDuplicateTask if the lock cannot be acquired.
+func (r *RDB) ScheduleUnique(msg *base.TaskMessage, processAt time.Time, ttl time.Duration) error {
+	bytes, err := json.Marshal(msg)
+	if err != nil {
+		return err
+	}
+	qkey := base.QueueKey(msg.Queue)
+	score := float64(processAt.Unix())
+	res, err := scheduleUniqueCmd.Run(r.client,
+		[]string{msg.UniqueKey, base.ScheduledQueue, base.AllQueues},
+		msg.ID.String(), int(ttl.Seconds()), score, bytes, qkey).Result()
+	if err != nil {
+		return err
+	}
+	n, ok := res.(int64)
+	if !ok {
+		return fmt.Errorf("could not cast %v to int64", res)
+	}
+	if n == 0 {
+		return ErrDuplicateTask
+	}
+	return nil
 }
 
 // KEYS[1] -> asynq:in_progress

internal/rdb/rdb_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/google/go-cmp/cmp/cmpopts"
 	h "github.com/hibiken/asynq/internal/asynqtest"
 	"github.com/hibiken/asynq/internal/base"
+	"github.com/rs/xid"
 )
 
 // TODO(hibiken): Get Redis address and db number from ENV variables.
@@ -69,6 +70,48 @@ func TestEnqueue(t *testing.T) {
 	}
 }
 
+func TestEnqueueUnique(t *testing.T) {
+	r := setup(t)
+	m1 := base.TaskMessage{
+		ID:        xid.New(),
+		Type:      "email",
+		Payload:   map[string]interface{}{"user_id": 123},
+		Queue:     base.DefaultQueueName,
+		UniqueKey: "email:user_id=123:default",
+	}
+
+	tests := []struct {
+		msg *base.TaskMessage
+		ttl time.Duration // uniqueness ttl
+	}{
+		{&m1, time.Minute},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client) // clean up db before each test case.
+
+		err := r.EnqueueUnique(tc.msg, tc.ttl)
+		if err != nil {
+			t.Errorf("First message: (*RDB).EnqueueUnique(%v, %v) = %v, want nil",
+				tc.msg, tc.ttl, err)
+			continue
+		}
+
+		got := r.EnqueueUnique(tc.msg, tc.ttl)
+		if got != ErrDuplicateTask {
+			t.Errorf("Second message: (*RDB).EnqueueUnique(%v, %v) = %v, want %v",
+				tc.msg, tc.ttl, got, ErrDuplicateTask)
+			continue
+		}
+
+		gotTTL := r.client.TTL(tc.msg.UniqueKey).Val()
+		if !cmp.Equal(tc.ttl.Seconds(), gotTTL.Seconds(), cmpopts.EquateApprox(0, 1)) {
+			t.Errorf("TTL %q = %v, want %v", tc.msg.UniqueKey, gotTTL, tc.ttl)
+			continue
+		}
+	}
+}
+
 func TestDequeue(t *testing.T) {
 	r := setup(t)
 	t1 := h.NewTaskMessage("send_email", map[string]interface{}{"subject": "hello!"})
@@ -188,6 +231,13 @@ func TestDone(t *testing.T) {
 	r := setup(t)
 	t1 := h.NewTaskMessage("send_email", nil)
 	t2 := h.NewTaskMessage("export_csv", nil)
+	t3 := &base.TaskMessage{
+		ID:        xid.New(),
+		Type:      "reindex",
+		Payload:   nil,
+		UniqueKey: "reindex:nil:default",
+		Queue:     "default",
+	}
 
 	tests := []struct {
 		inProgress     []*base.TaskMessage // initial state of the in-progress list
@@ -204,11 +254,25 @@ func TestDone(t *testing.T) {
 			target:         t1,
 			wantInProgress: []*base.TaskMessage{},
 		},
+		{
+			inProgress:     []*base.TaskMessage{t1, t2, t3},
+			target:         t3,
+			wantInProgress: []*base.TaskMessage{t1, t2},
+		},
 	}
 
 	for _, tc := range tests {
 		h.FlushDB(t, r.client) // clean up db before each test case
 		h.SeedInProgressQueue(t, r.client, tc.inProgress)
+		for _, msg := range tc.inProgress {
+			// Set uniqueness lock if unique key is present.
+			if len(msg.UniqueKey) > 0 {
+				err := r.client.SetNX(msg.UniqueKey, msg.ID.String(), time.Minute).Err()
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+		}
 
 		err := r.Done(tc.target)
 		if err != nil {
@@ -232,6 +296,10 @@ func TestDone(t *testing.T) {
 		if gotTTL > statsTTL {
 			t.Errorf("TTL %q = %v, want less than or equal to %v", processedKey, gotTTL, statsTTL)
 		}
+
+		if len(tc.target.UniqueKey) > 0 && r.client.Exists(tc.target.UniqueKey).Val() != 0 {
+			t.Errorf("Uniqueness lock %q still exists", tc.target.UniqueKey)
+		}
 	}
 }
 
@@ -344,6 +412,58 @@ func TestSchedule(t *testing.T) {
 	}
 }
 
+func TestScheduleUnique(t *testing.T) {
+	r := setup(t)
+	m1 := base.TaskMessage{
+		ID:        xid.New(),
+		Type:      "email",
+		Payload:   map[string]interface{}{"user_id": 123},
+		Queue:     base.DefaultQueueName,
+		UniqueKey: "email:user_id=123:default",
+	}
+
+	tests := []struct {
+		msg       *base.TaskMessage
+		processAt time.Time
+		ttl       time.Duration // uniqueness lock ttl
+	}{
+		{&m1, time.Now().Add(15 * time.Minute), time.Minute},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client) // clean up db before each test case
+
+		desc := fmt.Sprintf("(*RDB).ScheduleUnique(%v, %v, %v)", tc.msg, tc.processAt, tc.ttl)
+		err := r.ScheduleUnique(tc.msg, tc.processAt, tc.ttl)
+		if err != nil {
+			t.Errorf("Frist task: %s = %v, want nil", desc, err)
+			continue
+		}
+
+		gotScheduled := h.GetScheduledEntries(t, r.client)
+		if len(gotScheduled) != 1 {
+			t.Errorf("%s inserted %d items to %q, want 1 items inserted", desc, len(gotScheduled), base.ScheduledQueue)
+			continue
+		}
+		if int64(gotScheduled[0].Score) != tc.processAt.Unix() {
+			t.Errorf("%s inserted an item with score %d, want %d", desc, int64(gotScheduled[0].Score), tc.processAt.Unix())
+			continue
+		}
+
+		got := r.ScheduleUnique(tc.msg, tc.processAt, tc.ttl)
+		if got != ErrDuplicateTask {
+			t.Errorf("Second task: %s = %v, want %v",
+				desc, got, ErrDuplicateTask)
+		}
+
+		gotTTL := r.client.TTL(tc.msg.UniqueKey).Val()
+		if !cmp.Equal(tc.ttl.Seconds(), gotTTL.Seconds(), cmpopts.EquateApprox(0, 1)) {
+			t.Errorf("TTL %q = %v, want %v", tc.msg.UniqueKey, gotTTL, tc.ttl)
+			continue
+		}
+	}
+}
+
 func TestRetry(t *testing.T) {
 	r := setup(t)
 	t1 := h.NewTaskMessage("send_email", map[string]interface{}{"subject": "Hola!"})
@@ -784,8 +904,7 @@ func TestWriteProcessState(t *testing.T) {
 	}
 	// Check ProcessInfo TTL was set correctly
 	gotTTL := r.client.TTL(pkey).Val()
-	timeCmpOpt := cmpopts.EquateApproxTime(time.Second)
-	if !cmp.Equal(ttl, gotTTL, timeCmpOpt) {
+	if !cmp.Equal(ttl.Seconds(), gotTTL.Seconds(), cmpopts.EquateApprox(0, 1)) {
 		t.Errorf("TTL of %q was %v, want %v", pkey, gotTTL, ttl)
 	}
 	// Check ProcessInfo key was added to the set correctly
@@ -858,8 +977,7 @@ func TestWriteProcessStateWithWorkers(t *testing.T) {
 	}
 	// Check ProcessInfo TTL was set correctly
 	gotTTL := r.client.TTL(pkey).Val()
-	timeCmpOpt := cmpopts.EquateApproxTime(time.Second)
-	if !cmp.Equal(ttl, gotTTL, timeCmpOpt) {
+	if !cmp.Equal(ttl.Seconds(), gotTTL.Seconds(), cmpopts.EquateApprox(0, 1)) {
 		t.Errorf("TTL of %q was %v, want %v", pkey, gotTTL, ttl)
 	}
 	// Check ProcessInfo key was added to the set correctly
@@ -884,7 +1002,7 @@ func TestWriteProcessStateWithWorkers(t *testing.T) {
 		gotWorkers[key] = &w
 	}
 	wantWorkers := map[string]*base.WorkerInfo{
-		msg1.ID.String(): &base.WorkerInfo{
+		msg1.ID.String(): {
 			Host:    host,
 			PID:     pid,
 			ID:      msg1.ID,
@@ -893,7 +1011,7 @@ func TestWriteProcessStateWithWorkers(t *testing.T) {
 			Payload: msg1.Payload,
 			Started: w1Started,
 		},
-		msg2.ID.String(): &base.WorkerInfo{
+		msg2.ID.String(): {
 			Host:    host,
 			PID:     pid,
 			ID:      msg2.ID,

payload_test.go

@@ -14,333 +14,626 @@ import (
 	"github.com/hibiken/asynq/internal/base"
 )
 
-func TestPayloadGet(t *testing.T) {
-	names := []string{"luke", "anakin", "rey"}
-	primes := []int{2, 3, 5, 7, 11, 13, 17}
-	user := map[string]interface{}{"name": "Ken", "score": 3.14}
-	location := map[string]string{"address": "123 Main St.", "state": "NY", "zipcode": "10002"}
-	favs := map[string][]string{
-		"movies":   []string{"forrest gump", "star wars"},
-		"tv_shows": []string{"game of thrones", "HIMYM", "breaking bad"},
+type payloadTest struct {
+	data   map[string]interface{}
+	key    string
+	nonkey string
 }
-	counter := map[string]int{
-		"a": 1,
-		"b": 101,
-		"c": 42,
-	}
-	features := map[string]bool{
-		"A": false,
-		"B": true,
-		"C": true,
-	}
-	now := time.Now()
-	duration := 15 * time.Minute
 
-	data := map[string]interface{}{
-		"greeting":  "Hello",
-		"user_id":   9876,
-		"pi":        3.1415,
-		"enabled":   false,
-		"names":     names,
-		"primes":    primes,
-		"user":      user,
-		"location":  location,
-		"favs":      favs,
-		"counter":   counter,
-		"features":  features,
-		"timestamp": now,
-		"duration":  duration,
+func TestPayloadString(t *testing.T) {
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"name": "gopher"},
+			key:    "name",
+			nonkey: "unknown",
+		},
 	}
-	payload := Payload{data}
 
-	gotStr, err := payload.GetString("greeting")
-	if gotStr != "Hello" || err != nil {
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetString(tc.key)
+		if err != nil || got != tc.data[tc.key] {
 			t.Errorf("Payload.GetString(%q) = %v, %v, want %v, nil",
-			"greeting", gotStr, err, "Hello")
+				tc.key, got, err, tc.data[tc.key])
 		}
 
-	gotInt, err := payload.GetInt("user_id")
-	if gotInt != 9876 || err != nil {
-		t.Errorf("Payload.GetInt(%q) = %v, %v, want, %v, nil",
-			"user_id", gotInt, err, 9876)
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetString(tc.key)
+		if err != nil || got != tc.data[tc.key] {
+			t.Errorf("With Marshaling: Payload.GetString(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
 		}
 
-	gotFloat, err := payload.GetFloat64("pi")
-	if gotFloat != 3.1415 || err != nil {
-		t.Errorf("Payload.GetFloat64(%q) = %v, %v, want, %v, nil",
-			"pi", gotFloat, err, 3.141592)
+		// access non-existent key.
+		got, err = payload.GetString(tc.nonkey)
+		if err == nil || got != "" {
+			t.Errorf("Payload.GetString(%q) = %v, %v; want '', error",
+				tc.key, got, err)
 		}
-
-	gotBool, err := payload.GetBool("enabled")
-	if gotBool != false || err != nil {
-		t.Errorf("Payload.GetBool(%q) = %v, %v, want, %v, nil",
-			"enabled", gotBool, err, false)
-	}
-
-	gotStrSlice, err := payload.GetStringSlice("names")
-	if diff := cmp.Diff(gotStrSlice, names); diff != "" {
-		t.Errorf("Payload.GetStringSlice(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"names", gotStrSlice, err, names, diff)
-	}
-
-	gotIntSlice, err := payload.GetIntSlice("primes")
-	if diff := cmp.Diff(gotIntSlice, primes); diff != "" {
-		t.Errorf("Payload.GetIntSlice(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"primes", gotIntSlice, err, primes, diff)
-	}
-
-	gotStrMap, err := payload.GetStringMap("user")
-	if diff := cmp.Diff(gotStrMap, user); diff != "" {
-		t.Errorf("Payload.GetStringMap(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"user", gotStrMap, err, user, diff)
-	}
-
-	gotStrMapStr, err := payload.GetStringMapString("location")
-	if diff := cmp.Diff(gotStrMapStr, location); diff != "" {
-		t.Errorf("Payload.GetStringMapString(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"location", gotStrMapStr, err, location, diff)
-	}
-
-	gotStrMapStrSlice, err := payload.GetStringMapStringSlice("favs")
-	if diff := cmp.Diff(gotStrMapStrSlice, favs); diff != "" {
-		t.Errorf("Payload.GetStringMapStringSlice(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"favs", gotStrMapStrSlice, err, favs, diff)
-	}
-
-	gotStrMapInt, err := payload.GetStringMapInt("counter")
-	if diff := cmp.Diff(gotStrMapInt, counter); diff != "" {
-		t.Errorf("Payload.GetStringMapInt(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"counter", gotStrMapInt, err, counter, diff)
-	}
-
-	gotStrMapBool, err := payload.GetStringMapBool("features")
-	if diff := cmp.Diff(gotStrMapBool, features); diff != "" {
-		t.Errorf("Payload.GetStringMapBool(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"features", gotStrMapBool, err, features, diff)
-	}
-
-	gotTime, err := payload.GetTime("timestamp")
-	if !gotTime.Equal(now) {
-		t.Errorf("Payload.GetTime(%q) = %v, %v, want %v, nil",
-			"timestamp", gotTime, err, now)
-	}
-
-	gotDuration, err := payload.GetDuration("duration")
-	if gotDuration != duration {
-		t.Errorf("Payload.GetDuration(%q) = %v, %v, want %v, nil",
-			"duration", gotDuration, err, duration)
 	}
 }
 
-func TestPayloadGetWithMarshaling(t *testing.T) {
-	names := []string{"luke", "anakin", "rey"}
-	primes := []int{2, 3, 5, 7, 11, 13, 17}
-	user := map[string]interface{}{"name": "Ken", "score": 3.14}
-	location := map[string]string{"address": "123 Main St.", "state": "NY", "zipcode": "10002"}
+func TestPayloadInt(t *testing.T) {
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"user_id": 42},
+			key:    "user_id",
+			nonkey: "unknown",
+		},
+	}
+
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetInt(tc.key)
+		if err != nil || got != tc.data[tc.key] {
+			t.Errorf("Payload.GetInt(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetInt(tc.key)
+		if err != nil || got != tc.data[tc.key] {
+			t.Errorf("With Marshaling: Payload.GetInt(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// access non-existent key.
+		got, err = payload.GetInt(tc.nonkey)
+		if err == nil || got != 0 {
+			t.Errorf("Payload.GetInt(%q) = %v, %v; want 0, error",
+				tc.key, got, err)
+		}
+	}
+}
+
+func TestPayloadFloat64(t *testing.T) {
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"pi": 3.14},
+			key:    "pi",
+			nonkey: "unknown",
+		},
+	}
+
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetFloat64(tc.key)
+		if err != nil || got != tc.data[tc.key] {
+			t.Errorf("Payload.GetFloat64(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetFloat64(tc.key)
+		if err != nil || got != tc.data[tc.key] {
+			t.Errorf("With Marshaling: Payload.GetFloat64(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// access non-existent key.
+		got, err = payload.GetFloat64(tc.nonkey)
+		if err == nil || got != 0 {
+			t.Errorf("Payload.GetFloat64(%q) = %v, %v; want 0, error",
+				tc.key, got, err)
+		}
+	}
+}
+
+func TestPayloadBool(t *testing.T) {
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"enabled": true},
+			key:    "enabled",
+			nonkey: "unknown",
+		},
+	}
+
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetBool(tc.key)
+		if err != nil || got != tc.data[tc.key] {
+			t.Errorf("Payload.GetBool(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetBool(tc.key)
+		if err != nil || got != tc.data[tc.key] {
+			t.Errorf("With Marshaling: Payload.GetBool(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// access non-existent key.
+		got, err = payload.GetBool(tc.nonkey)
+		if err == nil || got != false {
+			t.Errorf("Payload.GetBool(%q) = %v, %v; want false, error",
+				tc.key, got, err)
+		}
+	}
+}
+
+func TestPayloadStringSlice(t *testing.T) {
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"names": []string{"luke", "rey", "anakin"}},
+			key:    "names",
+			nonkey: "unknown",
+		},
+	}
+
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetStringSlice(tc.key)
+		diff := cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("Payload.GetStringSlice(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetStringSlice(tc.key)
+		diff = cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("With Marshaling: Payload.GetStringSlice(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// access non-existent key.
+		got, err = payload.GetStringSlice(tc.nonkey)
+		if err == nil || got != nil {
+			t.Errorf("Payload.GetStringSlice(%q) = %v, %v; want nil, error",
+				tc.key, got, err)
+		}
+	}
+}
+
+func TestPayloadIntSlice(t *testing.T) {
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"nums": []int{9, 8, 7}},
+			key:    "nums",
+			nonkey: "unknown",
+		},
+	}
+
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetIntSlice(tc.key)
+		diff := cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("Payload.GetIntSlice(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetIntSlice(tc.key)
+		diff = cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("With Marshaling: Payload.GetIntSlice(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// access non-existent key.
+		got, err = payload.GetIntSlice(tc.nonkey)
+		if err == nil || got != nil {
+			t.Errorf("Payload.GetIntSlice(%q) = %v, %v; want nil, error",
+				tc.key, got, err)
+		}
+	}
+}
+
+func TestPayloadStringMap(t *testing.T) {
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"user": map[string]interface{}{"name": "Jon Doe", "score": 2.2}},
+			key:    "user",
+			nonkey: "unknown",
+		},
+	}
+
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetStringMap(tc.key)
+		diff := cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("Payload.GetStringMap(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetStringMap(tc.key)
+		diff = cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("With Marshaling: Payload.GetStringMap(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// access non-existent key.
+		got, err = payload.GetStringMap(tc.nonkey)
+		if err == nil || got != nil {
+			t.Errorf("Payload.GetStringMap(%q) = %v, %v; want nil, error",
+				tc.key, got, err)
+		}
+	}
+}
+
+func TestPayloadStringMapString(t *testing.T) {
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"address": map[string]string{"line": "123 Main St", "city": "San Francisco", "state": "CA"}},
+			key:    "address",
+			nonkey: "unknown",
+		},
+	}
+
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetStringMapString(tc.key)
+		diff := cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("Payload.GetStringMapString(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetStringMapString(tc.key)
+		diff = cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("With Marshaling: Payload.GetStringMapString(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// access non-existent key.
+		got, err = payload.GetStringMapString(tc.nonkey)
+		if err == nil || got != nil {
+			t.Errorf("Payload.GetStringMapString(%q) = %v, %v; want nil, error",
+				tc.key, got, err)
+		}
+	}
+}
+
+func TestPayloadStringMapStringSlice(t *testing.T) {
 	favs := map[string][]string{
-		"movies":   []string{"forrest gump", "star wars"},
-		"tv_shows": []string{"game of throwns", "HIMYM", "breaking bad"},
+		"movies":   {"forrest gump", "star wars"},
+		"tv_shows": {"game of thrones", "HIMYM", "breaking bad"},
 	}
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"favorites": favs},
+			key:    "favorites",
+			nonkey: "unknown",
+		},
+	}
+
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetStringMapStringSlice(tc.key)
+		diff := cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("Payload.GetStringMapStringSlice(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetStringMapStringSlice(tc.key)
+		diff = cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("With Marshaling: Payload.GetStringMapStringSlice(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// access non-existent key.
+		got, err = payload.GetStringMapStringSlice(tc.nonkey)
+		if err == nil || got != nil {
+			t.Errorf("Payload.GetStringMapStringSlice(%q) = %v, %v; want nil, error",
+				tc.key, got, err)
+		}
+	}
+}
+
+func TestPayloadStringMapInt(t *testing.T) {
 	counter := map[string]int{
 		"a": 1,
 		"b": 101,
 		"c": 42,
 	}
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"counts": counter},
+			key:    "counts",
+			nonkey: "unknown",
+		},
+	}
+
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetStringMapInt(tc.key)
+		diff := cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("Payload.GetStringMapInt(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetStringMapInt(tc.key)
+		diff = cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("With Marshaling: Payload.GetStringMapInt(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// access non-existent key.
+		got, err = payload.GetStringMapInt(tc.nonkey)
+		if err == nil || got != nil {
+			t.Errorf("Payload.GetStringMapInt(%q) = %v, %v; want nil, error",
+				tc.key, got, err)
+		}
+	}
+}
+
+func TestPayloadStringMapBool(t *testing.T) {
 	features := map[string]bool{
 		"A": false,
 		"B": true,
 		"C": true,
 	}
-	now := time.Now()
-	duration := 15 * time.Minute
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"features": features},
+			key:    "features",
+			nonkey: "unknown",
+		},
+	}
 
-	in := Payload{map[string]interface{}{
-		"subject":      "Hello",
-		"recipient_id": 9876,
-		"pi":           3.14,
-		"enabled":      true,
-		"names":        names,
-		"primes":       primes,
-		"user":         user,
-		"location":     location,
-		"favs":         favs,
-		"counter":      counter,
-		"features":     features,
-		"timestamp":    now,
-		"duration":     duration,
-	}}
-	// encode and then decode task messsage
-	inMsg := h.NewTaskMessage("testing", in.data)
-	data, err := json.Marshal(inMsg)
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetStringMapBool(tc.key)
+		diff := cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("Payload.GetStringMapBool(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
 		if err != nil {
 			t.Fatal(err)
 		}
-	var outMsg base.TaskMessage
-	err = json.Unmarshal(data, &outMsg)
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
 		if err != nil {
 			t.Fatal(err)
 		}
-	out := Payload{outMsg.Payload}
-
-	gotStr, err := out.GetString("subject")
-	if gotStr != "Hello" || err != nil {
-		t.Errorf("Payload.GetString(%q) = %v, %v; want %q, nil",
-			"subject", gotStr, err, "Hello")
+		payload = Payload{out.Payload}
+		got, err = payload.GetStringMapBool(tc.key)
+		diff = cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("With Marshaling: Payload.GetStringMapBool(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
 		}
 
-	gotInt, err := out.GetInt("recipient_id")
-	if gotInt != 9876 || err != nil {
-		t.Errorf("Payload.GetInt(%q) = %v, %v; want %v, nil",
-			"recipient_id", gotInt, err, 9876)
+		// access non-existent key.
+		got, err = payload.GetStringMapBool(tc.nonkey)
+		if err == nil || got != nil {
+			t.Errorf("Payload.GetStringMapBool(%q) = %v, %v; want nil, error",
+				tc.key, got, err)
+		}
+	}
 }
 
-	gotFloat, err := out.GetFloat64("pi")
-	if gotFloat != 3.14 || err != nil {
-		t.Errorf("Payload.GetFloat64(%q) = %v, %v; want %v, nil",
-			"pi", gotFloat, err, 3.14)
+func TestPayloadTime(t *testing.T) {
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"current": time.Now()},
+			key:    "current",
+			nonkey: "unknown",
+		},
 	}
 
-	gotBool, err := out.GetBool("enabled")
-	if gotBool != true || err != nil {
-		t.Errorf("Payload.GetBool(%q) = %v, %v; want %v, nil",
-			"enabled", gotBool, err, true)
-	}
+	for _, tc := range tests {
+		payload := Payload{tc.data}
 
-	gotStrSlice, err := out.GetStringSlice("names")
-	if diff := cmp.Diff(gotStrSlice, names); diff != "" {
-		t.Errorf("Payload.GetStringSlice(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"names", gotStrSlice, err, names, diff)
-	}
-
-	gotIntSlice, err := out.GetIntSlice("primes")
-	if diff := cmp.Diff(gotIntSlice, primes); diff != "" {
-		t.Errorf("Payload.GetIntSlice(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"primes", gotIntSlice, err, primes, diff)
-	}
-
-	gotStrMap, err := out.GetStringMap("user")
-	if diff := cmp.Diff(gotStrMap, user); diff != "" {
-		t.Errorf("Payload.GetStringMap(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"user", gotStrMap, err, user, diff)
-	}
-
-	gotStrMapStr, err := out.GetStringMapString("location")
-	if diff := cmp.Diff(gotStrMapStr, location); diff != "" {
-		t.Errorf("Payload.GetStringMapString(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"location", gotStrMapStr, err, location, diff)
-	}
-
-	gotStrMapStrSlice, err := out.GetStringMapStringSlice("favs")
-	if diff := cmp.Diff(gotStrMapStrSlice, favs); diff != "" {
-		t.Errorf("Payload.GetStringMapStringSlice(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"favs", gotStrMapStrSlice, err, favs, diff)
-	}
-
-	gotStrMapInt, err := out.GetStringMapInt("counter")
-	if diff := cmp.Diff(gotStrMapInt, counter); diff != "" {
-		t.Errorf("Payload.GetStringMapInt(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"counter", gotStrMapInt, err, counter, diff)
-	}
-
-	gotStrMapBool, err := out.GetStringMapBool("features")
-	if diff := cmp.Diff(gotStrMapBool, features); diff != "" {
-		t.Errorf("Payload.GetStringMapBool(%q) = %v, %v, want %v, nil;\n(-want,+got)\n%s",
-			"features", gotStrMapBool, err, features, diff)
-	}
-
-	gotTime, err := out.GetTime("timestamp")
-	if !gotTime.Equal(now) {
+		got, err := payload.GetTime(tc.key)
+		diff := cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
 			t.Errorf("Payload.GetTime(%q) = %v, %v, want %v, nil",
-			"timestamp", gotTime, err, now)
+				tc.key, got, err, tc.data[tc.key])
 		}
 
-	gotDuration, err := out.GetDuration("duration")
-	if gotDuration != duration {
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetTime(tc.key)
+		diff = cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("With Marshaling: Payload.GetTime(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
+		}
+
+		// access non-existent key.
+		got, err = payload.GetTime(tc.nonkey)
+		if err == nil || !got.IsZero() {
+			t.Errorf("Payload.GetTime(%q) = %v, %v; want %v, error",
+				tc.key, got, err, time.Time{})
+		}
+	}
+}
+
+func TestPayloadDuration(t *testing.T) {
+	tests := []payloadTest{
+		{
+			data:   map[string]interface{}{"duration": 15 * time.Minute},
+			key:    "duration",
+			nonkey: "unknown",
+		},
+	}
+
+	for _, tc := range tests {
+		payload := Payload{tc.data}
+
+		got, err := payload.GetDuration(tc.key)
+		diff := cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
 			t.Errorf("Payload.GetDuration(%q) = %v, %v, want %v, nil",
-			"duration", gotDuration, err, duration)
-	}
+				tc.key, got, err, tc.data[tc.key])
 		}
 
-func TestPayloadKeyNotFound(t *testing.T) {
-	payload := Payload{nil}
-
-	key := "something"
-	gotStr, err := payload.GetString(key)
-	if err == nil || gotStr != "" {
-		t.Errorf("Payload.GetString(%q) = %v, %v; want '', error",
-			key, gotStr, err)
+		// encode and then decode task messsage.
+		in := h.NewTaskMessage("testing", tc.data)
+		b, err := json.Marshal(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out base.TaskMessage
+		err = json.Unmarshal(b, &out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		payload = Payload{out.Payload}
+		got, err = payload.GetDuration(tc.key)
+		diff = cmp.Diff(got, tc.data[tc.key])
+		if err != nil || diff != "" {
+			t.Errorf("With Marshaling: Payload.GetDuration(%q) = %v, %v, want %v, nil",
+				tc.key, got, err, tc.data[tc.key])
 		}
 
-	gotInt, err := payload.GetInt(key)
-	if err == nil || gotInt != 0 {
-		t.Errorf("Payload.GetInt(%q) = %v, %v; want 0, error",
-			key, gotInt, err)
+		// access non-existent key.
+		got, err = payload.GetDuration(tc.nonkey)
+		if err == nil || got != 0 {
+			t.Errorf("Payload.GetDuration(%q) = %v, %v; want %v, error",
+				tc.key, got, err, time.Duration(0))
 		}
-
-	gotFloat, err := payload.GetFloat64(key)
-	if err == nil || gotFloat != 0 {
-		t.Errorf("Payload.GetFloat64(%q = %v, %v; want 0, error",
-			key, gotFloat, err)
-	}
-
-	gotBool, err := payload.GetBool(key)
-	if err == nil || gotBool != false {
-		t.Errorf("Payload.GetBool(%q) = %v, %v; want false, error",
-			key, gotBool, err)
-	}
-
-	gotStrSlice, err := payload.GetStringSlice(key)
-	if err == nil || gotStrSlice != nil {
-		t.Errorf("Payload.GetStringSlice(%q) = %v, %v; want nil, error",
-			key, gotStrSlice, err)
-	}
-
-	gotIntSlice, err := payload.GetIntSlice(key)
-	if err == nil || gotIntSlice != nil {
-		t.Errorf("Payload.GetIntSlice(%q) = %v, %v; want nil, error",
-			key, gotIntSlice, err)
-	}
-
-	gotStrMap, err := payload.GetStringMap(key)
-	if err == nil || gotStrMap != nil {
-		t.Errorf("Payload.GetStringMap(%q) = %v, %v; want nil, error",
-			key, gotStrMap, err)
-	}
-
-	gotStrMapStr, err := payload.GetStringMapString(key)
-	if err == nil || gotStrMapStr != nil {
-		t.Errorf("Payload.GetStringMapString(%q) = %v, %v; want nil, error",
-			key, gotStrMapStr, err)
-	}
-
-	gotStrMapStrSlice, err := payload.GetStringMapStringSlice(key)
-	if err == nil || gotStrMapStrSlice != nil {
-		t.Errorf("Payload.GetStringMapStringSlice(%q) = %v, %v; want nil, error",
-			key, gotStrMapStrSlice, err)
-	}
-
-	gotStrMapInt, err := payload.GetStringMapInt(key)
-	if err == nil || gotStrMapInt != nil {
-		t.Errorf("Payload.GetStringMapInt(%q) = %v, %v, want nil, error",
-			key, gotStrMapInt, err)
-	}
-
-	gotStrMapBool, err := payload.GetStringMapBool(key)
-	if err == nil || gotStrMapBool != nil {
-		t.Errorf("Payload.GetStringMapBool(%q) = %v, %v, want nil, error",
-			key, gotStrMapBool, err)
-	}
-
-	gotTime, err := payload.GetTime(key)
-	if err == nil || !gotTime.IsZero() {
-		t.Errorf("Payload.GetTime(%q) = %v, %v, want %v, error",
-			key, gotTime, err, time.Time{})
-	}
-
-	gotDuration, err := payload.GetDuration(key)
-	if err == nil || gotDuration != 0 {
-		t.Errorf("Payload.GetDuration(%q) = %v, %v, want 0, error",
-			key, gotDuration, err)
 	}
 }
 

servemux.go

@@ -26,6 +26,7 @@ type ServeMux struct {
 	mu  sync.RWMutex
 	m   map[string]muxEntry
 	es  []muxEntry // slice of entries sorted from longest to shortest.
+	mws []MiddlewareFunc
 }
 
 type muxEntry struct {
@@ -33,6 +34,11 @@ type muxEntry struct {
 	pattern string
 }
 
+// MiddlewareFunc is a function which receives an asynq.Handler and returns another asynq.Handler.
+// Typically, the returned handler is a closure which does something with the context and task passed
+// to it, and then calls the handler passed as parameter to the MiddlewareFunc.
+type MiddlewareFunc func(Handler) Handler
+
 // NewServeMux allocates and returns a new ServeMux.
 func NewServeMux() *ServeMux {
 	return new(ServeMux)
@@ -60,6 +66,9 @@ func (mux *ServeMux) Handler(t *Task) (h Handler, pattern string) {
 	if h == nil {
 		h, pattern = NotFoundHandler(), ""
 	}
+	for i := len(mux.mws) - 1; i >= 0; i-- {
+		h = mux.mws[i](h)
+	}
 	return h, pattern
 }
 
@@ -130,6 +139,16 @@ func (mux *ServeMux) HandleFunc(pattern string, handler func(context.Context, *T
 	mux.Handle(pattern, HandlerFunc(handler))
 }
 
+// Use appends a MiddlewareFunc to the chain.
+// Middlewares are executed in the order that they are applied to the ServeMux.
+func (mux *ServeMux) Use(mws ...MiddlewareFunc) {
+	mux.mu.Lock()
+	defer mux.mu.Unlock()
+	for _, fn := range mws {
+		mux.mws = append(mux.mws, fn)
+	}
+}
+
 // NotFound returns an error indicating that the handler was not found for the given task.
 func NotFound(ctx context.Context, task *Task) error {
 	return fmt.Errorf("handler not found for task %q", task.Type)

servemux_test.go

@@ -7,9 +7,12 @@ package asynq
 import (
 	"context"
 	"testing"
+
+	"github.com/google/go-cmp/cmp"
 )
 
-var called string
+var called string    // identity of the handler that was called.
+var invoked []string // list of middlewares in the order they were invoked.
 
 // makeFakeHandler returns a handler that updates the global called variable
 // to the given identity.
@@ -20,6 +23,17 @@ func makeFakeHandler(identity string) Handler {
 	})
 }
 
+// makeFakeMiddleware returns a middleware function that appends the given identity
+//to the global invoked slice.
+func makeFakeMiddleware(identity string) MiddlewareFunc {
+	return func(next Handler) Handler {
+		return HandlerFunc(func(ctx context.Context, t *Task) error {
+			invoked = append(invoked, identity)
+			return next.ProcessTask(ctx, t)
+		})
+	}
+}
+
 // A list of pattern, handler pair that is registered with mux.
 var serveMuxRegister = []struct {
 	pattern string
@@ -114,3 +128,43 @@ func TestServeMuxNotFound(t *testing.T) {
 		}
 	}
 }
+
+var middlewareTests = []struct {
+	typename    string   // task's type name
+	middlewares []string // middlewares to use. They should be called in this order.
+	want        string   // identifier of the handler that should be called
+}{
+	{"email:signup", []string{"logging", "expiration"}, "signup email handler"},
+	{"csv:export", []string{}, "csv export handler"},
+	{"email:daily", []string{"expiration", "logging"}, "default email handler"},
+}
+
+func TestServeMuxMiddlewares(t *testing.T) {
+	for _, tc := range middlewareTests {
+		mux := NewServeMux()
+		for _, e := range serveMuxRegister {
+			mux.Handle(e.pattern, e.h)
+		}
+		var mws []MiddlewareFunc
+		for _, s := range tc.middlewares {
+			mws = append(mws, makeFakeMiddleware(s))
+		}
+		mux.Use(mws...)
+
+		invoked = []string{} // reset to empty slice
+		called = ""          // reset to zero value
+
+		task := NewTask(tc.typename, nil)
+		if err := mux.ProcessTask(context.Background(), task); err != nil {
+			t.Fatal(err)
+		}
+
+		if diff := cmp.Diff(invoked, tc.middlewares); diff != "" {
+			t.Errorf("invoked middlewares were %v, want %v", invoked, tc.middlewares)
+		}
+
+		if called != tc.want {
+			t.Errorf("%q handler was called for task %q, want %q to be called", called, task.Type, tc.want)
+		}
+	}
+}

tools/asynqmon/cmd/workers.go

@@ -22,7 +22,7 @@ var workersCmd = &cobra.Command{
 	Short: "Shows all running workers information",
 	Long: `Workers (asynqmon workers) will show all running workers information.
 
-The command shows the follwoing for each worker:
+The command shows the following for each worker:
 * Process in which the worker is running
 * ID of the task worker is processing
 * Type of the task worker is processing