v0.17.1

More details: https://redis.io/commands/KEYS

CHANGELOG.md

@@ -7,6 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.17.1] - 2021-04-04
+
+### Fixed
+
+- Fix bug in internal `RDB.memoryUsage` method.
+
+## [0.17.0] - 2021-03-24
+
+### Added
+
+- `DialTimeout`, `ReadTimeout`, and `WriteTimeout` options are added to `RedisConnOpt`.
+
+## [0.16.1] - 2021-03-20
+
+### Fixed
+
+- Replace `KEYS` command with `SCAN` as recommended by [redis doc](https://redis.io/commands/KEYS).
+
 ## [0.16.0] - 2021-03-10
 
 ### Added

asynq.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/go-redis/redis/v7"
 )
@@ -68,6 +69,26 @@ type RedisClientOpt struct {
 	// See: https://redis.io/commands/select.
 	DB int
 
+	// Dial timeout for establishing new connections.
+	// Default is 5 seconds.
+	DialTimeout time.Duration
+
+	// Timeout for socket reads.
+	// If timeout is reached, read commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is 3 seconds.
+	ReadTimeout time.Duration
+
+	// Timeout for socket writes.
+	// If timeout is reached, write commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is ReadTimout.
+	WriteTimeout time.Duration
+
 	// Maximum number of socket connections.
 	// Default is 10 connections per every CPU as reported by runtime.NumCPU.
 	PoolSize int
@@ -84,6 +105,9 @@ func (opt RedisClientOpt) MakeRedisClient() interface{} {
 		Username:     opt.Username,
 		Password:     opt.Password,
 		DB:           opt.DB,
+		DialTimeout:  opt.DialTimeout,
+		ReadTimeout:  opt.ReadTimeout,
+		WriteTimeout: opt.WriteTimeout,
 		PoolSize:     opt.PoolSize,
 		TLSConfig:    opt.TLSConfig,
 	})
@@ -116,6 +140,26 @@ type RedisFailoverClientOpt struct {
 	// See: https://redis.io/commands/select.
 	DB int
 
+	// Dial timeout for establishing new connections.
+	// Default is 5 seconds.
+	DialTimeout time.Duration
+
+	// Timeout for socket reads.
+	// If timeout is reached, read commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is 3 seconds.
+	ReadTimeout time.Duration
+
+	// Timeout for socket writes.
+	// If timeout is reached, write commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is ReadTimeout
+	WriteTimeout time.Duration
+
 	// Maximum number of socket connections.
 	// Default is 10 connections per every CPU as reported by runtime.NumCPU.
 	PoolSize int
@@ -133,6 +177,9 @@ func (opt RedisFailoverClientOpt) MakeRedisClient() interface{} {
 		Username:         opt.Username,
 		Password:         opt.Password,
 		DB:               opt.DB,
+		DialTimeout:      opt.DialTimeout,
+		ReadTimeout:      opt.ReadTimeout,
+		WriteTimeout:     opt.WriteTimeout,
 		PoolSize:         opt.PoolSize,
 		TLSConfig:        opt.TLSConfig,
 	})
@@ -157,6 +204,26 @@ type RedisClusterClientOpt struct {
 	// See: https://redis.io/commands/auth.
 	Password string
 
+	// Dial timeout for establishing new connections.
+	// Default is 5 seconds.
+	DialTimeout time.Duration
+
+	// Timeout for socket reads.
+	// If timeout is reached, read commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is 3 seconds.
+	ReadTimeout time.Duration
+
+	// Timeout for socket writes.
+	// If timeout is reached, write commands will fail with a timeout error
+	// instead of blocking.
+	//
+	// Use value -1 for no timeout and 0 for default.
+	// Default is ReadTimeout.
+	WriteTimeout time.Duration
+
 	// TLS Config used to connect to a server.
 	// TLS will be negotiated only if this field is set.
 	TLSConfig *tls.Config
@@ -168,6 +235,9 @@ func (opt RedisClusterClientOpt) MakeRedisClient() interface{} {
 		MaxRedirects: opt.MaxRedirects,
 		Username:     opt.Username,
 		Password:     opt.Password,
+		DialTimeout:  opt.DialTimeout,
+		ReadTimeout:  opt.ReadTimeout,
+		WriteTimeout: opt.WriteTimeout,
 		TLSConfig:    opt.TLSConfig,
 	})
 }

internal/base/base.go

@@ -19,7 +19,7 @@ import (
 )
 
 // Version of asynq library and CLI.
-const Version = "0.16.0"
+const Version = "0.17.1"
 
 // DefaultQueueName is the queue name used if none are specified by user.
 const DefaultQueueName = "default"

internal/rdb/inspect.go

@@ -172,10 +172,22 @@ func (r *RDB) CurrentStats(qname string) (*Stats, error) {
 }
 
 func (r *RDB) memoryUsage(qname string) (int64, error) {
-	keys, err := r.client.Keys(fmt.Sprintf("asynq:{%s}*", qname)).Result()
+	var (
+		keys   []string
+		data   []string
+		cursor uint64
+		err    error
+	)
+	for {
+		data, cursor, err = r.client.Scan(cursor, fmt.Sprintf("asynq:{%s}*", qname), 100).Result()
 		if err != nil {
 			return 0, err
 		}
+		keys = append(keys, data...)
+		if cursor == 0 {
+			break
+		}
+	}
 	var usg int64
 	for _, k := range keys {
 		n, err := r.client.MemoryUsage(k).Result()