aboutsummaryrefslogtreecommitdiffstats
path: root/modules/queue
diff options
context:
space:
mode:
Diffstat (limited to 'modules/queue')
-rw-r--r--modules/queue/base_levelqueue_common.go2
-rw-r--r--modules/queue/base_redis.go2
-rw-r--r--modules/queue/base_test.go26
-rw-r--r--modules/queue/manager.go5
-rw-r--r--modules/queue/manager_test.go13
-rw-r--r--modules/queue/workerqueue_test.go57
6 files changed, 51 insertions, 54 deletions
diff --git a/modules/queue/base_levelqueue_common.go b/modules/queue/base_levelqueue_common.go
index 78d3b85a8a..d37093b84d 100644
--- a/modules/queue/base_levelqueue_common.go
+++ b/modules/queue/base_levelqueue_common.go
@@ -83,7 +83,7 @@ func prepareLevelDB(cfg *BaseConfig) (conn string, db *leveldb.DB, err error) {
}
conn = cfg.ConnStr
}
- for i := 0; i < 10; i++ {
+ for range 10 {
if db, err = nosql.GetManager().GetLevelDB(conn); err == nil {
break
}
diff --git a/modules/queue/base_redis.go b/modules/queue/base_redis.go
index a1e234943d..bea0fd7a98 100644
--- a/modules/queue/base_redis.go
+++ b/modules/queue/base_redis.go
@@ -29,7 +29,7 @@ func newBaseRedisGeneric(cfg *BaseConfig, unique bool) (baseQueue, error) {
client := nosql.GetManager().GetRedisClient(cfg.ConnStr)
var err error
- for i := 0; i < 10; i++ {
+ for range 10 {
err = client.Ping(graceful.GetManager().ShutdownContext()).Err()
if err == nil {
break
diff --git a/modules/queue/base_test.go b/modules/queue/base_test.go
index 01b52b3c16..8e7c18d740 100644
--- a/modules/queue/base_test.go
+++ b/modules/queue/base_test.go
@@ -17,11 +17,11 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
q, err := newFn(cfg)
assert.NoError(t, err)
- ctx := context.Background()
+ ctx := t.Context()
_ = q.RemoveAll(ctx)
cnt, err := q.Len(ctx)
assert.NoError(t, err)
- assert.EqualValues(t, 0, cnt)
+ assert.Equal(t, 0, cnt)
// push the first item
err = q.PushItem(ctx, []byte("foo"))
@@ -29,7 +29,7 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
cnt, err = q.Len(ctx)
assert.NoError(t, err)
- assert.EqualValues(t, 1, cnt)
+ assert.Equal(t, 1, cnt)
// push a duplicate item
err = q.PushItem(ctx, []byte("foo"))
@@ -45,10 +45,10 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
has, err := q.HasItem(ctx, []byte("foo"))
assert.NoError(t, err)
if !isUnique {
- assert.EqualValues(t, 2, cnt)
+ assert.Equal(t, 2, cnt)
assert.False(t, has) // non-unique queues don't check for duplicates
} else {
- assert.EqualValues(t, 1, cnt)
+ assert.Equal(t, 1, cnt)
assert.True(t, has)
}
@@ -59,18 +59,18 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
// pop the first item (and the duplicate if non-unique)
it, err := q.PopItem(ctx)
assert.NoError(t, err)
- assert.EqualValues(t, "foo", string(it))
+ assert.Equal(t, "foo", string(it))
if !isUnique {
it, err = q.PopItem(ctx)
assert.NoError(t, err)
- assert.EqualValues(t, "foo", string(it))
+ assert.Equal(t, "foo", string(it))
}
// pop another item
it, err = q.PopItem(ctx)
assert.NoError(t, err)
- assert.EqualValues(t, "bar", string(it))
+ assert.Equal(t, "bar", string(it))
// pop an empty queue (timeout, cancel)
ctxTimed, cancel := context.WithTimeout(ctx, 10*time.Millisecond)
@@ -87,7 +87,7 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
// test blocking push if queue is full
for i := 0; i < cfg.Length; i++ {
- err = q.PushItem(ctx, []byte(fmt.Sprintf("item-%d", i)))
+ err = q.PushItem(ctx, fmt.Appendf(nil, "item-%d", i))
assert.NoError(t, err)
}
ctxTimed, cancel = context.WithTimeout(ctx, 10*time.Millisecond)
@@ -107,13 +107,13 @@ func testQueueBasic(t *testing.T, newFn func(cfg *BaseConfig) (baseQueue, error)
// remove all
cnt, err = q.Len(ctx)
assert.NoError(t, err)
- assert.EqualValues(t, cfg.Length, cnt)
+ assert.Equal(t, cfg.Length, cnt)
_ = q.RemoveAll(ctx)
cnt, err = q.Len(ctx)
assert.NoError(t, err)
- assert.EqualValues(t, 0, cnt)
+ assert.Equal(t, 0, cnt)
})
}
@@ -121,12 +121,12 @@ func TestBaseDummy(t *testing.T) {
q, err := newBaseDummy(&BaseConfig{}, true)
assert.NoError(t, err)
- ctx := context.Background()
+ ctx := t.Context()
assert.NoError(t, q.PushItem(ctx, []byte("foo")))
cnt, err := q.Len(ctx)
assert.NoError(t, err)
- assert.EqualValues(t, 0, cnt)
+ assert.Equal(t, 0, cnt)
has, err := q.HasItem(ctx, []byte("foo"))
assert.NoError(t, err)
diff --git a/modules/queue/manager.go b/modules/queue/manager.go
index 079e2bee7a..ae6c51872d 100644
--- a/modules/queue/manager.go
+++ b/modules/queue/manager.go
@@ -6,6 +6,7 @@ package queue
import (
"context"
"errors"
+ "maps"
"sync"
"time"
@@ -70,9 +71,7 @@ func (m *Manager) ManagedQueues() map[int64]ManagedWorkerPoolQueue {
defer m.mu.Unlock()
queues := make(map[int64]ManagedWorkerPoolQueue, len(m.Queues))
- for k, v := range m.Queues {
- queues[k] = v
- }
+ maps.Copy(queues, m.Queues)
return queues
}
diff --git a/modules/queue/manager_test.go b/modules/queue/manager_test.go
index 15dd1b4f2f..fda498cc84 100644
--- a/modules/queue/manager_test.go
+++ b/modules/queue/manager_test.go
@@ -4,7 +4,6 @@
package queue
import (
- "context"
"path/filepath"
"testing"
@@ -48,7 +47,7 @@ CONN_STR = redis://
assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/common"), q.baseConfig.DataFullDir)
assert.Equal(t, 100000, q.baseConfig.Length)
assert.Equal(t, 20, q.batchLength)
- assert.Equal(t, "", q.baseConfig.ConnStr)
+ assert.Empty(t, q.baseConfig.ConnStr)
assert.Equal(t, "default_queue", q.baseConfig.QueueFullName)
assert.Equal(t, "default_queue_unique", q.baseConfig.SetFullName)
assert.NotZero(t, q.GetWorkerMaxNumber())
@@ -80,7 +79,7 @@ MAX_WORKERS = 123
assert.NoError(t, err)
- q1 := createWorkerPoolQueue[string](context.Background(), "no-such", cfgProvider, nil, false)
+ q1 := createWorkerPoolQueue[string](t.Context(), "no-such", cfgProvider, nil, false)
assert.Equal(t, "no-such", q1.GetName())
assert.Equal(t, "dummy", q1.GetType()) // no handler, so it becomes dummy
assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/dir1"), q1.baseConfig.DataFullDir)
@@ -96,13 +95,13 @@ MAX_WORKERS = 123
assert.Equal(t, "string", q1.GetItemTypeName())
qid1 := GetManager().qidCounter
- q2 := createWorkerPoolQueue(context.Background(), "sub", cfgProvider, func(s ...int) (unhandled []int) { return nil }, false)
+ q2 := createWorkerPoolQueue(t.Context(), "sub", cfgProvider, func(s ...int) (unhandled []int) { return nil }, false)
assert.Equal(t, "sub", q2.GetName())
assert.Equal(t, "level", q2.GetType())
assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/dir2"), q2.baseConfig.DataFullDir)
assert.Equal(t, 102, q2.baseConfig.Length)
assert.Equal(t, 22, q2.batchLength)
- assert.Equal(t, "", q2.baseConfig.ConnStr)
+ assert.Empty(t, q2.baseConfig.ConnStr)
assert.Equal(t, "sub_q2", q2.baseConfig.QueueFullName)
assert.Equal(t, "sub_q2_u2", q2.baseConfig.SetFullName)
assert.Equal(t, 123, q2.GetWorkerMaxNumber())
@@ -118,7 +117,7 @@ MAX_WORKERS = 123
assert.Equal(t, 120, q1.workerMaxNum)
stop := runWorkerPoolQueue(q2)
- assert.NoError(t, GetManager().GetManagedQueue(qid2).FlushWithContext(context.Background(), 0))
- assert.NoError(t, GetManager().FlushAll(context.Background(), 0))
+ assert.NoError(t, GetManager().GetManagedQueue(qid2).FlushWithContext(t.Context(), 0))
+ assert.NoError(t, GetManager().FlushAll(t.Context(), 0))
stop()
}
diff --git a/modules/queue/workerqueue_test.go b/modules/queue/workerqueue_test.go
index c0841a1752..a6c369d5f9 100644
--- a/modules/queue/workerqueue_test.go
+++ b/modules/queue/workerqueue_test.go
@@ -4,7 +4,6 @@
package queue
import (
- "context"
"slices"
"strconv"
"sync"
@@ -58,15 +57,15 @@ func TestWorkerPoolQueueUnhandled(t *testing.T) {
testRecorder.Record("push:%v", i)
assert.NoError(t, q.Push(i))
}
- assert.NoError(t, q.FlushWithContext(context.Background(), 0))
+ assert.NoError(t, q.FlushWithContext(t.Context(), 0))
stop()
ok := true
for i := 0; i < queueSetting.Length; i++ {
if i%2 == 0 {
- ok = ok && assert.EqualValues(t, 2, m[i], "test %s: item %d", t.Name(), i)
+ ok = ok && assert.Equal(t, 2, m[i], "test %s: item %d", t.Name(), i)
} else {
- ok = ok && assert.EqualValues(t, 1, m[i], "test %s: item %d", t.Name(), i)
+ ok = ok && assert.Equal(t, 1, m[i], "test %s: item %d", t.Name(), i)
}
}
if !ok {
@@ -78,17 +77,17 @@ func TestWorkerPoolQueueUnhandled(t *testing.T) {
runCount := 2 // we can run these tests even hundreds times to see its stability
t.Run("1/1", func(t *testing.T) {
- for i := 0; i < runCount; i++ {
+ for range runCount {
test(t, setting.QueueSettings{BatchLength: 1, MaxWorkers: 1})
}
})
t.Run("3/1", func(t *testing.T) {
- for i := 0; i < runCount; i++ {
+ for range runCount {
test(t, setting.QueueSettings{BatchLength: 3, MaxWorkers: 1})
}
})
t.Run("4/5", func(t *testing.T) {
- for i := 0; i < runCount; i++ {
+ for range runCount {
test(t, setting.QueueSettings{BatchLength: 4, MaxWorkers: 5})
}
})
@@ -97,17 +96,17 @@ func TestWorkerPoolQueueUnhandled(t *testing.T) {
func TestWorkerPoolQueuePersistence(t *testing.T) {
runCount := 2 // we can run these tests even hundreds times to see its stability
t.Run("1/1", func(t *testing.T) {
- for i := 0; i < runCount; i++ {
+ for range runCount {
testWorkerPoolQueuePersistence(t, setting.QueueSettings{BatchLength: 1, MaxWorkers: 1, Length: 100})
}
})
t.Run("3/1", func(t *testing.T) {
- for i := 0; i < runCount; i++ {
+ for range runCount {
testWorkerPoolQueuePersistence(t, setting.QueueSettings{BatchLength: 3, MaxWorkers: 1, Length: 100})
}
})
t.Run("4/5", func(t *testing.T) {
- for i := 0; i < runCount; i++ {
+ for range runCount {
testWorkerPoolQueuePersistence(t, setting.QueueSettings{BatchLength: 4, MaxWorkers: 5, Length: 100})
}
})
@@ -142,7 +141,7 @@ func testWorkerPoolQueuePersistence(t *testing.T, queueSetting setting.QueueSett
q, _ := newWorkerPoolQueueForTest("pr_patch_checker_test", queueSetting, testHandler, true)
stop := runWorkerPoolQueue(q)
- for i := 0; i < testCount; i++ {
+ for i := range testCount {
_ = q.Push("task-" + strconv.Itoa(i))
}
close(startWhenAllReady)
@@ -166,7 +165,7 @@ func testWorkerPoolQueuePersistence(t *testing.T, queueSetting setting.QueueSett
q, _ := newWorkerPoolQueueForTest("pr_patch_checker_test", queueSetting, testHandler, true)
stop := runWorkerPoolQueue(q)
- assert.NoError(t, q.FlushWithContext(context.Background(), 0))
+ assert.NoError(t, q.FlushWithContext(t.Context(), 0))
stop()
}
@@ -174,7 +173,7 @@ func testWorkerPoolQueuePersistence(t *testing.T, queueSetting setting.QueueSett
assert.NotEmpty(t, tasksQ1)
assert.NotEmpty(t, tasksQ2)
- assert.EqualValues(t, testCount, len(tasksQ1)+len(tasksQ2))
+ assert.Equal(t, testCount, len(tasksQ1)+len(tasksQ2))
}
func TestWorkerPoolQueueActiveWorkers(t *testing.T) {
@@ -187,34 +186,34 @@ func TestWorkerPoolQueueActiveWorkers(t *testing.T) {
q, _ := newWorkerPoolQueueForTest("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 1, Length: 100}, handler, false)
stop := runWorkerPoolQueue(q)
- for i := 0; i < 5; i++ {
+ for i := range 5 {
assert.NoError(t, q.Push(i))
}
time.Sleep(50 * time.Millisecond)
- assert.EqualValues(t, 1, q.GetWorkerNumber())
- assert.EqualValues(t, 1, q.GetWorkerActiveNumber())
+ assert.Equal(t, 1, q.GetWorkerNumber())
+ assert.Equal(t, 1, q.GetWorkerActiveNumber())
time.Sleep(500 * time.Millisecond)
- assert.EqualValues(t, 1, q.GetWorkerNumber())
- assert.EqualValues(t, 0, q.GetWorkerActiveNumber())
+ assert.Equal(t, 1, q.GetWorkerNumber())
+ assert.Equal(t, 0, q.GetWorkerActiveNumber())
time.Sleep(workerIdleDuration)
- assert.EqualValues(t, 1, q.GetWorkerNumber()) // there is at least one worker after the queue begins working
+ assert.Equal(t, 1, q.GetWorkerNumber()) // there is at least one worker after the queue begins working
stop()
q, _ = newWorkerPoolQueueForTest("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 3, Length: 100}, handler, false)
stop = runWorkerPoolQueue(q)
- for i := 0; i < 15; i++ {
+ for i := range 15 {
assert.NoError(t, q.Push(i))
}
time.Sleep(50 * time.Millisecond)
- assert.EqualValues(t, 3, q.GetWorkerNumber())
- assert.EqualValues(t, 3, q.GetWorkerActiveNumber())
+ assert.Equal(t, 3, q.GetWorkerNumber())
+ assert.Equal(t, 3, q.GetWorkerActiveNumber())
time.Sleep(500 * time.Millisecond)
- assert.EqualValues(t, 3, q.GetWorkerNumber())
- assert.EqualValues(t, 0, q.GetWorkerActiveNumber())
+ assert.Equal(t, 3, q.GetWorkerNumber())
+ assert.Equal(t, 0, q.GetWorkerActiveNumber())
time.Sleep(workerIdleDuration)
- assert.EqualValues(t, 1, q.GetWorkerNumber()) // there is at least one worker after the queue begins working
+ assert.Equal(t, 1, q.GetWorkerNumber()) // there is at least one worker after the queue begins working
stop()
}
@@ -241,13 +240,13 @@ func TestWorkerPoolQueueShutdown(t *testing.T) {
}
<-handlerCalled
time.Sleep(200 * time.Millisecond) // wait for a while to make sure all workers are active
- assert.EqualValues(t, 4, q.GetWorkerActiveNumber())
+ assert.Equal(t, 4, q.GetWorkerActiveNumber())
stop() // stop triggers shutdown
- assert.EqualValues(t, 0, q.GetWorkerActiveNumber())
+ assert.Equal(t, 0, q.GetWorkerActiveNumber())
// no item was ever handled, so we still get all of them again
q, _ = newWorkerPoolQueueForTest("test-workpoolqueue", qs, handler, false)
- assert.EqualValues(t, 20, q.GetQueueItemNumber())
+ assert.Equal(t, 20, q.GetQueueItemNumber())
}
func TestWorkerPoolQueueWorkerIdleReset(t *testing.T) {
@@ -275,7 +274,7 @@ func TestWorkerPoolQueueWorkerIdleReset(t *testing.T) {
}
q, _ = newWorkerPoolQueueForTest("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 2, Length: 100}, handler, false)
stop := runWorkerPoolQueue(q)
- for i := 0; i < 100; i++ {
+ for i := range 100 {
assert.NoError(t, q.Push(i))
}
time.Sleep(500 * time.Millisecond)
generated by cgit v1.2.3 (git 2.39.1) at 2025-07-06 09:45:51 +0000