v0.9.1

Add helper functions to extract task metadata from context
remove typo and redundant code
2025-10-20 21:26:14 +08:00 · 2020-05-29 05:42:40 -07:00 · 2020-05-29 05:40:42 -07:00 · 2020-05-22 05:11:54 -07:00 · 2020-05-17 13:25:24 -07:00 · 2020-05-16 11:00:44 -07:00
25 changed files with 1657 additions and 344 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,41 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

 ## [Unreleased]

+## [0.9.1] - 2020-05-29
+
+### Added
+
+- `GetTaskID`, `GetRetryCount`, and `GetMaxRetry` functions were added to extract task metadata from context.
+
+## [0.9.0] - 2020-05-16
+
+### Changed
+
+- `Logger` interface has changed. Please see the godoc for the new interface.
+
+### Added
+
+- `LogLevel` type is added. Server's log level can be specified through `LogLevel` field in `Config`.
+
+## [0.8.3] - 2020-05-08
+
+### Added
+
+- `Close` method is added to `Client`.
+
+## [0.8.2] - 2020-05-03
+
+### Fixed
+
+- [Fixed cancelfunc leak](https://github.com/hibiken/asynq/pull/145)
+
+## [0.8.1] - 2020-04-27
+
+### Added
+
+- `ParseRedisURI` helper function is added to create a `RedisConnOpt` from a URI string.
+- `SetDefaultOptions` method is added to `Client`.
+
 ## [0.8.0] - 2020-04-19

 ### Changed
--- a/README.md
+++ b/README.md
@@ -15,7 +15,7 @@ Highlevel overview of how Asynq works:

 - Client puts task on a queue
 - Server pulls task off queues and starts a worker goroutine for each task
- Workers process tasks concurrently
+- Tasks are processed concurrently by multiple workers

 Task queues are used as a mechanism to distribute work across multiple machines.  
 A system can consist of multiple worker servers and brokers, giving way to high availability and horizontal scaling.
@@ -24,24 +24,24 @@ A system can consist of multiple worker servers and brokers, giving way to high

 ## Stability and Compatibility

-**Important Note**: Current major version is zero (v0.x.x) to accomodate rapid development and fast iteration while getting early feedback from users. The public API could change without a major version update before v1.0.0 release.
+**Important Note**: Current major version is zero (v0.x.x) to accomodate rapid development and fast iteration while getting early feedback from users (Feedback on APIs are appreciated!). The public API could change without a major version update before v1.0.0 release.

 **Status**: The library is currently undergoing heavy development with frequent, breaking API changes.

 ## Features

- Guaranteed at least one execution of a task
+- Guaranteed [at least one execution](https://www.cloudcomputingpatterns.org/at_least_once_delivery/) of a task
 - Scheduling of tasks
 - Durability since tasks are written to Redis
- Retries of failed tasks
- Concurrency management via configuration
- Weighted priority queues
- Strict priority queues
+- [Retries](https://github.com/hibiken/asynq/wiki/Task-Retry) of failed tasks
+- [Weighted priority queues](https://github.com/hibiken/asynq/wiki/Priority-Queues#weighted-priority-queues)
+- [Strict priority queues](https://github.com/hibiken/asynq/wiki/Priority-Queues#strict-priority-queues)
 - Low latency to add a task since writes are fast in Redis
- De-duplication of tasks using unique option
- Allow timeout and deadline per task
- Flexible handler interface with support for middlewares
- CLI to inspect and remote-control queues and tasks
+- De-duplication of tasks using [unique option](https://github.com/hibiken/asynq/wiki/Unique-Tasks)
+- Allow [timeout and deadline per task](https://github.com/hibiken/asynq/wiki/Task-Timeout-and-Cancelation)
+- [Flexible handler interface with support for middlewares](https://github.com/hibiken/asynq/wiki/Handler-Deep-Dive)
+- [Support Redis Sentinels](https://github.com/hibiken/asynq/wiki/Automatic-Failover) for HA
+- [CLI](#command-line-tool) to inspect and remote-control queues and tasks

 ## Quickstart

@@ -62,13 +62,16 @@ import (
    "github.com/hibiken/asynq"
 )

-// A list of background task types.
+// A list of task types.
 const (
    EmailDelivery   = "email:deliver"
    ImageProcessing = "image:process"
 )

-// Write function NewXXXTask to create a task.
+//----------------------------------------------
+// Write a function NewXXXTask to create a task.
+// A task consists of a type and a payload.
+//----------------------------------------------

 func NewEmailDeliveryTask(userID int, tmplID string) *asynq.Task {
    payload := map[string]interface{}{"user_id": userID, "template_id": tmplID}
@@ -80,8 +83,13 @@ func NewImageProcessingTask(src, dst string) *asynq.Task {
    return asynq.NewTask(ImageProcessing, payload)
 }

-// Write function HandleXXXTask to handle the given task.
-// NOTE: It satisfies the asynq.HandlerFunc interface.
+//---------------------------------------------------------------
+// Write a function HandleXXXTask to handle the input task.
+// Note that it satisfies the asynq.HandlerFunc interface.
+// 
+// Handler doesn't need to be a function. You can define a type 
+// that satisfies asynq.Handler interface. See examples below.
+//---------------------------------------------------------------

 func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
    userID, err := t.Payload.GetInt("user_id")
@@ -97,7 +105,12 @@ func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
    return nil
 }

-func HandleImageProcessingTask(ctx context.Context, t *asynq.Task) error {
+// ImageProcessor implements asynq.Handler interface.
+type ImageProcesser struct {
+    // ... fields for struct
+}
+
+func (p *ImageProcessor) ProcessTask(ctx context.Context, t *asynq.Task) error {
    src, err := t.Payload.GetString("src")
    if err != nil {
        return err
@@ -110,6 +123,10 @@ func HandleImageProcessingTask(ctx context.Context, t *asynq.Task) error {
    // Image processing logic ...
    return nil
 }
+
+func NewImageProcessor() *ImageProcessor {
+    // ... return an instance
+}
 ```

 In your web application code, import the above package and use [`Client`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Client) to put tasks on the queue.  
@@ -131,8 +148,12 @@ const redisAddr = "127.0.0.1:6379"
 func main() {
    r := asynq.RedisClientOpt{Addr: redisAddr}
    c := asynq.NewClient(r)
+    defer c.Close()

+    // ------------------------------------------------------
    // Example 1: Enqueue task to be processed immediately.
+    //            Use (*Client).Enqueue method.
+    // ------------------------------------------------------

    t := tasks.NewEmailDeliveryTask(42, "some:template:id")
    err := c.Enqueue(t)
@@ -141,7 +162,10 @@ func main() {
    }


+    // ------------------------------------------------------------
    // Example 2: Schedule task to be processed in the future.
+    //            Use (*Client).EnqueueIn or (*Client).EnqueueAt.
+    // ------------------------------------------------------------

    t = tasks.NewEmailDeliveryTask(42, "other:template:id")
    err = c.EnqueueIn(24*time.Hour, t)
@@ -150,18 +174,33 @@ func main() {
    }


-    // Example 3: Pass options to tune task processing behavior.
-    // Options include MaxRetry, Queue, Timeout, Deadline, Unique etc.
+    // ----------------------------------------------------------------------------
+    // Example 3: Set options to tune task processing behavior.
+    //            Options include MaxRetry, Queue, Timeout, Deadline, Unique etc.
+    // ----------------------------------------------------------------------------
+
+    c.SetDefaultOptions(tasks.ImageProcessing, asynq.MaxRetry(10), 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))
+    err = c.Enqueue(t)
+    if err != nil {
+        log.Fatal("could not enqueue task: %v", err)
+    }
+
+    // ---------------------------------------------------------------------------
+    // Example 4: Pass options to tune task processing behavior at enqueue time.
+    //            Options passed at enqueue time override default ones, if any.
+    // ---------------------------------------------------------------------------
+
+    t = tasks.NewImageProcessingTask("some/blobstore/url", "other/blobstore/url")
+    err = c.Enqueue(t, asynq.Queue("critical"), asynq.Timeout(30*time.Second))
    if err != nil {
        log.Fatal("could not enqueue task: %v", err)
    }
 }
 ```

-Next, create a work server binary to process these tasks in the background.  
+Next, create a worker server to process these tasks in the background.  
 To start the background workers, use [`Server`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Server) and provide your [`Handler`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Handler) to process the tasks.

 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.
@@ -170,6 +209,8 @@ You can optionally use [`ServeMux`](https://pkg.go.dev/github.com/hibiken/asynq?
 package main

 import (
+    "log"
+
    "github.com/hibiken/asynq"
    "your/app/package/tasks"
 )
@@ -194,7 +235,7 @@ func main() {
    // mux maps a type to a handler
    mux := asynq.NewServeMux()
    mux.HandleFunc(tasks.EmailDelivery, tasks.HandleEmailDeliveryTask)
-    mux.HandleFunc(tasks.ImageProcessing, tasks.HandleImageProcessingTask)
+    mux.Handle(tasks.ImageProcessing, tasks.NewImageProcessor())
    // ...register other handlers...

    if err := srv.Run(mux); err != nil {
--- a/asynq.go
+++ b/asynq.go
@@ -7,6 +7,9 @@ package asynq
 import (
 	"crypto/tls"
 	"fmt"
+	"net/url"
+	"strconv"
+	"strings"

 	"github.com/go-redis/redis/v7"
 )
@@ -94,6 +97,79 @@ type RedisFailoverClientOpt struct {
 	TLSConfig *tls.Config
 }

+// ParseRedisURI parses redis uri string and returns RedisConnOpt if uri is valid.
+// It returns a non-nil error if uri cannot be parsed.
+//
+// Three URI schemes are supported, which are redis:, redis-socket:, and redis-sentinel:.
+// Supported formats are:
+//     redis://[:password@]host[:port][/dbnumber]
+//     redis-socket://[:password@]path[?db=dbnumber]
+//     redis-sentinel://[:password@]host1[:port][,host2:[:port]][,hostN:[:port]][?master=masterName]
+func ParseRedisURI(uri string) (RedisConnOpt, error) {
+	u, err := url.Parse(uri)
+	if err != nil {
+		return nil, fmt.Errorf("asynq: could not parse redis uri: %v", err)
+	}
+	switch u.Scheme {
+	case "redis":
+		return parseRedisURI(u)
+	case "redis-socket":
+		return parseRedisSocketURI(u)
+	case "redis-sentinel":
+		return parseRedisSentinelURI(u)
+	default:
+		return nil, fmt.Errorf("asynq: unsupported uri scheme: %q", u.Scheme)
+	}
+}
+
+func parseRedisURI(u *url.URL) (RedisConnOpt, error) {
+	var db int
+	var err error
+	if len(u.Path) > 0 {
+		xs := strings.Split(strings.Trim(u.Path, "/"), "/")
+		db, err = strconv.Atoi(xs[0])
+		if err != nil {
+			return nil, fmt.Errorf("asynq: could not parse redis uri: database number should be the first segment of the path")
+		}
+	}
+	var password string
+	if v, ok := u.User.Password(); ok {
+		password = v
+	}
+	return RedisClientOpt{Addr: u.Host, DB: db, Password: password}, nil
+}
+
+func parseRedisSocketURI(u *url.URL) (RedisConnOpt, error) {
+	const errPrefix = "asynq: could not parse redis socket uri"
+	if len(u.Path) == 0 {
+		return nil, fmt.Errorf("%s: path does not exist", errPrefix)
+	}
+	q := u.Query()
+	var db int
+	var err error
+	if n := q.Get("db"); n != "" {
+		db, err = strconv.Atoi(n)
+		if err != nil {
+			return nil, fmt.Errorf("%s: query param `db` should be a number", errPrefix)
+		}
+	}
+	var password string
+	if v, ok := u.User.Password(); ok {
+		password = v
+	}
+	return RedisClientOpt{Network: "unix", Addr: u.Path, DB: db, Password: password}, nil
+}
+
+func parseRedisSentinelURI(u *url.URL) (RedisConnOpt, error) {
+	addrs := strings.Split(u.Host, ",")
+	master := u.Query().Get("master")
+	var password string
+	if v, ok := u.User.Password(); ok {
+		password = v
+	}
+	return RedisFailoverClientOpt{MasterName: master, SentinelAddrs: addrs, Password: password}, nil
+}
+
 // createRedisClient returns a redis client given a redis connection configuration.
 //
 // Passing an unexpected type as a RedisConnOpt argument will cause panic.
--- a/asynq_test.go
+++ b/asynq_test.go
@@ -5,7 +5,7 @@
 package asynq

 import (
-	"os"
+	"flag"
 	"sort"
 	"testing"

@@ -15,16 +15,28 @@ import (
 	"github.com/hibiken/asynq/internal/log"
 )

-// This file defines test helper functions used by
-// other test files.
+//============================================================================
+// This file defines helper functions and variables used in other test files.
+//============================================================================

-// redis used for package testing.
-const (
-	redisAddr = "localhost:6379"
-	redisDB   = 14
+// variables used for package testing.
+var (
+	redisAddr string
+	redisDB   int
+
+	testLogLevel = FatalLevel
 )

-var testLogger = log.NewLogger(os.Stderr)
+var testLogger *log.Logger
+
+func init() {
+	flag.StringVar(&redisAddr, "redis_addr", "localhost:6379", "redis address to use in testing")
+	flag.IntVar(&redisDB, "redis_db", 14, "redis db number to use in testing")
+	flag.Var(&testLogLevel, "loglevel", "log level to use in testing")
+
+	testLogger = log.NewLogger(nil)
+	testLogger.SetLevel(toInternalLogLevel(testLogLevel))
+}

 func setup(tb testing.TB) *redis.Client {
 	tb.Helper()
@@ -44,3 +56,106 @@ var sortTaskOpt = cmp.Transformer("SortMsg", func(in []*Task) []*Task {
 	})
 	return out
 })
+
+func TestParseRedisURI(t *testing.T) {
+	tests := []struct {
+		uri  string
+		want RedisConnOpt
+	}{
+		{
+			"redis://localhost:6379",
+			RedisClientOpt{Addr: "localhost:6379"},
+		},
+		{
+			"redis://localhost:6379/3",
+			RedisClientOpt{Addr: "localhost:6379", DB: 3},
+		},
+		{
+			"redis://:mypassword@localhost:6379",
+			RedisClientOpt{Addr: "localhost:6379", Password: "mypassword"},
+		},
+		{
+			"redis://:mypassword@127.0.0.1:6379/11",
+			RedisClientOpt{Addr: "127.0.0.1:6379", Password: "mypassword", DB: 11},
+		},
+		{
+			"redis-socket:///var/run/redis/redis.sock",
+			RedisClientOpt{Network: "unix", Addr: "/var/run/redis/redis.sock"},
+		},
+		{
+			"redis-socket://:mypassword@/var/run/redis/redis.sock",
+			RedisClientOpt{Network: "unix", Addr: "/var/run/redis/redis.sock", Password: "mypassword"},
+		},
+		{
+			"redis-socket:///var/run/redis/redis.sock?db=7",
+			RedisClientOpt{Network: "unix", Addr: "/var/run/redis/redis.sock", DB: 7},
+		},
+		{
+			"redis-socket://:mypassword@/var/run/redis/redis.sock?db=12",
+			RedisClientOpt{Network: "unix", Addr: "/var/run/redis/redis.sock", Password: "mypassword", DB: 12},
+		},
+		{
+			"redis-sentinel://localhost:5000,localhost:5001,localhost:5002?master=mymaster",
+			RedisFailoverClientOpt{
+				MasterName:    "mymaster",
+				SentinelAddrs: []string{"localhost:5000", "localhost:5001", "localhost:5002"},
+			},
+		},
+		{
+			"redis-sentinel://:mypassword@localhost:5000,localhost:5001,localhost:5002?master=mymaster",
+			RedisFailoverClientOpt{
+				MasterName:    "mymaster",
+				SentinelAddrs: []string{"localhost:5000", "localhost:5001", "localhost:5002"},
+				Password:      "mypassword",
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		got, err := ParseRedisURI(tc.uri)
+		if err != nil {
+			t.Errorf("ParseRedisURI(%q) returned an error: %v", tc.uri, err)
+			continue
+		}
+
+		if diff := cmp.Diff(tc.want, got); diff != "" {
+			t.Errorf("ParseRedisURI(%q) = %+v, want %+v\n(-want,+got)\n%s", tc.uri, got, tc.want, diff)
+		}
+	}
+}
+
+func TestParseRedisURIErrors(t *testing.T) {
+	tests := []struct {
+		desc string
+		uri  string
+	}{
+		{
+			"unsupported scheme",
+			"rdb://localhost:6379",
+		},
+		{
+			"missing scheme",
+			"localhost:6379",
+		},
+		{
+			"multiple db numbers",
+			"redis://localhost:6379/1,2,3",
+		},
+		{
+			"missing path for socket connection",
+			"redis-socket://?db=one",
+		},
+		{
+			"non integer for db numbers for socket",
+			"redis-socket:///some/path/to/redis?db=one",
+		},
+	}
+
+	for _, tc := range tests {
+		_, err := ParseRedisURI(tc.uri)
+		if err == nil {
+			t.Errorf("%s: ParseRedisURI(%q) succeeded for malformed input, want error",
+				tc.desc, tc.uri)
+		}
+	}
+}
--- a/client.go
+++ b/client.go
@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"sync"
 	"time"

 	"github.com/hibiken/asynq/internal/base"
@@ -23,13 +24,18 @@ import (
 //
 // Clients are safe for concurrent use by multiple goroutines.
 type Client struct {
-	rdb *rdb.RDB
+	mu   sync.Mutex
+	opts map[string][]Option
+	rdb  *rdb.RDB
 }

 // NewClient and returns a new Client given a redis connection option.
 func NewClient(r RedisConnOpt) *Client {
 	rdb := rdb.NewRDB(createRedisClient(r))
-	return &Client{rdb}
+	return &Client{
+		opts: make(map[string][]Option),
+		rdb:  rdb,
+	}
 }

 // Option specifies the task processing behavior.
@@ -159,10 +165,19 @@ func serializePayload(payload map[string]interface{}) string {
 	return b.String()
 }

-const (
-	// Max retry count by default
-	defaultMaxRetry = 25
-)
+// Default max retry count used if nothing is specified.
+const defaultMaxRetry = 25
+
+// SetDefaultOptions sets options to be used for a given task type.
+// The argument opts specifies the behavior of task processing.
+// If there are conflicting Option values the last one overrides others.
+//
+// Default options can be overridden by options passed at enqueue time.
+func (c *Client) SetDefaultOptions(taskType string, opts ...Option) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.opts[taskType] = opts
+}

 // EnqueueAt schedules task to be enqueued at the specified time.
 //
@@ -171,6 +186,40 @@ const (
 // The argument opts specifies the behavior of task processing.
 // If there are conflicting Option values the last one overrides others.
 func (c *Client) EnqueueAt(t time.Time, task *Task, opts ...Option) error {
+	return c.enqueueAt(t, task, opts...)
+}
+
+// Enqueue enqueues task to be processed immediately.
+//
+// Enqueue returns nil if the task is enqueued successfully, otherwise returns a non-nil error.
+//
+// The argument opts specifies the behavior of task processing.
+// If there are conflicting Option values the last one overrides others.
+func (c *Client) Enqueue(task *Task, opts ...Option) error {
+	return c.enqueueAt(time.Now(), task, opts...)
+}
+
+// EnqueueIn schedules task to be enqueued after the specified delay.
+//
+// EnqueueIn returns nil if the task is scheduled successfully, otherwise returns a non-nil error.
+//
+// The argument opts specifies the behavior of task processing.
+// If there are conflicting Option values the last one overrides others.
+func (c *Client) EnqueueIn(d time.Duration, task *Task, opts ...Option) error {
+	return c.enqueueAt(time.Now().Add(d), task, opts...)
+}
+
+// Close closes the connection with redis server.
+func (c *Client) Close() error {
+	return c.rdb.Close()
+}
+
+func (c *Client) enqueueAt(t time.Time, task *Task, opts ...Option) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if defaults, ok := c.opts[task.Type]; ok {
+		opts = append(defaults, opts...)
+	}
 	opt := composeOptions(opts...)
 	msg := &base.TaskMessage{
 		ID:        xid.New(),
@@ -194,26 +243,6 @@ func (c *Client) EnqueueAt(t time.Time, task *Task, opts ...Option) error {
 	return err
 }

-// Enqueue enqueues task to be processed immediately.
-//
-// Enqueue returns nil if the task is enqueued successfully, otherwise returns a non-nil error.
-//
-// The argument opts specifies the behavior of task processing.
-// If there are conflicting Option values the last one overrides others.
-func (c *Client) Enqueue(task *Task, opts ...Option) error {
-	return c.EnqueueAt(time.Now(), task, opts...)
-}
-
-// EnqueueIn schedules task to be enqueued after the specified delay.
-//
-// EnqueueIn returns nil if the task is scheduled successfully, otherwise returns a non-nil error.
-//
-// The argument opts specifies the behavior of task processing.
-// If there are conflicting Option values the last one overrides others.
-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, uniqueTTL time.Duration) error {
 	if uniqueTTL > 0 {
 		return c.rdb.EnqueueUnique(msg, uniqueTTL)
--- a/client_test.go
+++ b/client_test.go
@@ -15,6 +15,11 @@ import (
 	"github.com/hibiken/asynq/internal/base"
 )

+var (
+	noTimeout  = time.Duration(0).String()
+	noDeadline = time.Time{}.Format(time.RFC3339)
+)
+
 func TestClientEnqueueAt(t *testing.T) {
 	r := setup(t)
 	client := NewClient(RedisClientOpt{
@@ -27,9 +32,6 @@ func TestClientEnqueueAt(t *testing.T) {
 	var (
 		now          = time.Now()
 		oneHourLater = now.Add(time.Hour)
-
-		noTimeout  = time.Duration(0).String()
-		noDeadline = time.Time{}.Format(time.RFC3339)
 	)

 	tests := []struct {
@@ -113,11 +115,6 @@ func TestClientEnqueue(t *testing.T) {

 	task := NewTask("send_email", map[string]interface{}{"to": "customer@gmail.com", "from": "merchant@example.com"})

-	var (
-		noTimeout  = time.Duration(0).String()
-		noDeadline = time.Time{}.Format(time.RFC3339)
-	)
-
 	tests := []struct {
 		desc         string
 		task         *Task
@@ -287,11 +284,6 @@ func TestClientEnqueueIn(t *testing.T) {

 	task := NewTask("send_email", map[string]interface{}{"to": "customer@gmail.com", "from": "merchant@example.com"})

-	var (
-		noTimeout  = time.Duration(0).String()
-		noDeadline = time.Time{}.Format(time.RFC3339)
-	)
-
 	tests := []struct {
 		desc          string
 		task          *Task
@@ -364,6 +356,86 @@ func TestClientEnqueueIn(t *testing.T) {
 	}
 }

+func TestClientDefaultOptions(t *testing.T) {
+	r := setup(t)
+
+	tests := []struct {
+		desc        string
+		defaultOpts []Option // options set at the client level.
+		opts        []Option // options used at enqueue time.
+		task        *Task
+		queue       string // queue that the message should go into.
+		want        *base.TaskMessage
+	}{
+		{
+			desc:        "With queue routing option",
+			defaultOpts: []Option{Queue("feed")},
+			opts:        []Option{},
+			task:        NewTask("feed:import", nil),
+			queue:       "feed",
+			want: &base.TaskMessage{
+				Type:     "feed:import",
+				Payload:  nil,
+				Retry:    defaultMaxRetry,
+				Queue:    "feed",
+				Timeout:  noTimeout,
+				Deadline: noDeadline,
+			},
+		},
+		{
+			desc:        "With multiple options",
+			defaultOpts: []Option{Queue("feed"), MaxRetry(5)},
+			opts:        []Option{},
+			task:        NewTask("feed:import", nil),
+			queue:       "feed",
+			want: &base.TaskMessage{
+				Type:     "feed:import",
+				Payload:  nil,
+				Retry:    5,
+				Queue:    "feed",
+				Timeout:  noTimeout,
+				Deadline: noDeadline,
+			},
+		},
+		{
+			desc:        "With overriding options at enqueue time",
+			defaultOpts: []Option{Queue("feed"), MaxRetry(5)},
+			opts:        []Option{Queue("critical")},
+			task:        NewTask("feed:import", nil),
+			queue:       "critical",
+			want: &base.TaskMessage{
+				Type:     "feed:import",
+				Payload:  nil,
+				Retry:    5,
+				Queue:    "critical",
+				Timeout:  noTimeout,
+				Deadline: noDeadline,
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r)
+		c := NewClient(RedisClientOpt{Addr: redisAddr, DB: redisDB})
+		c.SetDefaultOptions(tc.task.Type, tc.defaultOpts...)
+		err := c.Enqueue(tc.task, tc.opts...)
+		if err != nil {
+			t.Fatal(err)
+		}
+		enqueued := h.GetEnqueuedMessages(t, r, tc.queue)
+		if len(enqueued) != 1 {
+			t.Errorf("%s;\nexpected queue %q to have one message; got %d messages in the queue.",
+				tc.desc, tc.queue, len(enqueued))
+			continue
+		}
+		got := enqueued[0]
+		if diff := cmp.Diff(tc.want, got, h.IgnoreIDOpt); diff != "" {
+			t.Errorf("%s;\nmismatch found in enqueued task message; (-want,+got)\n%s",
+				tc.desc, diff)
+		}
+	}
+}
+
 func TestUniqueKey(t *testing.T) {
 	tests := []struct {
 		desc  string
--- a/context.go
+++ b/context.go
@@ -0,0 +1,85 @@
+// Copyright 2020 Kentaro Hibino. All rights reserved.
+// Use of this source code is governed by a MIT license
+// that can be found in the LICENSE file.
+
+package asynq
+
+import (
+	"context"
+	"time"
+
+	"github.com/hibiken/asynq/internal/base"
+)
+
+// A taskMetadata holds task scoped data to put in context.
+type taskMetadata struct {
+	id         string
+	maxRetry   int
+	retryCount int
+}
+
+// ctxKey type is unexported to prevent collisions with context keys defined in
+// other packages.
+type ctxKey int
+
+// metadataCtxKey is the context key for the task metadata.
+// Its value of zero is arbitrary.
+const metadataCtxKey ctxKey = 0
+
+// createContext returns a context and cancel function for a given task message.
+func createContext(msg *base.TaskMessage) (ctx context.Context, cancel context.CancelFunc) {
+	metadata := taskMetadata{
+		id:         msg.ID.String(),
+		maxRetry:   msg.Retry,
+		retryCount: msg.Retried,
+	}
+	ctx = context.WithValue(context.Background(), metadataCtxKey, metadata)
+	timeout, err := time.ParseDuration(msg.Timeout)
+	if err == nil && timeout != 0 {
+		ctx, cancel = context.WithTimeout(ctx, timeout)
+	}
+	deadline, err := time.Parse(time.RFC3339, msg.Deadline)
+	if err == nil && !deadline.IsZero() {
+		ctx, cancel = context.WithDeadline(ctx, deadline)
+	}
+	if cancel == nil {
+		ctx, cancel = context.WithCancel(ctx)
+	}
+	return ctx, cancel
+}
+
+// GetTaskID extracts a task ID from a context, if any.
+//
+// ID of a task is guaranteed to be unique.
+// ID of a task doesn't change if the task is being retried.
+func GetTaskID(ctx context.Context) (id string, ok bool) {
+	metadata, ok := ctx.Value(metadataCtxKey).(taskMetadata)
+	if !ok {
+		return "", false
+	}
+	return metadata.id, true
+}
+
+// GetRetryCount extracts retry count from a context, if any.
+//
+// Return value n indicates the number of times associated task has been
+// retried so far.
+func GetRetryCount(ctx context.Context) (n int, ok bool) {
+	metadata, ok := ctx.Value(metadataCtxKey).(taskMetadata)
+	if !ok {
+		return 0, false
+	}
+	return metadata.retryCount, true
+}
+
+// GetMaxRetry extracts maximum retry from a context, if any.
+//
+// Return value n indicates the maximum number of times the assoicated task
+// can be retried if ProcessTask returns a non-nil error.
+func GetMaxRetry(ctx context.Context) (n int, ok bool) {
+	metadata, ok := ctx.Value(metadataCtxKey).(taskMetadata)
+	if !ok {
+		return 0, false
+	}
+	return metadata.maxRetry, true
+}
--- a/context_test.go
+++ b/context_test.go
@@ -0,0 +1,157 @@
+// Copyright 2020 Kentaro Hibino. All rights reserved.
+// Use of this source code is governed by a MIT license
+// that can be found in the LICENSE file.
+
+package asynq
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/hibiken/asynq/internal/base"
+	"github.com/rs/xid"
+)
+
+func TestCreateContextWithTimeRestrictions(t *testing.T) {
+	var (
+		noTimeout  = time.Duration(0)
+		noDeadline = time.Time{}
+	)
+
+	tests := []struct {
+		desc         string
+		timeout      time.Duration
+		deadline     time.Time
+		wantDeadline time.Time
+	}{
+		{"only with timeout", 10 * time.Second, noDeadline, time.Now().Add(10 * time.Second)},
+		{"only with deadline", noTimeout, time.Now().Add(time.Hour), time.Now().Add(time.Hour)},
+		{"with timeout and deadline (timeout < deadline)", 10 * time.Second, time.Now().Add(time.Hour), time.Now().Add(10 * time.Second)},
+		{"with timeout and deadline (timeout > deadline)", 10 * time.Minute, time.Now().Add(30 * time.Second), time.Now().Add(30 * time.Second)},
+	}
+
+	for _, tc := range tests {
+		msg := &base.TaskMessage{
+			Type:     "something",
+			ID:       xid.New(),
+			Timeout:  tc.timeout.String(),
+			Deadline: tc.deadline.Format(time.RFC3339),
+		}
+
+		ctx, cancel := createContext(msg)
+
+		select {
+		case x := <-ctx.Done():
+			t.Errorf("%s: <-ctx.Done() == %v, want nothing (it should block)", tc.desc, x)
+		default:
+		}
+
+		got, ok := ctx.Deadline()
+		if !ok {
+			t.Errorf("%s: ctx.Deadline() returned false, want deadline to be set", tc.desc)
+		}
+		if !cmp.Equal(tc.wantDeadline, got, cmpopts.EquateApproxTime(time.Second)) {
+			t.Errorf("%s: ctx.Deadline() returned %v, want %v", tc.desc, got, tc.wantDeadline)
+		}
+
+		cancel()
+
+		select {
+		case <-ctx.Done():
+		default:
+			t.Errorf("ctx.Done() blocked, want it to be non-blocking")
+		}
+	}
+}
+
+func TestCreateContextWithoutTimeRestrictions(t *testing.T) {
+	msg := &base.TaskMessage{
+		Type:     "something",
+		ID:       xid.New(),
+		Timeout:  time.Duration(0).String(),        // zero value to indicate no timeout
+		Deadline: time.Time{}.Format(time.RFC3339), // zero value to indicate no deadline
+	}
+
+	ctx, cancel := createContext(msg)
+
+	select {
+	case x := <-ctx.Done():
+		t.Errorf("<-ctx.Done() == %v, want nothing (it should block)", x)
+	default:
+	}
+
+	_, ok := ctx.Deadline()
+	if ok {
+		t.Error("ctx.Deadline() returned true, want deadline to not be set")
+	}
+
+	cancel()
+
+	select {
+	case <-ctx.Done():
+	default:
+		t.Error("ctx.Done() blocked, want it to be non-blocking")
+	}
+}
+
+func TestGetTaskMetadataFromContext(t *testing.T) {
+	tests := []struct {
+		desc string
+		msg  *base.TaskMessage
+	}{
+		{"with zero retried message", &base.TaskMessage{Type: "something", ID: xid.New(), Retry: 25, Retried: 0}},
+		{"with non-zero retried message", &base.TaskMessage{Type: "something", ID: xid.New(), Retry: 10, Retried: 5}},
+	}
+
+	for _, tc := range tests {
+		ctx, _ := createContext(tc.msg)
+
+		id, ok := GetTaskID(ctx)
+		if !ok {
+			t.Errorf("%s: GetTaskID(ctx) returned ok == false", tc.desc)
+		}
+		if ok && id != tc.msg.ID.String() {
+			t.Errorf("%s: GetTaskID(ctx) returned id == %q, want %q", tc.desc, id, tc.msg.ID.String())
+		}
+
+		retried, ok := GetRetryCount(ctx)
+		if !ok {
+			t.Errorf("%s: GetRetryCount(ctx) returned ok == false", tc.desc)
+		}
+		if ok && retried != tc.msg.Retried {
+			t.Errorf("%s: GetRetryCount(ctx) returned n == %d want %d", tc.desc, retried, tc.msg.Retried)
+		}
+
+		maxRetry, ok := GetMaxRetry(ctx)
+		if !ok {
+			t.Errorf("%s: GetMaxRetry(ctx) returned ok == false", tc.desc)
+		}
+		if ok && maxRetry != tc.msg.Retry {
+			t.Errorf("%s: GetMaxRetry(ctx) returned n == %d want %d", tc.desc, maxRetry, tc.msg.Retry)
+		}
+	}
+}
+
+func TestGetTaskMetadataFromContextError(t *testing.T) {
+	tests := []struct {
+		desc string
+		ctx  context.Context
+	}{
+		{"with background context", context.Background()},
+	}
+
+	for _, tc := range tests {
+		if _, ok := GetTaskID(tc.ctx); ok {
+			t.Errorf("%s: GetTaskID(ctx) returned ok == true", tc.desc)
+		}
+		if _, ok := GetRetryCount(tc.ctx); ok {
+			t.Errorf("%s: GetRetryCount(ctx) returned ok == true", tc.desc)
+		}
+		if _, ok := GetMaxRetry(tc.ctx); ok {
+			t.Errorf("%s: GetMaxRetry(ctx) returned ok == true", tc.desc)
+		}
+	}
+}
--- a/example_test.go
+++ b/example_test.go
@@ -0,0 +1,95 @@
+// Copyright 2020 Kentaro Hibino. All rights reserved.
+// Use of this source code is governed by a MIT license
+// that can be found in the LICENSE file.
+
+package asynq_test
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"os/signal"
+
+	"github.com/hibiken/asynq"
+	"golang.org/x/sys/unix"
+)
+
+func ExampleServer_Run() {
+	srv := asynq.NewServer(
+		asynq.RedisClientOpt{Addr: ":6379"},
+		asynq.Config{Concurrency: 20},
+	)
+
+	h := asynq.NewServeMux()
+	// ... Register handlers
+
+	// Run blocks and waits for os signal to terminate the program.
+	if err := srv.Run(h); err != nil {
+		log.Fatal(err)
+	}
+}
+
+func ExampleServer_Stop() {
+	srv := asynq.NewServer(
+		asynq.RedisClientOpt{Addr: ":6379"},
+		asynq.Config{Concurrency: 20},
+	)
+
+	h := asynq.NewServeMux()
+	// ... Register handlers
+
+	if err := srv.Start(h); err != nil {
+		log.Fatal(err)
+	}
+
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, unix.SIGTERM, unix.SIGINT)
+	<-sigs // wait for termination signal
+
+	srv.Stop()
+}
+
+func ExampleServer_Quiet() {
+	srv := asynq.NewServer(
+		asynq.RedisClientOpt{Addr: ":6379"},
+		asynq.Config{Concurrency: 20},
+	)
+
+	h := asynq.NewServeMux()
+	// ... Register handlers
+
+	if err := srv.Start(h); err != nil {
+		log.Fatal(err)
+	}
+
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, unix.SIGTERM, unix.SIGINT, unix.SIGTSTP)
+	// Handle SIGTERM, SIGINT to exit the program.
+	// Handle SIGTSTP to stop processing new tasks.
+	for {
+		s := <-sigs
+		if s == unix.SIGTSTP {
+			srv.Quiet() // stop processing new tasks
+			continue
+		}
+		break
+	}
+
+	srv.Stop()
+}
+
+func ExampleParseRedisURI() {
+	rconn, err := asynq.ParseRedisURI("redis://localhost:6379/10")
+	if err != nil {
+		log.Fatal(err)
+	}
+	r, ok := rconn.(asynq.RedisClientOpt)
+	if !ok {
+		log.Fatal("unexpected type")
+	}
+	fmt.Println(r.Addr)
+	fmt.Println(r.DB)
+	// Output:
+	// localhost:6379
+	// 10
+}
--- a/heartbeat.go
+++ b/heartbeat.go
@@ -9,12 +9,13 @@ import (
 	"time"

 	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/log"
 )

 // heartbeater is responsible for writing process info to redis periodically to
 // indicate that the background worker process is up.
 type heartbeater struct {
-	logger Logger
+	logger *log.Logger
 	broker base.Broker

 	ss *base.ServerState
@@ -26,18 +27,25 @@ type heartbeater struct {
 	interval time.Duration
 }

-func newHeartbeater(l Logger, b base.Broker, ss *base.ServerState, interval time.Duration) *heartbeater {
+type heartbeaterParams struct {
+	logger      *log.Logger
+	broker      base.Broker
+	serverState *base.ServerState
+	interval    time.Duration
+}
+
+func newHeartbeater(params heartbeaterParams) *heartbeater {
 	return &heartbeater{
-		logger:   l,
-		broker:   b,
-		ss:       ss,
+		logger:   params.logger,
+		broker:   params.broker,
+		ss:       params.serverState,
 		done:     make(chan struct{}),
-		interval: interval,
+		interval: params.interval,
 	}
 }

 func (h *heartbeater) terminate() {
-	h.logger.Info("Heartbeater shutting down...")
+	h.logger.Debug("Heartbeater shutting down...")
 	// Signal the heartbeater goroutine to stop.
 	h.done <- struct{}{}
 }
@@ -53,7 +61,7 @@ func (h *heartbeater) start(wg *sync.WaitGroup) {
 			select {
 			case <-h.done:
 				h.broker.ClearServerState(h.ss)
-				h.logger.Info("Heartbeater done")
+				h.logger.Debug("Heartbeater done")
 				return
 			case <-time.After(h.interval):
 				h.beat()
@@ -67,6 +75,6 @@ func (h *heartbeater) beat() {
 	// and short enough to expire quickly once the process is shut down or killed.
 	err := h.broker.WriteServerState(h.ss, h.interval*2)
 	if err != nil {
-		h.logger.Error("could not write heartbeat data: %v", err)
+		h.logger.Errorf("could not write heartbeat data: %v", err)
 	}
 }
--- a/heartbeat_test.go
+++ b/heartbeat_test.go
@@ -38,7 +38,12 @@ func TestHeartbeater(t *testing.T) {
 		h.FlushDB(t, r)

 		state := base.NewServerState(tc.host, tc.pid, tc.concurrency, tc.queues, false)
-		hb := newHeartbeater(testLogger, rdbClient, state, tc.interval)
+		hb := newHeartbeater(heartbeaterParams{
+			logger:      testLogger,
+			broker:      rdbClient,
+			serverState: state,
+			interval:    tc.interval,
+		})

 		var wg sync.WaitGroup
 		hb.start(&wg)
@@ -115,7 +120,12 @@ func TestHeartbeaterWithRedisDown(t *testing.T) {
 	r := rdb.NewRDB(setup(t))
 	testBroker := testbroker.NewTestBroker(r)
 	ss := base.NewServerState("localhost", 1234, 10, map[string]int{"default": 1}, false)
-	hb := newHeartbeater(testLogger, testBroker, ss, time.Second)
+	hb := newHeartbeater(heartbeaterParams{
+		logger:      testLogger,
+		broker:      testBroker,
+		serverState: ss,
+		interval:    time.Second,
+	})

 	testBroker.Sleep()
 	var wg sync.WaitGroup
--- a/internal/base/base_test.go
+++ b/internal/base/base_test.go
@@ -82,7 +82,8 @@ func TestServerInfoKey(t *testing.T) {
 	for _, tc := range tests {
 		got := ServerInfoKey(tc.hostname, tc.pid, tc.sid)
 		if got != tc.want {
-			t.Errorf("ServerInfoKey(%q, %d) = %q, want %q", tc.hostname, tc.pid, got, tc.want)
+			t.Errorf("ServerInfoKey(%q, %d, %q) = %q, want %q",
+				tc.hostname, tc.pid, tc.sid, got, tc.want)
 		}
 	}
 }
@@ -101,7 +102,8 @@ func TestWorkersKey(t *testing.T) {
 	for _, tc := range tests {
 		got := WorkersKey(tc.hostname, tc.pid, tc.sid)
 		if got != tc.want {
-			t.Errorf("WorkersKey(%q, %d) = %q, want = %q", tc.hostname, tc.pid, got, tc.want)
+			t.Errorf("WorkersKey(%q, %d, %q) = %q, want = %q",
+				tc.hostname, tc.pid, tc.sid, got, tc.want)
 		}
 	}
 }
--- a/internal/log/log.go
+++ b/internal/log/log.go
@@ -6,52 +6,210 @@
 package log

 import (
+	"fmt"
 	"io"
 	stdlog "log"
 	"os"
+	"sync"
 )

-// 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),
-	}
+// Base supports logging at various log levels.
+type Base interface {
+	// Debug logs a message at Debug level.
+	Debug(args ...interface{})
+
+	// Info logs a message at Info level.
+	Info(args ...interface{})
+
+	// Warn logs a message at Warning level.
+	Warn(args ...interface{})
+
+	// Error logs a message at Error level.
+	Error(args ...interface{})
+
+	// Fatal logs a message at Fatal level
+	// and process will exit with status set to 1.
+	Fatal(args ...interface{})
 }

-// Logger is a wrapper object around log.Logger from the standard library.
+// baseLogger is a wrapper object around log.Logger from the standard library.
 // It supports logging at various log levels.
-type Logger struct {
+type baseLogger 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...)
+func (l *baseLogger) Debug(args ...interface{}) {
+	l.prefixPrint("DEBUG: ", args...)
 }

 // Info logs a message at Info level.
-func (l *Logger) Info(format string, args ...interface{}) {
-	format = "INFO: " + format
-	l.Printf(format, args...)
+func (l *baseLogger) Info(args ...interface{}) {
+	l.prefixPrint("INFO: ", args...)
 }

 // Warn logs a message at Warning level.
-func (l *Logger) Warn(format string, args ...interface{}) {
-	format = "WARN: " + format
-	l.Printf(format, args...)
+func (l *baseLogger) Warn(args ...interface{}) {
+	l.prefixPrint("WARN: ", args...)
 }

 // Error logs a message at Error level.
-func (l *Logger) Error(format string, args ...interface{}) {
-	format = "ERROR: " + format
-	l.Printf(format, args...)
+func (l *baseLogger) Error(args ...interface{}) {
+	l.prefixPrint("ERROR: ", 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...)
+func (l *baseLogger) Fatal(args ...interface{}) {
+	l.prefixPrint("FATAL: ", args...)
 	os.Exit(1)
 }
+
+func (l *baseLogger) prefixPrint(prefix string, args ...interface{}) {
+	args = append([]interface{}{prefix}, args...)
+	l.Print(args...)
+}
+
+// newBase creates and returns a new instance of baseLogger.
+func newBase(out io.Writer) *baseLogger {
+	prefix := fmt.Sprintf("asynq: pid=%d ", os.Getpid())
+	return &baseLogger{
+		stdlog.New(out, prefix, stdlog.Ldate|stdlog.Ltime|stdlog.Lmicroseconds|stdlog.LUTC),
+	}
+}
+
+// NewLogger creates and returns a new instance of Logger.
+// Log level is set to DebugLevel by default.
+func NewLogger(base Base) *Logger {
+	if base == nil {
+		base = newBase(os.Stderr)
+	}
+	return &Logger{base: base, level: DebugLevel}
+}
+
+// Logger logs message to io.Writer at various log levels.
+type Logger struct {
+	base Base
+
+	mu sync.Mutex
+	// Minimum log level for this logger.
+	// Message with level lower than this level won't be outputted.
+	level Level
+}
+
+// Level represents a log level.
+type Level int32
+
+const (
+	// DebugLevel is the lowest level of logging.
+	// Debug logs are intended for debugging and development purposes.
+	DebugLevel Level = iota
+
+	// InfoLevel is used for general informational log messages.
+	InfoLevel
+
+	// WarnLevel is used for undesired but relatively expected events,
+	// which may indicate a problem.
+	WarnLevel
+
+	// ErrorLevel is used for undesired and unexpected events that
+	// the program can recover from.
+	ErrorLevel
+
+	// FatalLevel is used for undesired and unexpected events that
+	// the program cannot recover from.
+	FatalLevel
+)
+
+// String is part of the fmt.Stringer interface.
+//
+// Used for testing and debugging purposes.
+func (l Level) String() string {
+	switch l {
+	case DebugLevel:
+		return "debug"
+	case InfoLevel:
+		return "info"
+	case WarnLevel:
+		return "warning"
+	case ErrorLevel:
+		return "error"
+	case FatalLevel:
+		return "fatal"
+	default:
+		return "unknown"
+	}
+}
+
+// canLogAt reports whether logger can log at level v.
+func (l *Logger) canLogAt(v Level) bool {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return v >= l.level
+}
+
+func (l *Logger) Debug(args ...interface{}) {
+	if !l.canLogAt(DebugLevel) {
+		return
+	}
+	l.base.Debug(args...)
+}
+
+func (l *Logger) Info(args ...interface{}) {
+	if !l.canLogAt(InfoLevel) {
+		return
+	}
+	l.base.Info(args...)
+}
+
+func (l *Logger) Warn(args ...interface{}) {
+	if !l.canLogAt(WarnLevel) {
+		return
+	}
+	l.base.Warn(args...)
+}
+
+func (l *Logger) Error(args ...interface{}) {
+	if !l.canLogAt(ErrorLevel) {
+		return
+	}
+	l.base.Error(args...)
+}
+
+func (l *Logger) Fatal(args ...interface{}) {
+	if !l.canLogAt(FatalLevel) {
+		return
+	}
+	l.base.Fatal(args...)
+}
+
+func (l *Logger) Debugf(format string, args ...interface{}) {
+	l.Debug(fmt.Sprintf(format, args...))
+}
+
+func (l *Logger) Infof(format string, args ...interface{}) {
+	l.Info(fmt.Sprintf(format, args...))
+}
+
+func (l *Logger) Warnf(format string, args ...interface{}) {
+	l.Warn(fmt.Sprintf(format, args...))
+}
+
+func (l *Logger) Errorf(format string, args ...interface{}) {
+	l.Error(fmt.Sprintf(format, args...))
+}
+
+func (l *Logger) Fatalf(format string, args ...interface{}) {
+	l.Fatal(fmt.Sprintf(format, args...))
+}
+
+// SetLevel sets the logger level.
+// It panics if v is less than DebugLevel or greater than FatalLevel.
+func (l *Logger) SetLevel(v Level) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if v < DebugLevel || v > FatalLevel {
+		panic("log: invalid log level")
+	}
+	l.level = v
+}
--- a/internal/log/log_test.go
+++ b/internal/log/log_test.go
@@ -13,6 +13,7 @@ import (

 // regexp for timestamps
 const (
+	rgxPID          = `[0-9]+`
 	rgxdate         = `[0-9][0-9][0-9][0-9]/[0-9][0-9]/[0-9][0-9]`
 	rgxtime         = `[0-9][0-9]:[0-9][0-9]:[0-9][0-9]`
 	rgxmicroseconds = `\.[0-9][0-9][0-9][0-9][0-9][0-9]`
@@ -27,20 +28,22 @@ type tester struct {
 func TestLoggerDebug(t *testing.T) {
 	tests := []tester{
 		{
-			desc:        "without trailing newline, logger adds newline",
-			message:     "hello, world!",
-			wantPattern: fmt.Sprintf("^%s %s%s DEBUG: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			desc:    "without trailing newline, logger adds newline",
+			message: "hello, world!",
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s DEBUG: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 		{
-			desc:        "with trailing newline, logger preserves newline",
-			message:     "hello, world!\n",
-			wantPattern: fmt.Sprintf("^%s %s%s DEBUG: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			desc:    "with trailing newline, logger preserves newline",
+			message: "hello, world!\n",
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s DEBUG: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 	}

 	for _, tc := range tests {
 		var buf bytes.Buffer
-		logger := NewLogger(&buf)
+		logger := NewLogger(newBase(&buf))

 		logger.Debug(tc.message)

@@ -50,7 +53,7 @@ func TestLoggerDebug(t *testing.T) {
 			t.Fatal("pattern did not compile:", err)
 		}
 		if !matched {
-			t.Errorf("logger.info(%q) outputted %q, should match pattern %q",
+			t.Errorf("logger.Debug(%q) outputted %q, should match pattern %q",
 				tc.message, got, tc.wantPattern)
 		}
 	}
@@ -59,20 +62,22 @@ func TestLoggerDebug(t *testing.T) {
 func TestLoggerInfo(t *testing.T) {
 	tests := []tester{
 		{
-			desc:        "without trailing newline, logger adds newline",
-			message:     "hello, world!",
-			wantPattern: fmt.Sprintf("^%s %s%s INFO: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			desc:    "without trailing newline, logger adds newline",
+			message: "hello, world!",
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s INFO: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 		{
-			desc:        "with trailing newline, logger preserves newline",
-			message:     "hello, world!\n",
-			wantPattern: fmt.Sprintf("^%s %s%s INFO: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			desc:    "with trailing newline, logger preserves newline",
+			message: "hello, world!\n",
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s INFO: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 	}

 	for _, tc := range tests {
 		var buf bytes.Buffer
-		logger := NewLogger(&buf)
+		logger := NewLogger(newBase(&buf))

 		logger.Info(tc.message)

@@ -82,7 +87,7 @@ func TestLoggerInfo(t *testing.T) {
 			t.Fatal("pattern did not compile:", err)
 		}
 		if !matched {
-			t.Errorf("logger.info(%q) outputted %q, should match pattern %q",
+			t.Errorf("logger.Info(%q) outputted %q, should match pattern %q",
 				tc.message, got, tc.wantPattern)
 		}
 	}
@@ -91,20 +96,22 @@ func TestLoggerInfo(t *testing.T) {
 func TestLoggerWarn(t *testing.T) {
 	tests := []tester{
 		{
-			desc:        "without trailing newline, logger adds newline",
-			message:     "hello, world!",
-			wantPattern: fmt.Sprintf("^%s %s%s WARN: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			desc:    "without trailing newline, logger adds newline",
+			message: "hello, world!",
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s WARN: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 		{
-			desc:        "with trailing newline, logger preserves newline",
-			message:     "hello, world!\n",
-			wantPattern: fmt.Sprintf("^%s %s%s WARN: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			desc:    "with trailing newline, logger preserves newline",
+			message: "hello, world!\n",
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s WARN: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 	}

 	for _, tc := range tests {
 		var buf bytes.Buffer
-		logger := NewLogger(&buf)
+		logger := NewLogger(newBase(&buf))

 		logger.Warn(tc.message)

@@ -114,7 +121,7 @@ func TestLoggerWarn(t *testing.T) {
 			t.Fatal("pattern did not compile:", err)
 		}
 		if !matched {
-			t.Errorf("logger.info(%q) outputted %q, should match pattern %q",
+			t.Errorf("logger.Warn(%q) outputted %q, should match pattern %q",
 				tc.message, got, tc.wantPattern)
 		}
 	}
@@ -123,20 +130,22 @@ func TestLoggerWarn(t *testing.T) {
 func TestLoggerError(t *testing.T) {
 	tests := []tester{
 		{
-			desc:        "without trailing newline, logger adds newline",
-			message:     "hello, world!",
-			wantPattern: fmt.Sprintf("^%s %s%s ERROR: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			desc:    "without trailing newline, logger adds newline",
+			message: "hello, world!",
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s ERROR: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 		{
-			desc:        "with trailing newline, logger preserves newline",
-			message:     "hello, world!\n",
-			wantPattern: fmt.Sprintf("^%s %s%s ERROR: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			desc:    "with trailing newline, logger preserves newline",
+			message: "hello, world!\n",
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s ERROR: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 	}

 	for _, tc := range tests {
 		var buf bytes.Buffer
-		logger := NewLogger(&buf)
+		logger := NewLogger(newBase(&buf))

 		logger.Error(tc.message)

@@ -146,8 +155,234 @@ func TestLoggerError(t *testing.T) {
 			t.Fatal("pattern did not compile:", err)
 		}
 		if !matched {
-			t.Errorf("logger.info(%q) outputted %q, should match pattern %q",
+			t.Errorf("logger.Error(%q) outputted %q, should match pattern %q",
 				tc.message, got, tc.wantPattern)
 		}
 	}
 }
+
+type formatTester struct {
+	desc        string
+	format      string
+	args        []interface{}
+	wantPattern string // regexp that log output must match
+}
+
+func TestLoggerDebugf(t *testing.T) {
+	tests := []formatTester{
+		{
+			desc:   "Formats message with DEBUG prefix",
+			format: "hello, %s!",
+			args:   []interface{}{"Gopher"},
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s DEBUG: hello, Gopher!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
+		},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+
+		logger.Debugf(tc.format, tc.args...)
+
+		got := buf.String()
+		matched, err := regexp.MatchString(tc.wantPattern, got)
+		if err != nil {
+			t.Fatal("pattern did not compile:", err)
+		}
+		if !matched {
+			t.Errorf("logger.Debugf(%q, %v) outputted %q, should match pattern %q",
+				tc.format, tc.args, got, tc.wantPattern)
+		}
+	}
+}
+
+func TestLoggerInfof(t *testing.T) {
+	tests := []formatTester{
+		{
+			desc:   "Formats message with INFO prefix",
+			format: "%d,%d,%d",
+			args:   []interface{}{1, 2, 3},
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s INFO: 1,2,3\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
+		},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+
+		logger.Infof(tc.format, tc.args...)
+
+		got := buf.String()
+		matched, err := regexp.MatchString(tc.wantPattern, got)
+		if err != nil {
+			t.Fatal("pattern did not compile:", err)
+		}
+		if !matched {
+			t.Errorf("logger.Infof(%q, %v) outputted %q, should match pattern %q",
+				tc.format, tc.args, got, tc.wantPattern)
+		}
+	}
+}
+
+func TestLoggerWarnf(t *testing.T) {
+	tests := []formatTester{
+		{
+			desc:   "Formats message with WARN prefix",
+			format: "hello, %s",
+			args:   []interface{}{"Gophers"},
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s WARN: hello, Gophers\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
+		},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+
+		logger.Warnf(tc.format, tc.args...)
+
+		got := buf.String()
+		matched, err := regexp.MatchString(tc.wantPattern, got)
+		if err != nil {
+			t.Fatal("pattern did not compile:", err)
+		}
+		if !matched {
+			t.Errorf("logger.Warnf(%q, %v) outputted %q, should match pattern %q",
+				tc.format, tc.args, got, tc.wantPattern)
+		}
+	}
+}
+
+func TestLoggerErrorf(t *testing.T) {
+	tests := []formatTester{
+		{
+			desc:   "Formats message with ERROR prefix",
+			format: "hello, %s",
+			args:   []interface{}{"Gophers"},
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s ERROR: hello, Gophers\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
+		},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+
+		logger.Errorf(tc.format, tc.args...)
+
+		got := buf.String()
+		matched, err := regexp.MatchString(tc.wantPattern, got)
+		if err != nil {
+			t.Fatal("pattern did not compile:", err)
+		}
+		if !matched {
+			t.Errorf("logger.Errorf(%q, %v) outputted %q, should match pattern %q",
+				tc.format, tc.args, got, tc.wantPattern)
+		}
+	}
+}
+
+func TestLoggerWithLowerLevels(t *testing.T) {
+	// Logger should not log messages at a level
+	// lower than the specified level.
+	tests := []struct {
+		level Level
+		op    string
+	}{
+		// with level one above
+		{InfoLevel, "Debug"},
+		{InfoLevel, "Debugf"},
+		{WarnLevel, "Info"},
+		{WarnLevel, "Infof"},
+		{ErrorLevel, "Warn"},
+		{ErrorLevel, "Warnf"},
+		{FatalLevel, "Error"},
+		{FatalLevel, "Errorf"},
+		// with skip level
+		{WarnLevel, "Debug"},
+		{ErrorLevel, "Infof"},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+		logger.SetLevel(tc.level)
+
+		switch tc.op {
+		case "Debug":
+			logger.Debug("hello")
+		case "Debugf":
+			logger.Debugf("hello, %s", "world")
+		case "Info":
+			logger.Info("hello")
+		case "Infof":
+			logger.Infof("hello, %s", "world")
+		case "Warn":
+			logger.Warn("hello")
+		case "Warnf":
+			logger.Warnf("hello, %s", "world")
+		case "Error":
+			logger.Error("hello")
+		case "Errorf":
+			logger.Errorf("hello, %s", "world")
+		default:
+			t.Fatalf("unexpected op: %q", tc.op)
+		}
+
+		if buf.String() != "" {
+			t.Errorf("logger.%s outputted log message when level is set to %v", tc.op, tc.level)
+		}
+	}
+}
+
+func TestLoggerWithSameOrHigherLevels(t *testing.T) {
+	// Logger should log messages at a level
+	// same as or higher than the specified level.
+	tests := []struct {
+		level Level
+		op    string
+	}{
+		// same level
+		{DebugLevel, "Debug"},
+		{InfoLevel, "Infof"},
+		{WarnLevel, "Warn"},
+		{ErrorLevel, "Errorf"},
+		// higher level
+		{DebugLevel, "Info"},
+		{InfoLevel, "Warnf"},
+		{WarnLevel, "Error"},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+		logger.SetLevel(tc.level)
+
+		switch tc.op {
+		case "Debug":
+			logger.Debug("hello")
+		case "Debugf":
+			logger.Debugf("hello, %s", "world")
+		case "Info":
+			logger.Info("hello")
+		case "Infof":
+			logger.Infof("hello, %s", "world")
+		case "Warn":
+			logger.Warn("hello")
+		case "Warnf":
+			logger.Warnf("hello, %s", "world")
+		case "Error":
+			logger.Error("hello")
+		case "Errorf":
+			logger.Errorf("hello, %s", "world")
+		default:
+			t.Fatalf("unexpected op: %q", tc.op)
+		}
+
+		if buf.String() == "" {
+			t.Errorf("logger.%s did not output log message when level is set to %v", tc.op, tc.level)
+		}
+	}
+}
--- a/processor.go
+++ b/processor.go
@@ -13,12 +13,13 @@ import (
 	"time"

 	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/log"
 	"github.com/hibiken/asynq/internal/rdb"
 	"golang.org/x/time/rate"
 )

 type processor struct {
-	logger Logger
+	logger *log.Logger
 	broker base.Broker

 	ss *base.ServerState
@@ -63,8 +64,8 @@ type processor struct {

 type retryDelayFunc func(n int, err error, task *Task) time.Duration

-type newProcessorParams struct {
-	logger          Logger
+type processorParams struct {
+	logger          *log.Logger
 	broker          base.Broker
 	ss              *base.ServerState
 	retryDelayFunc  retryDelayFunc
@@ -75,7 +76,7 @@ type newProcessorParams struct {
 }

 // newProcessor constructs a new processor.
-func newProcessor(params newProcessorParams) *processor {
+func newProcessor(params processorParams) *processor {
 	info := params.ss.GetInfo()
 	qcfg := normalizeQueueCfg(info.Queues)
 	orderedQueues := []string(nil)
@@ -105,7 +106,7 @@ func newProcessor(params newProcessorParams) *processor {
 // It's safe to call this method multiple times.
 func (p *processor) stop() {
 	p.once.Do(func() {
-		p.logger.Info("Processor shutting down...")
+		p.logger.Debug("Processor shutting down...")
 		// Unblock if processor is waiting for sema token.
 		close(p.abort)
 		// Signal the processor goroutine to stop processing tasks
@@ -144,7 +145,7 @@ func (p *processor) start(wg *sync.WaitGroup) {
 		for {
 			select {
 			case <-p.done:
-				p.logger.Info("Processor done")
+				p.logger.Debug("Processor done")
 				return
 			default:
 				p.exec()
@@ -158,19 +159,20 @@ func (p *processor) start(wg *sync.WaitGroup) {
 func (p *processor) exec() {
 	qnames := p.queues()
 	msg, err := p.broker.Dequeue(qnames...)
-	if err == rdb.ErrNoProcessableTask { // TODO: Need to decouple this error from rdb to support other brokers
+	switch {
+	case err == rdb.ErrNoProcessableTask:
 		// queues are empty, this is a normal behavior.
-		if len(p.queueConfig) > 1 {
+		if len(qnames) > 1 {
 			// sleep to avoid slamming redis and let scheduler move tasks into queues.
 			// Note: With multiple queues, we are not using blocking pop operation and
 			// polling queues instead. This adds significant load to redis.
 			time.Sleep(time.Second)
 		}
+		p.logger.Debug("All queues are empty")
 		return
-	}
-	if err != nil {
+	case err != nil:
 		if p.errLogLimiter.Allow() {
-			p.logger.Error("Dequeue error: %v", err)
+			p.logger.Errorf("Dequeue error: %v", err)
 		}
 		return
 	}
@@ -185,22 +187,24 @@ func (p *processor) exec() {
 		go func() {
 			defer func() {
 				p.ss.DeleteWorkerStats(msg)
-				<-p.sema /* release token */
+				<-p.sema // release token
+			}()
+
+			ctx, cancel := createContext(msg)
+			p.cancelations.Add(msg.ID.String(), cancel)
+			defer func() {
+				cancel()
+				p.cancelations.Delete(msg.ID.String())
 			}()

 			resCh := make(chan error, 1)
 			task := NewTask(msg.Type, msg.Payload)
-			ctx, cancel := createContext(msg)
-			p.cancelations.Add(msg.ID.String(), cancel)
-			go func() {
-				resCh <- perform(ctx, task, p.handler)
-				p.cancelations.Delete(msg.ID.String())
-			}()
+			go func() { resCh <- perform(ctx, task, p.handler) }()

 			select {
 			case <-p.quit:
 				// time is up, quit this worker goroutine.
-				p.logger.Warn("Quitting worker. task id=%s", msg.ID)
+				p.logger.Warnf("Quitting worker. task id=%s", msg.ID)
 				return
 			case resErr := <-resCh:
 				// Note: One of three things should happen.
@@ -229,17 +233,17 @@ func (p *processor) exec() {
 func (p *processor) restore() {
 	n, err := p.broker.RequeueAll()
 	if err != nil {
-		p.logger.Error("Could not restore unfinished tasks: %v", err)
+		p.logger.Errorf("Could not restore unfinished tasks: %v", err)
 	}
 	if n > 0 {
-		p.logger.Info("Restored %d unfinished tasks back to queue", n)
+		p.logger.Infof("Restored %d unfinished tasks back to queue", n)
 	}
 }

 func (p *processor) requeue(msg *base.TaskMessage) {
 	err := p.broker.Requeue(msg)
 	if err != nil {
-		p.logger.Error("Could not push task id=%s back to queue: %v", msg.ID, err)
+		p.logger.Errorf("Could not push task id=%s back to queue: %v", msg.ID, err)
 	}
 }

@@ -247,7 +251,7 @@ func (p *processor) markAsDone(msg *base.TaskMessage) {
 	err := p.broker.Done(msg)
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not remove task id=%s from %q", msg.ID, base.InProgressQueue)
-		p.logger.Warn("%s; Will retry syncing", errMsg)
+		p.logger.Warnf("%s; Will retry syncing", errMsg)
 		p.syncRequestCh <- &syncRequest{
 			fn: func() error {
 				return p.broker.Done(msg)
@@ -263,7 +267,7 @@ func (p *processor) retry(msg *base.TaskMessage, e error) {
 	err := p.broker.Retry(msg, retryAt, e.Error())
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not move task id=%s from %q to %q", msg.ID, base.InProgressQueue, base.RetryQueue)
-		p.logger.Warn("%s; Will retry syncing", errMsg)
+		p.logger.Warnf("%s; Will retry syncing", errMsg)
 		p.syncRequestCh <- &syncRequest{
 			fn: func() error {
 				return p.broker.Retry(msg, retryAt, e.Error())
@@ -274,11 +278,11 @@ func (p *processor) retry(msg *base.TaskMessage, e error) {
 }

 func (p *processor) kill(msg *base.TaskMessage, e error) {
-	p.logger.Warn("Retry exhausted for task id=%s", msg.ID)
+	p.logger.Warnf("Retry exhausted for task id=%s", msg.ID)
 	err := p.broker.Kill(msg, e.Error())
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not move task id=%s from %q to %q", msg.ID, base.InProgressQueue, base.DeadQueue)
-		p.logger.Warn("%s; Will retry syncing", errMsg)
+		p.logger.Warnf("%s; Will retry syncing", errMsg)
 		p.syncRequestCh <- &syncRequest{
 			fn: func() error {
 				return p.broker.Kill(msg, e.Error())
@@ -306,7 +310,7 @@ func (p *processor) queues() []string {
 	}
 	var names []string
 	for qname, priority := range p.queueConfig {
-		for i := 0; i < int(priority); i++ {
+		for i := 0; i < priority; i++ {
 			names = append(names, qname)
 		}
 	}
@@ -401,20 +405,3 @@ func gcd(xs ...int) int {
 	}
 	return res
 }
-
-// createContext returns a context and cancel function for a given task message.
-func createContext(msg *base.TaskMessage) (ctx context.Context, cancel context.CancelFunc) {
-	ctx = context.Background()
-	timeout, err := time.ParseDuration(msg.Timeout)
-	if err == nil && timeout != 0 {
-		ctx, cancel = context.WithTimeout(ctx, timeout)
-	}
-	deadline, err := time.Parse(time.RFC3339, msg.Deadline)
-	if err == nil && !deadline.IsZero() {
-		ctx, cancel = context.WithDeadline(ctx, deadline)
-	}
-	if cancel == nil {
-		ctx, cancel = context.WithCancel(ctx)
-	}
-	return ctx, cancel
-}
--- a/processor_test.go
+++ b/processor_test.go
@@ -17,7 +17,6 @@ import (
 	h "github.com/hibiken/asynq/internal/asynqtest"
 	"github.com/hibiken/asynq/internal/base"
 	"github.com/hibiken/asynq/internal/rdb"
-	"github.com/rs/xid"
 )

 func TestProcessorSuccess(t *testing.T) {
@@ -37,19 +36,16 @@ func TestProcessorSuccess(t *testing.T) {
 	tests := []struct {
 		enqueued      []*base.TaskMessage // initial default queue state
 		incoming      []*base.TaskMessage // tasks to be enqueued during run
-		wait          time.Duration       // wait duration between starting and stopping processor for this test case
 		wantProcessed []*Task             // tasks to be processed at the end
 	}{
 		{
 			enqueued:      []*base.TaskMessage{m1},
 			incoming:      []*base.TaskMessage{m2, m3, m4},
-			wait:          time.Second,
 			wantProcessed: []*Task{t1, t2, t3, t4},
 		},
 		{
 			enqueued:      []*base.TaskMessage{},
 			incoming:      []*base.TaskMessage{m1},
-			wait:          time.Second,
 			wantProcessed: []*Task{t1},
 		},
 	}
@@ -68,21 +64,19 @@ func TestProcessorSuccess(t *testing.T) {
 			return nil
 		}
 		ss := base.NewServerState("localhost", 1234, 10, defaultQueueConfig, false)
-		cancelations := base.NewCancelations()
-		p := newProcessor(newProcessorParams{
+		p := newProcessor(processorParams{
 			logger:          testLogger,
 			broker:          rdbClient,
 			ss:              ss,
 			retryDelayFunc:  defaultDelayFunc,
 			syncCh:          nil,
-			cancelations:    cancelations,
+			cancelations:    base.NewCancelations(),
 			errHandler:      nil,
 			shutdownTimeout: defaultShutdownTimeout,
 		})
 		p.handler = HandlerFunc(handler)

-		var wg sync.WaitGroup
-		p.start(&wg)
+		p.start(&sync.WaitGroup{})
 		for _, msg := range tc.incoming {
 			err := rdbClient.Enqueue(msg)
 			if err != nil {
@@ -90,7 +84,7 @@ func TestProcessorSuccess(t *testing.T) {
 				t.Fatal(err)
 			}
 		}
-		time.Sleep(tc.wait)
+		time.Sleep(time.Second) // wait for one second to allow all enqueued tasks to be processed.
 		p.terminate()

 		if diff := cmp.Diff(tc.wantProcessed, processed, sortTaskOpt, cmp.AllowUnexported(Payload{})); diff != "" {
@@ -175,21 +169,19 @@ func TestProcessorRetry(t *testing.T) {
 			n++
 		}
 		ss := base.NewServerState("localhost", 1234, 10, defaultQueueConfig, false)
-		cancelations := base.NewCancelations()
-		p := newProcessor(newProcessorParams{
+		p := newProcessor(processorParams{
 			logger:          testLogger,
 			broker:          rdbClient,
 			ss:              ss,
 			retryDelayFunc:  delayFunc,
 			syncCh:          nil,
-			cancelations:    cancelations,
+			cancelations:    base.NewCancelations(),
 			errHandler:      ErrorHandlerFunc(errHandler),
 			shutdownTimeout: defaultShutdownTimeout,
 		})
 		p.handler = tc.handler

-		var wg sync.WaitGroup
-		p.start(&wg)
+		p.start(&sync.WaitGroup{})
 		for _, msg := range tc.incoming {
 			err := rdbClient.Enqueue(msg)
 			if err != nil {
@@ -200,7 +192,7 @@ func TestProcessorRetry(t *testing.T) {
 		time.Sleep(tc.wait)
 		p.terminate()

-		cmpOpt := cmpopts.EquateApprox(0, float64(time.Second)) // allow up to second difference in zset score
+		cmpOpt := cmpopts.EquateApprox(0, float64(time.Second)) // allow up to a second difference in zset score
 		gotRetry := h.GetRetryEntries(t, r)
 		if diff := cmp.Diff(tc.wantRetry, gotRetry, h.SortZSetEntryOpt, cmpOpt); diff != "" {
 			t.Errorf("mismatch found in %q after running processor; (-want, +got)\n%s", base.RetryQueue, diff)
@@ -249,15 +241,14 @@ func TestProcessorQueues(t *testing.T) {
 	}

 	for _, tc := range tests {
-		cancelations := base.NewCancelations()
 		ss := base.NewServerState("localhost", 1234, 10, tc.queueCfg, false)
-		p := newProcessor(newProcessorParams{
+		p := newProcessor(processorParams{
 			logger:          testLogger,
 			broker:          nil,
 			ss:              ss,
 			retryDelayFunc:  defaultDelayFunc,
 			syncCh:          nil,
-			cancelations:    cancelations,
+			cancelations:    base.NewCancelations(),
 			errHandler:      nil,
 			shutdownTimeout: defaultShutdownTimeout,
 		})
@@ -326,22 +317,20 @@ func TestProcessorWithStrictPriority(t *testing.T) {
 			"low":                 1,
 		}
 		// Note: Set concurrency to 1 to make sure tasks are processed one at a time.
-		cancelations := base.NewCancelations()
 		ss := base.NewServerState("localhost", 1234, 1 /* concurrency */, queueCfg, true /*strict*/)
-		p := newProcessor(newProcessorParams{
+		p := newProcessor(processorParams{
 			logger:          testLogger,
 			broker:          rdbClient,
 			ss:              ss,
 			retryDelayFunc:  defaultDelayFunc,
 			syncCh:          nil,
-			cancelations:    cancelations,
+			cancelations:    base.NewCancelations(),
 			errHandler:      nil,
 			shutdownTimeout: defaultShutdownTimeout,
 		})
 		p.handler = HandlerFunc(handler)

-		var wg sync.WaitGroup
-		p.start(&wg)
+		p.start(&sync.WaitGroup{})
 		time.Sleep(tc.wait)
 		p.terminate()

@@ -401,88 +390,6 @@ func TestPerform(t *testing.T) {
 	}
 }

-func TestCreateContextWithTimeRestrictions(t *testing.T) {
-	var (
-		noTimeout  = time.Duration(0)
-		noDeadline = time.Time{}
-	)
-
-	tests := []struct {
-		desc         string
-		timeout      time.Duration
-		deadline     time.Time
-		wantDeadline time.Time
-	}{
-		{"only with timeout", 10 * time.Second, noDeadline, time.Now().Add(10 * time.Second)},
-		{"only with deadline", noTimeout, time.Now().Add(time.Hour), time.Now().Add(time.Hour)},
-		{"with timeout and deadline (timeout < deadline)", 10 * time.Second, time.Now().Add(time.Hour), time.Now().Add(10 * time.Second)},
-		{"with timeout and deadline (timeout > deadline)", 10 * time.Minute, time.Now().Add(30 * time.Second), time.Now().Add(30 * time.Second)},
-	}
-
-	for _, tc := range tests {
-		msg := &base.TaskMessage{
-			Type:     "something",
-			ID:       xid.New(),
-			Timeout:  tc.timeout.String(),
-			Deadline: tc.deadline.Format(time.RFC3339),
-		}
-
-		ctx, cancel := createContext(msg)
-
-		select {
-		case x := <-ctx.Done():
-			t.Errorf("%s: <-ctx.Done() == %v, want nothing (it should block)", tc.desc, x)
-		default:
-		}
-
-		got, ok := ctx.Deadline()
-		if !ok {
-			t.Errorf("%s: ctx.Deadline() returned false, want deadline to be set", tc.desc)
-		}
-		if !cmp.Equal(tc.wantDeadline, got, cmpopts.EquateApproxTime(time.Second)) {
-			t.Errorf("%s: ctx.Deadline() returned %v, want %v", tc.desc, got, tc.wantDeadline)
-		}
-
-		cancel()
-
-		select {
-		case <-ctx.Done():
-		default:
-			t.Errorf("ctx.Done() blocked, want it to be non-blocking")
-		}
-	}
-}
-
-func TestCreateContextWithoutTimeRestrictions(t *testing.T) {
-	msg := &base.TaskMessage{
-		Type:     "something",
-		ID:       xid.New(),
-		Timeout:  time.Duration(0).String(),        // zero value to indicate no timeout
-		Deadline: time.Time{}.Format(time.RFC3339), // zero value to indicate no deadline
-	}
-
-	ctx, cancel := createContext(msg)
-
-	select {
-	case x := <-ctx.Done():
-		t.Errorf("<-ctx.Done() == %v, want nothing (it should block)", x)
-	default:
-	}
-
-	_, ok := ctx.Deadline()
-	if ok {
-		t.Error("ctx.Deadline() returned true, want deadline to not be set")
-	}
-
-	cancel()
-
-	select {
-	case <-ctx.Done():
-	default:
-		t.Error("ctx.Done() blocked, want it to be non-blocking")
-	}
-}
-
 func TestGCD(t *testing.T) {
 	tests := []struct {
 		input []int
--- a/scheduler.go
+++ b/scheduler.go
@@ -9,10 +9,11 @@ import (
 	"time"

 	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/log"
 )

 type scheduler struct {
-	logger Logger
+	logger *log.Logger
 	broker base.Broker

 	// channel to communicate back to the long running "scheduler" goroutine.
@@ -25,22 +26,29 @@ type scheduler struct {
 	qnames []string
 }

-func newScheduler(l Logger, b base.Broker, avgInterval time.Duration, qcfg map[string]int) *scheduler {
+type schedulerParams struct {
+	logger   *log.Logger
+	broker   base.Broker
+	interval time.Duration
+	queues   map[string]int
+}
+
+func newScheduler(params schedulerParams) *scheduler {
 	var qnames []string
-	for q := range qcfg {
+	for q := range params.queues {
 		qnames = append(qnames, q)
 	}
 	return &scheduler{
-		logger:      l,
-		broker:      b,
+		logger:      params.logger,
+		broker:      params.broker,
 		done:        make(chan struct{}),
-		avgInterval: avgInterval,
+		avgInterval: params.interval,
 		qnames:      qnames,
 	}
 }

 func (s *scheduler) terminate() {
-	s.logger.Info("Scheduler shutting down...")
+	s.logger.Debug("Scheduler shutting down...")
 	// Signal the scheduler goroutine to stop polling.
 	s.done <- struct{}{}
 }
@@ -53,7 +61,7 @@ func (s *scheduler) start(wg *sync.WaitGroup) {
 		for {
 			select {
 			case <-s.done:
-				s.logger.Info("Scheduler done")
+				s.logger.Debug("Scheduler done")
 				return
 			case <-time.After(s.avgInterval):
 				s.exec()
@@ -64,6 +72,6 @@ func (s *scheduler) start(wg *sync.WaitGroup) {

 func (s *scheduler) exec() {
 	if err := s.broker.CheckAndEnqueue(s.qnames...); err != nil {
-		s.logger.Error("Could not enqueue scheduled tasks: %v", err)
+		s.logger.Errorf("Could not enqueue scheduled tasks: %v", err)
 	}
 }
--- a/scheduler_test.go
+++ b/scheduler_test.go
@@ -19,7 +19,12 @@ func TestScheduler(t *testing.T) {
 	r := setup(t)
 	rdbClient := rdb.NewRDB(r)
 	const pollInterval = time.Second
-	s := newScheduler(testLogger, rdbClient, pollInterval, defaultQueueConfig)
+	s := newScheduler(schedulerParams{
+		logger:   testLogger,
+		broker:   rdbClient,
+		interval: pollInterval,
+		queues:   defaultQueueConfig,
+	})
 	t1 := h.NewTaskMessage("gen_thumbnail", nil)
 	t2 := h.NewTaskMessage("send_email", nil)
 	t3 := h.NewTaskMessage("reindex", nil)
--- a/servemux.go
+++ b/servemux.go
@@ -15,7 +15,7 @@ import (
 // ServeMux is a multiplexer for asynchronous tasks.
 // It matches the type of each task against a list of registered patterns
 // and calls the handler for the pattern that most closely matches the
-// taks's type name.
+// task's type name.
 //
 // Longer patterns take precedence over shorter ones, so that if there are
 // handlers registered for both "images" and "images:thumbnails",
--- a/server.go
+++ b/server.go
@@ -12,6 +12,7 @@ import (
 	"math/rand"
 	"os"
 	"runtime"
+	"strings"
 	"sync"
 	"time"

@@ -35,7 +36,7 @@ import (
 type Server struct {
 	ss *base.ServerState

-	logger Logger
+	logger *log.Logger

 	broker base.Broker

@@ -112,6 +113,11 @@ type Config struct {
 	// If unset, default logger is used.
 	Logger Logger

+	// LogLevel specifies the minimum log level to enable.
+	//
+	// If unset, InfoLevel is used by default.
+	LogLevel LogLevel
+
 	// ShutdownTimeout specifies the duration to wait to let workers finish their tasks
 	// before forcing them to abort when stopping the server.
 	//
@@ -133,23 +139,104 @@ func (fn ErrorHandlerFunc) HandleError(task *Task, err error, retried, maxRetry
 	fn(task, err, retried, maxRetry)
 }

-// Logger implements logging with various log levels.
+// Logger supports logging at various log levels.
 type Logger interface {
 	// Debug logs a message at Debug level.
-	Debug(format string, args ...interface{})
+	Debug(args ...interface{})

 	// Info logs a message at Info level.
-	Info(format string, args ...interface{})
+	Info(args ...interface{})

 	// Warn logs a message at Warning level.
-	Warn(format string, args ...interface{})
+	Warn(args ...interface{})

 	// Error logs a message at Error level.
-	Error(format string, args ...interface{})
+	Error(args ...interface{})

 	// Fatal logs a message at Fatal level
 	// and process will exit with status set to 1.
-	Fatal(format string, args ...interface{})
+	Fatal(args ...interface{})
+}
+
+// LogLevel represents logging level.
+//
+// It satisfies flag.Value interface.
+type LogLevel int32
+
+const (
+	// Note: reserving value zero to differentiate unspecified case.
+	level_unspecified LogLevel = iota
+
+	// DebugLevel is the lowest level of logging.
+	// Debug logs are intended for debugging and development purposes.
+	DebugLevel
+
+	// InfoLevel is used for general informational log messages.
+	InfoLevel
+
+	// WarnLevel is used for undesired but relatively expected events,
+	// which may indicate a problem.
+	WarnLevel
+
+	// ErrorLevel is used for undesired and unexpected events that
+	// the program can recover from.
+	ErrorLevel
+
+	// FatalLevel is used for undesired and unexpected events that
+	// the program cannot recover from.
+	FatalLevel
+)
+
+// String is part of the flag.Value interface.
+func (l *LogLevel) String() string {
+	switch *l {
+	case DebugLevel:
+		return "debug"
+	case InfoLevel:
+		return "info"
+	case WarnLevel:
+		return "warn"
+	case ErrorLevel:
+		return "error"
+	case FatalLevel:
+		return "fatal"
+	}
+	panic(fmt.Sprintf("asynq: unexpected log level: %v", *l))
+}
+
+// Set is part of the flag.Value interface.
+func (l *LogLevel) Set(val string) error {
+	switch strings.ToLower(val) {
+	case "debug":
+		*l = DebugLevel
+	case "info":
+		*l = InfoLevel
+	case "warn", "warning":
+		*l = WarnLevel
+	case "error":
+		*l = ErrorLevel
+	case "fatal":
+		*l = FatalLevel
+	default:
+		return fmt.Errorf("asynq: unsupported log level %q", val)
+	}
+	return nil
+}
+
+func toInternalLogLevel(l LogLevel) log.Level {
+	switch l {
+	case DebugLevel:
+		return log.DebugLevel
+	case InfoLevel:
+		return log.InfoLevel
+	case WarnLevel:
+		return log.WarnLevel
+	case ErrorLevel:
+		return log.ErrorLevel
+	case FatalLevel:
+		return log.FatalLevel
+	}
+	panic(fmt.Sprintf("asynq: unexpected log level: %v", l))
 }

 // Formula taken from https://github.com/mperham/sidekiq.
@@ -185,14 +272,16 @@ func NewServer(r RedisConnOpt, cfg Config) *Server {
 	if len(queues) == 0 {
 		queues = defaultQueueConfig
 	}
-	logger := cfg.Logger
-	if logger == nil {
-		logger = log.NewLogger(os.Stderr)
-	}
 	shutdownTimeout := cfg.ShutdownTimeout
 	if shutdownTimeout == 0 {
 		shutdownTimeout = defaultShutdownTimeout
 	}
+	logger := log.NewLogger(cfg.Logger)
+	loglevel := cfg.LogLevel
+	if loglevel == level_unspecified {
+		loglevel = InfoLevel
+	}
+	logger.SetLevel(toInternalLogLevel(loglevel))

 	host, err := os.Hostname()
 	if err != nil {
@@ -204,11 +293,30 @@ func NewServer(r RedisConnOpt, cfg Config) *Server {
 	ss := base.NewServerState(host, pid, n, queues, cfg.StrictPriority)
 	syncCh := make(chan *syncRequest)
 	cancels := base.NewCancelations()
-	syncer := newSyncer(logger, syncCh, 5*time.Second)
-	heartbeater := newHeartbeater(logger, rdb, ss, 5*time.Second)
-	scheduler := newScheduler(logger, rdb, 5*time.Second, queues)
-	subscriber := newSubscriber(logger, rdb, cancels)
-	processor := newProcessor(newProcessorParams{
+
+	syncer := newSyncer(syncerParams{
+		logger:     logger,
+		requestsCh: syncCh,
+		interval:   5 * time.Second,
+	})
+	heartbeater := newHeartbeater(heartbeaterParams{
+		logger:      logger,
+		broker:      rdb,
+		serverState: ss,
+		interval:    5 * time.Second,
+	})
+	scheduler := newScheduler(schedulerParams{
+		logger:   logger,
+		broker:   rdb,
+		interval: 5 * time.Second,
+		queues:   queues,
+	})
+	subscriber := newSubscriber(subscriberParams{
+		logger:       logger,
+		broker:       rdb,
+		cancelations: cancels,
+	})
+	processor := newProcessor(processorParams{
 		logger:          logger,
 		broker:          rdb,
 		ss:              ss,
@@ -291,13 +399,6 @@ func (srv *Server) Start(handler Handler) error {
 	srv.ss.SetStatus(base.StatusRunning)
 	srv.processor.handler = handler

-	type prefixLogger interface {
-		SetPrefix(prefix string)
-	}
-	// If logger supports setting prefix, then set prefix for log output.
-	if l, ok := srv.logger.(prefixLogger); ok {
-		l.SetPrefix(fmt.Sprintf("asynq: pid=%d ", os.Getpid()))
-	}
 	srv.logger.Info("Starting processing")

 	srv.heartbeater.start(&srv.wg)
@@ -319,7 +420,6 @@ func (srv *Server) Stop() {
 		return
 	}

-	fmt.Println() // print newline for prettier log.
 	srv.logger.Info("Starting graceful shutdown")
 	// Note: The order of termination is important.
 	// Sender goroutines should be terminated before the receiver goroutines.
@@ -335,12 +435,14 @@ func (srv *Server) Stop() {
 	srv.broker.Close()
 	srv.ss.SetStatus(base.StatusStopped)

-	srv.logger.Info("Bye!")
+	srv.logger.Info("Exiting")
 }

 // Quiet signals the server to stop pulling new tasks off queues.
 // Quiet should be used before stopping the server.
 func (srv *Server) Quiet() {
+	srv.logger.Info("Stopping processor")
 	srv.processor.stop()
 	srv.ss.SetStatus(base.StatusQuiet)
+	srv.logger.Info("Processor stopped")
 }
--- a/server_test.go
+++ b/server_test.go
@@ -6,9 +6,13 @@ package asynq

 import (
 	"context"
+	"fmt"
+	"syscall"
 	"testing"
 	"time"

+	"github.com/hibiken/asynq/internal/rdb"
+	"github.com/hibiken/asynq/internal/testbroker"
 	"go.uber.org/goleak"
 )

@@ -24,6 +28,7 @@ func TestServer(t *testing.T) {
 	c := NewClient(r)
 	srv := NewServer(r, Config{
 		Concurrency: 10,
+		LogLevel:    testLogLevel,
 	})

 	// no-op handler
@@ -49,8 +54,37 @@ func TestServer(t *testing.T) {
 	srv.Stop()
 }

+func TestServerRun(t *testing.T) {
+	// https://github.com/go-redis/redis/issues/1029
+	ignoreOpt := goleak.IgnoreTopFunction("github.com/go-redis/redis/v7/internal/pool.(*ConnPool).reaper")
+	defer goleak.VerifyNoLeaks(t, ignoreOpt)
+
+	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
+
+	done := make(chan struct{})
+	// Make sure server exits when receiving TERM signal.
+	go func() {
+		time.Sleep(2 * time.Second)
+		syscall.Kill(syscall.Getpid(), syscall.SIGTERM)
+		done <- struct{}{}
+	}()
+
+	go func() {
+		select {
+		case <-time.After(10 * time.Second):
+			t.Fatal("server did not stop after receiving TERM signal")
+		case <-done:
+		}
+	}()
+
+	mux := NewServeMux()
+	if err := srv.Run(mux); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestServerErrServerStopped(t *testing.T) {
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{})
+	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
 	handler := NewServeMux()
 	if err := srv.Start(handler); err != nil {
 		t.Fatal(err)
@@ -63,7 +97,7 @@ func TestServerErrServerStopped(t *testing.T) {
 }

 func TestServerErrNilHandler(t *testing.T) {
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{})
+	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
 	err := srv.Start(nil)
 	if err == nil {
 		t.Error("Starting server with nil handler: (*Server).Start(nil) did not return error")
@@ -72,7 +106,7 @@ func TestServerErrNilHandler(t *testing.T) {
 }

 func TestServerErrServerRunning(t *testing.T) {
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{})
+	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
 	handler := NewServeMux()
 	if err := srv.Start(handler); err != nil {
 		t.Fatal(err)
@@ -83,3 +117,124 @@ func TestServerErrServerRunning(t *testing.T) {
 	}
 	srv.Stop()
 }
+
+func TestServerWithRedisDown(t *testing.T) {
+	// Make sure that server does not panic and exit if redis is down.
+	defer func() {
+		if r := recover(); r != nil {
+			t.Errorf("panic occurred: %v", r)
+		}
+	}()
+	r := rdb.NewRDB(setup(t))
+	testBroker := testbroker.NewTestBroker(r)
+	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
+	srv.broker = testBroker
+	srv.scheduler.broker = testBroker
+	srv.heartbeater.broker = testBroker
+	srv.processor.broker = testBroker
+	srv.subscriber.broker = testBroker
+	testBroker.Sleep()
+
+	// no-op handler
+	h := func(ctx context.Context, task *Task) error {
+		return nil
+	}
+
+	err := srv.Start(HandlerFunc(h))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	time.Sleep(3 * time.Second)
+
+	srv.Stop()
+}
+
+func TestServerWithFlakyBroker(t *testing.T) {
+	// Make sure that server does not panic and exit if redis is down.
+	defer func() {
+		if r := recover(); r != nil {
+			t.Errorf("panic occurred: %v", r)
+		}
+	}()
+	r := rdb.NewRDB(setup(t))
+	testBroker := testbroker.NewTestBroker(r)
+	srv := NewServer(RedisClientOpt{Addr: redisAddr, DB: redisDB}, Config{LogLevel: testLogLevel})
+	srv.broker = testBroker
+	srv.scheduler.broker = testBroker
+	srv.heartbeater.broker = testBroker
+	srv.processor.broker = testBroker
+	srv.subscriber.broker = testBroker
+
+	c := NewClient(RedisClientOpt{Addr: redisAddr, DB: redisDB})
+
+	h := func(ctx context.Context, task *Task) error {
+		// force task retry.
+		if task.Type == "bad_task" {
+			return fmt.Errorf("could not process %q", task.Type)
+		}
+		time.Sleep(2 * time.Second)
+		return nil
+	}
+
+	err := srv.Start(HandlerFunc(h))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for i := 0; i < 10; i++ {
+		err := c.Enqueue(NewTask("enqueued", nil), MaxRetry(i))
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = c.Enqueue(NewTask("bad_task", nil))
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = c.EnqueueIn(time.Duration(i)*time.Second, NewTask("scheduled", nil))
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// simulate redis going down.
+	testBroker.Sleep()
+
+	time.Sleep(3 * time.Second)
+
+	// simulate redis comes back online.
+	testBroker.Wakeup()
+
+	time.Sleep(3 * time.Second)
+
+	srv.Stop()
+}
+
+func TestLogLevel(t *testing.T) {
+	tests := []struct {
+		flagVal string
+		want    LogLevel
+		wantStr string
+	}{
+		{"debug", DebugLevel, "debug"},
+		{"Info", InfoLevel, "info"},
+		{"WARN", WarnLevel, "warn"},
+		{"warning", WarnLevel, "warn"},
+		{"Error", ErrorLevel, "error"},
+		{"fatal", FatalLevel, "fatal"},
+	}
+
+	for _, tc := range tests {
+		level := new(LogLevel)
+		if err := level.Set(tc.flagVal); err != nil {
+			t.Fatal(err)
+		}
+		if *level != tc.want {
+			t.Errorf("Set(%q): got %v, want %v", tc.flagVal, level, &tc.want)
+			continue
+		}
+		if got := level.String(); got != tc.wantStr {
+			t.Errorf("String() returned %q, want %q", got, tc.wantStr)
+		}
+	}
+}
--- a/subscriber.go
+++ b/subscriber.go
@@ -10,10 +10,11 @@ import (

 	"github.com/go-redis/redis/v7"
 	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/log"
 )

 type subscriber struct {
-	logger Logger
+	logger *log.Logger
 	broker base.Broker

 	// channel to communicate back to the long running "subscriber" goroutine.
@@ -26,18 +27,24 @@ type subscriber struct {
 	retryTimeout time.Duration
 }

-func newSubscriber(l Logger, b base.Broker, cancelations *base.Cancelations) *subscriber {
+type subscriberParams struct {
+	logger       *log.Logger
+	broker       base.Broker
+	cancelations *base.Cancelations
+}
+
+func newSubscriber(params subscriberParams) *subscriber {
 	return &subscriber{
-		logger:       l,
-		broker:       b,
+		logger:       params.logger,
+		broker:       params.broker,
 		done:         make(chan struct{}),
-		cancelations: cancelations,
+		cancelations: params.cancelations,
 		retryTimeout: 5 * time.Second,
 	}
 }

 func (s *subscriber) terminate() {
-	s.logger.Info("Subscriber shutting down...")
+	s.logger.Debug("Subscriber shutting down...")
 	// Signal the subscriber goroutine to stop.
 	s.done <- struct{}{}
 }
@@ -54,12 +61,12 @@ func (s *subscriber) start(wg *sync.WaitGroup) {
 		for {
 			pubsub, err = s.broker.CancelationPubSub()
 			if err != nil {
-				s.logger.Error("cannot subscribe to cancelation channel: %v", err)
+				s.logger.Errorf("cannot subscribe to cancelation channel: %v", err)
 				select {
 				case <-time.After(s.retryTimeout):
 					continue
 				case <-s.done:
-					s.logger.Info("Subscriber done")
+					s.logger.Debug("Subscriber done")
 					return
 				}
 			}
@@ -70,7 +77,7 @@ func (s *subscriber) start(wg *sync.WaitGroup) {
 			select {
 			case <-s.done:
 				pubsub.Close()
-				s.logger.Info("Subscriber done")
+				s.logger.Debug("Subscriber done")
 				return
 			case msg := <-cancelCh:
 				cancel, ok := s.cancelations.Get(msg.Payload)
--- a/subscriber_test.go
+++ b/subscriber_test.go
@@ -38,7 +38,11 @@ func TestSubscriber(t *testing.T) {
 		cancelations := base.NewCancelations()
 		cancelations.Add(tc.registeredID, fakeCancelFunc)

-		subscriber := newSubscriber(testLogger, rdbClient, cancelations)
+		subscriber := newSubscriber(subscriberParams{
+			logger:       testLogger,
+			broker:       rdbClient,
+			cancelations: cancelations,
+		})
 		var wg sync.WaitGroup
 		subscriber.start(&wg)
 		defer subscriber.terminate()
@@ -75,7 +79,11 @@ func TestSubscriberWithRedisDown(t *testing.T) {
 	testBroker := testbroker.NewTestBroker(r)

 	cancelations := base.NewCancelations()
-	subscriber := newSubscriber(testLogger, testBroker, cancelations)
+	subscriber := newSubscriber(subscriberParams{
+		logger:       testLogger,
+		broker:       testBroker,
+		cancelations: cancelations,
+	})
 	subscriber.retryTimeout = 1 * time.Second // set shorter retry timeout for testing purpose.

 	testBroker.Sleep() // simulate a situation where subscriber cannot connect to redis.
--- a/syncer.go
+++ b/syncer.go
@@ -7,12 +7,14 @@ package asynq
 import (
 	"sync"
 	"time"
+
+	"github.com/hibiken/asynq/internal/log"
 )

 // syncer is responsible for queuing up failed requests to redis and retry
 // those requests to sync state between the background process and redis.
 type syncer struct {
-	logger Logger
+	logger *log.Logger

 	requestsCh <-chan *syncRequest

@@ -28,17 +30,23 @@ type syncRequest struct {
 	errMsg string       // error message
 }

-func newSyncer(l Logger, requestsCh <-chan *syncRequest, interval time.Duration) *syncer {
+type syncerParams struct {
+	logger     *log.Logger
+	requestsCh <-chan *syncRequest
+	interval   time.Duration
+}
+
+func newSyncer(params syncerParams) *syncer {
 	return &syncer{
-		logger:     l,
-		requestsCh: requestsCh,
+		logger:     params.logger,
+		requestsCh: params.requestsCh,
 		done:       make(chan struct{}),
-		interval:   interval,
+		interval:   params.interval,
 	}
 }

 func (s *syncer) terminate() {
-	s.logger.Info("Syncer shutting down...")
+	s.logger.Debug("Syncer shutting down...")
 	// Signal the syncer goroutine to stop.
 	s.done <- struct{}{}
 }
@@ -57,7 +65,7 @@ func (s *syncer) start(wg *sync.WaitGroup) {
 						s.logger.Error(req.errMsg)
 					}
 				}
-				s.logger.Info("Syncer done")
+				s.logger.Debug("Syncer done")
 				return
 			case req := <-s.requestsCh:
 				requests = append(requests, req)
--- a/syncer_test.go
+++ b/syncer_test.go
@@ -27,7 +27,11 @@ func TestSyncer(t *testing.T) {

 	const interval = time.Second
 	syncRequestCh := make(chan *syncRequest)
-	syncer := newSyncer(testLogger, syncRequestCh, interval)
+	syncer := newSyncer(syncerParams{
+		logger:     testLogger,
+		requestsCh: syncRequestCh,
+		interval:   interval,
+	})
 	var wg sync.WaitGroup
 	syncer.start(&wg)
 	defer syncer.terminate()
@@ -52,7 +56,11 @@ func TestSyncer(t *testing.T) {
 func TestSyncerRetry(t *testing.T) {
 	const interval = time.Second
 	syncRequestCh := make(chan *syncRequest)
-	syncer := newSyncer(testLogger, syncRequestCh, interval)
+	syncer := newSyncer(syncerParams{
+		logger:     testLogger,
+		requestsCh: syncRequestCh,
+		interval:   interval,
+	})

 	var wg sync.WaitGroup
 	syncer.start(&wg)
Author	SHA1	Message	Date
Ken Hibino	69ad583278	v0.9.1	2020-05-29 05:42:40 -07:00
Ken Hibino	23f46dde52	Add helper functions to extract task metadata from context	2020-05-29 05:40:42 -07:00
lihe	39188fe930	remove typo and redundant code	2020-05-22 05:11:54 -07:00
Ken Hibino	4492ed9255	Change internal constructor signatures. Created "params" type to avoid positional arguments. Personally it feels more explicit and reads better.	2020-05-17 13:25:24 -07:00
Ken Hibino	4e3e053989	Update readme	2020-05-16 11:00:44 -07:00
Ken Hibino	aef0775c05	v0.9.0	2020-05-16 08:02:57 -07:00
Ken Hibino	de146993d2	Add log messages around Server.Quiet	2020-05-16 08:01:39 -07:00
Ken Hibino	60cbf8dc5a	Minor code cleanup	2020-05-16 08:00:35 -07:00
Ken Hibino	fb38086590	Clean up log messages Moved development purpose log messages to DEBUG level.	2020-05-16 08:00:35 -07:00
Ken Hibino	cfcd19a222	Change default log level to info	2020-05-16 08:00:35 -07:00
Ken Hibino	24ee4b9693	Define test flags for package testing Added test flags for - redis address (defaults to "localhost:6379") - redis db number (defaults to 14) - log level (defaults to FATAL)	2020-05-16 08:00:35 -07:00
Ken Hibino	7849b395bd	Update changelog	2020-05-16 08:00:35 -07:00
Ken Hibino	fa3082e5bb	Change LogLevel to satisfy flag.Value interface	2020-05-16 08:00:35 -07:00
Ken Hibino	d13f7e900f	Allow setting minimum log level for logger	2020-05-16 08:00:35 -07:00
Ken Hibino	b63476ddc8	Simplify Logger interface	2020-05-16 08:00:35 -07:00
Ken Hibino	210b026b01	Add log messages around Server.Quiet	2020-05-16 07:12:08 -07:00
Ken Hibino	556b2103fe	Minor code cleanup	2020-05-15 08:19:35 -07:00
Ken Hibino	0289bc7a10	Clean up log messages Moved development purpose log messages to DEBUG level.	2020-05-11 20:28:49 -07:00
Ken Hibino	ae942c93e5	Change default log level to info	2020-05-11 20:28:49 -07:00
Ken Hibino	0faf97f146	Define test flags for package testing Added test flags for - redis address (defaults to "localhost:6379") - redis db number (defaults to 14) - log level (defaults to FATAL)	2020-05-11 06:22:43 -07:00
Ken Hibino	711bfa371f	Update changelog	2020-05-11 06:22:43 -07:00
Ken Hibino	73d62844e6	Change LogLevel to satisfy flag.Value interface	2020-05-11 06:22:43 -07:00
Ken Hibino	00b82904c6	Allow setting minimum log level for logger	2020-05-11 06:22:43 -07:00
Ken Hibino	a866369866	Simplify Logger interface	2020-05-11 06:22:43 -07:00
Ken Hibino	26b78136ba	v0.8.3	2020-05-08 06:21:01 -07:00
t-asaka	44aad7f037	Add redis conn close func to client	2020-05-08 06:15:14 -07:00
Ken Hibino	9884d5f2fa	v0.8.2	2020-05-03 16:55:34 -07:00
Ken Hibino	826f1ecff4	Update docs	2020-05-03 16:54:39 -07:00
Ken Hibino	24f2b64c6c	Make sure to invoke CancelFunc in all cases	2020-05-03 15:58:23 -07:00
Ken Hibino	1c1474c55c	Add tests to simulate cases where server cannot talk to redis	2020-05-02 07:05:26 -07:00
Ken Hibino	5161b9368a	Clean up tests	2020-05-02 07:05:26 -07:00
Ken Hibino	0c998a8e17	Add test for signal handling	2020-04-28 06:56:05 -07:00
Ken Hibino	49160f2536	v0.8.1	2020-04-27 06:49:12 -07:00
Ken Hibino	e33d297d8e	Add SetDefaultOptions method to Client	2020-04-27 06:45:13 -07:00
Ken Hibino	eb8ced6bdd	Add ParseRedisURI helper function	2020-04-25 13:06:20 -07:00
Ken Hibino	789a9fd711	Update readme	2020-04-20 07:52:26 -07:00
Ken Hibino	5924cdac33	Add example tests	2020-04-19 11:36:43 -07:00