mirror of
https://github.com/hibiken/asynq.git
synced 2026-04-14 22:55:58 +08:00
Compare commits
2 Commits
dependabot
...
pr-1100
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0a9dd34460 | ||
|
|
07898eade0 |
13
asynq.go
13
asynq.go
@@ -471,7 +471,7 @@ func (opt RedisClusterClientOpt) MakeRedisClient() interface{} {
|
||||
// redis://[:password@]host[:port][/dbnumber]
|
||||
// rediss://[:password@]host[:port][/dbnumber]
|
||||
// redis-socket://[:password@]path[?db=dbnumber]
|
||||
// redis-sentinel://[:password@]host1[:port][,host2:[:port]][,hostN:[:port]][?master=masterName]
|
||||
// redis-sentinel://[:password@]host1[:port][,host2:[:port]][,hostN:[:port]][/dbnumber][?master=masterName]
|
||||
func ParseRedisURI(uri string) (RedisConnOpt, error) {
|
||||
u, err := url.Parse(uri)
|
||||
if err != nil {
|
||||
@@ -545,11 +545,20 @@ func parseRedisSocketURI(u *url.URL) (RedisConnOpt, error) {
|
||||
func parseRedisSentinelURI(u *url.URL) (RedisConnOpt, error) {
|
||||
addrs := strings.Split(u.Host, ",")
|
||||
master := u.Query().Get("master")
|
||||
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 sentinel uri: database number should be the first segment of the path")
|
||||
}
|
||||
}
|
||||
var password string
|
||||
if v, ok := u.User.Password(); ok {
|
||||
password = v
|
||||
}
|
||||
return RedisFailoverClientOpt{MasterName: master, SentinelAddrs: addrs, SentinelPassword: password}, nil
|
||||
return RedisFailoverClientOpt{MasterName: master, SentinelAddrs: addrs, SentinelPassword: password, DB: db}, nil
|
||||
}
|
||||
|
||||
// ResultWriter is a client interface to write result data for a task.
|
||||
|
||||
@@ -146,6 +146,24 @@ func TestParseRedisURI(t *testing.T) {
|
||||
MasterName: "mymaster",
|
||||
SentinelAddrs: []string{"localhost:5000", "localhost:5001", "localhost:5002"},
|
||||
SentinelPassword: "mypassword",
|
||||
DB: 0,
|
||||
},
|
||||
},
|
||||
{
|
||||
"redis-sentinel://localhost:5000,localhost:5001,localhost:5002/3?master=mymaster",
|
||||
RedisFailoverClientOpt{
|
||||
MasterName: "mymaster",
|
||||
SentinelAddrs: []string{"localhost:5000", "localhost:5001", "localhost:5002"},
|
||||
DB: 3,
|
||||
},
|
||||
},
|
||||
{
|
||||
"redis-sentinel://:mypassword@localhost:5000,localhost:5001,localhost:5002/7?master=mymaster",
|
||||
RedisFailoverClientOpt{
|
||||
MasterName: "mymaster",
|
||||
SentinelAddrs: []string{"localhost:5000", "localhost:5001", "localhost:5002"},
|
||||
SentinelPassword: "mypassword",
|
||||
DB: 7,
|
||||
},
|
||||
},
|
||||
}
|
||||
@@ -188,6 +206,10 @@ func TestParseRedisURIErrors(t *testing.T) {
|
||||
"non integer for db numbers for socket",
|
||||
"redis-socket:///some/path/to/redis?db=one",
|
||||
},
|
||||
{
|
||||
"non integer for db number for sentinel",
|
||||
"redis-sentinel://localhost:5000/abc?master=mymaster",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
|
||||
4
go.mod
4
go.mod
@@ -1,6 +1,6 @@
|
||||
module github.com/hibiken/asynq
|
||||
|
||||
go 1.25.0
|
||||
go 1.24.0
|
||||
|
||||
require (
|
||||
github.com/google/go-cmp v0.7.0
|
||||
@@ -10,7 +10,7 @@ require (
|
||||
github.com/spf13/cast v1.10.0
|
||||
go.uber.org/goleak v1.3.0
|
||||
golang.org/x/sys v0.37.0
|
||||
golang.org/x/time v0.15.0
|
||||
golang.org/x/time v0.14.0
|
||||
google.golang.org/protobuf v1.36.10
|
||||
)
|
||||
|
||||
|
||||
4
go.sum
4
go.sum
@@ -34,8 +34,8 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
|
||||
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
|
||||
golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
|
||||
golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||
golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
|
||||
golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
|
||||
google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
|
||||
google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
|
||||
@@ -264,9 +264,7 @@ for i=1,2 do
|
||||
if (table.getn(ids) > 0) then
|
||||
for _, id in ipairs(ids) do
|
||||
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
|
||||
if bytes then
|
||||
sample_total = sample_total + bytes
|
||||
end
|
||||
sample_total = sample_total + bytes
|
||||
end
|
||||
local n = redis.call("LLEN", KEYS[i])
|
||||
local avg = sample_total / table.getn(ids)
|
||||
@@ -283,9 +281,7 @@ for i=3,6 do
|
||||
if (table.getn(ids) > 0) then
|
||||
for _, id in ipairs(ids) do
|
||||
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
|
||||
if bytes then
|
||||
sample_total = sample_total + bytes
|
||||
end
|
||||
sample_total = sample_total + bytes
|
||||
end
|
||||
local n = redis.call("ZCARD", KEYS[i])
|
||||
local avg = sample_total / table.getn(ids)
|
||||
@@ -308,17 +304,13 @@ if table.getn(groups) > 0 then
|
||||
local ids = redis.call("ZRANGE", group_key, 0, sample_size - 1)
|
||||
for _, id in ipairs(ids) do
|
||||
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
|
||||
if bytes then
|
||||
agg_task_sample_total = agg_task_sample_total + bytes
|
||||
agg_task_sample_size = agg_task_sample_size + 1
|
||||
end
|
||||
agg_task_sample_total = agg_task_sample_total + bytes
|
||||
agg_task_sample_size = agg_task_sample_size + 1
|
||||
end
|
||||
end
|
||||
end
|
||||
if agg_task_sample_size > 0 then
|
||||
local avg = agg_task_sample_total / agg_task_sample_size
|
||||
memusg = memusg + (avg * agg_task_count)
|
||||
end
|
||||
local avg = agg_task_sample_total / agg_task_sample_size
|
||||
memusg = memusg + (avg * agg_task_count)
|
||||
end
|
||||
return memusg
|
||||
`)
|
||||
|
||||
@@ -1488,7 +1488,6 @@ func (r *RDB) CancelationPubSub() (*redis.PubSub, error) {
|
||||
pubsub := r.client.Subscribe(ctx, base.CancelChannel)
|
||||
_, err := pubsub.Receive(ctx)
|
||||
if err != nil {
|
||||
pubsub.Close()
|
||||
return nil, errors.E(op, errors.Unknown, fmt.Sprintf("redis pubsub receive error: %v", err))
|
||||
}
|
||||
return pubsub, nil
|
||||
|
||||
@@ -3274,29 +3274,6 @@ func TestCancelationPubSub(t *testing.T) {
|
||||
mu.Unlock()
|
||||
}
|
||||
|
||||
func TestCancelationPubSubReceiveError(t *testing.T) {
|
||||
// Use a client connected to a non-existent Redis server to trigger
|
||||
// a Receive() error. This verifies that the pubsub connection is
|
||||
// closed on error, preventing connection leaks.
|
||||
client := redis.NewClient(&redis.Options{
|
||||
Addr: "localhost:0", // invalid port — connection will fail
|
||||
})
|
||||
r := NewRDB(client)
|
||||
defer r.Close()
|
||||
|
||||
pubsub, err := r.CancelationPubSub()
|
||||
if err == nil {
|
||||
// If no error, we must clean up the pubsub.
|
||||
if pubsub != nil {
|
||||
pubsub.Close()
|
||||
}
|
||||
t.Fatal("(*RDB).CancelationPubSub() expected to return an error when redis is unreachable")
|
||||
}
|
||||
if pubsub != nil {
|
||||
t.Error("(*RDB).CancelationPubSub() expected nil pubsub on error")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriteResult(t *testing.T) {
|
||||
r := setup(t)
|
||||
defer r.Close()
|
||||
|
||||
Reference in New Issue
Block a user