mirror of
https://github.com/hibiken/asynq.git
synced 2026-04-14 22:55:58 +08:00
Compare commits
10 Commits
preview-v0
...
dependabot
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5c52b7c3ad | ||
|
|
23905a286f | ||
|
|
5586efeae7 | ||
|
|
dd3c923f44 | ||
|
|
f81c78e68d | ||
|
|
2fd155e31d | ||
|
|
d704b68a42 | ||
|
|
ff887e1f89 | ||
|
|
5de9b1faf0 | ||
|
|
8261a03f0d |
2
go.mod
2
go.mod
@@ -11,7 +11,7 @@ require (
|
||||
go.uber.org/goleak v1.3.0
|
||||
golang.org/x/sys v0.37.0
|
||||
golang.org/x/time v0.14.0
|
||||
google.golang.org/protobuf v1.36.10
|
||||
google.golang.org/protobuf v1.36.11
|
||||
)
|
||||
|
||||
require (
|
||||
|
||||
4
go.sum
4
go.sum
@@ -36,7 +36,7 @@ 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.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=
|
||||
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
|
||||
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
|
||||
@@ -264,8 +264,10 @@ 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
|
||||
end
|
||||
local n = redis.call("LLEN", KEYS[i])
|
||||
local avg = sample_total / table.getn(ids)
|
||||
memusg = memusg + (avg * n)
|
||||
@@ -281,8 +283,10 @@ 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
|
||||
end
|
||||
local n = redis.call("ZCARD", KEYS[i])
|
||||
local avg = sample_total / table.getn(ids)
|
||||
memusg = memusg + (avg * n)
|
||||
@@ -304,13 +308,17 @@ 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
|
||||
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
|
||||
end
|
||||
return memusg
|
||||
`)
|
||||
|
||||
@@ -1488,6 +1488,7 @@ 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,6 +3274,29 @@ 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