sync comment & config & variable name

Author: wxiaoguang

modules/queue/workerqueue.go

@@ -178,11 +178,11 @@ func getNewQueueFn(t string) (string, func(cfg *BaseConfig, unique bool) (baseQu
 	case "redis":
 		return t, newBaseRedisGeneric
 	default: // level(leveldb,levelqueue,persistable-channel)
-		return "levelqueue", newBaseLevelQueueGeneric
+		return "level", newBaseLevelQueueGeneric
 	}
 }
 
-func NewWorkerPoolQueueByIniConfig[T any](name string, queueSetting setting.QueueSettings, handler HandlerFuncT[T], unique bool) *WorkerPoolQueue[T] {
+func NewWorkerPoolQueueBySetting[T any](name string, queueSetting setting.QueueSettings, handler HandlerFuncT[T], unique bool) *WorkerPoolQueue[T] {
 	if handler == nil {
 		log.Debug("Use dummy queue for %q because handler is nil and caller doesn't want to process the queue items", name)
 		queueSetting.Type = "dummy"
@@ -234,7 +234,7 @@ func createWorkerPoolQueue[T any](name string, cfgProvider setting.ConfigProvide
 		log.Error("Failed to get queue settings for %q: %v", name, err)
 		return nil
 	}
-	w := NewWorkerPoolQueueByIniConfig(name, queueSetting, handler, unique)
+	w := NewWorkerPoolQueueBySetting(name, queueSetting, handler, unique)
 	GetManager().AddManagedQueue(w)
 	return w
 }

modules/queue/workerqueue_test.go

@@ -39,9 +39,9 @@ func TestWorkerPoolQueueUnhandled(t *testing.T) {
 
 	mu := sync.Mutex{}
 
-	test := func(iniCfg setting.QueueSettings) {
-		iniCfg.Length = 100
-		iniCfg.Datadir = t.TempDir() + "/test-queue"
+	test := func(queueSetting setting.QueueSettings) {
+		queueSetting.Length = 100
+		queueSetting.Datadir = t.TempDir() + "/test-queue"
 		m := map[int]int{}
 		// odds are handled once, evens are handled twice
 		handler := func(items ...int) (unhandled []int) {
@@ -56,15 +56,15 @@ func TestWorkerPoolQueueUnhandled(t *testing.T) {
 			return unhandled
 		}
 
-		q := NewWorkerPoolQueueByIniConfig("test-workpoolqueue", iniCfg, handler, false)
+		q := NewWorkerPoolQueueBySetting("test-workpoolqueue", queueSetting, handler, false)
 		stop := runWorkerPoolQueue(q)
-		for i := 0; i < iniCfg.Length; i++ {
+		for i := 0; i < queueSetting.Length; i++ {
 			assert.NoError(t, q.Push(i))
 		}
 		assert.NoError(t, q.FlushWithContext(context.Background(), 0))
 		stop()
 
-		for i := 0; i < iniCfg.Length; i++ {
+		for i := 0; i < queueSetting.Length; i++ {
 			assert.EqualValues(t, i%2, m[i]%2)
 		}
 	}
@@ -106,9 +106,9 @@ func TestWorkerPoolQueuePersistence(t *testing.T) {
 	})
 }
 
-func testWorkerPoolQueuePersistence(t *testing.T, iniCfg setting.QueueSettings) {
-	testCount := iniCfg.Length
-	iniCfg.Datadir = t.TempDir() + "/test-queue"
+func testWorkerPoolQueuePersistence(t *testing.T, queueSetting setting.QueueSettings) {
+	testCount := queueSetting.Length
+	queueSetting.Datadir = t.TempDir() + "/test-queue"
 
 	mu := sync.Mutex{}
 
@@ -132,7 +132,7 @@ func testWorkerPoolQueuePersistence(t *testing.T, iniCfg setting.QueueSettings)
 			return nil
 		}
 
-		q := NewWorkerPoolQueueByIniConfig("pr_patch_checker_test", iniCfg, testHandler, true)
+		q := NewWorkerPoolQueueBySetting("pr_patch_checker_test", queueSetting, testHandler, true)
 		stop := runWorkerPoolQueue(q)
 		for i := 0; i < testCount; i++ {
 			_ = q.Push("task-" + strconv.Itoa(i))
@@ -156,7 +156,7 @@ func testWorkerPoolQueuePersistence(t *testing.T, iniCfg setting.QueueSettings)
 			return nil
 		}
 
-		q := NewWorkerPoolQueueByIniConfig("pr_patch_checker_test", iniCfg, testHandler, true)
+		q := NewWorkerPoolQueueBySetting("pr_patch_checker_test", queueSetting, testHandler, true)
 		stop := runWorkerPoolQueue(q)
 		assert.NoError(t, q.FlushWithContext(context.Background(), 0))
 		stop()

modules/setting/queue.go

@@ -16,9 +16,9 @@ type QueueSettings struct {
 	Type    string
 	Datadir string
 	ConnStr string // for levelqueue or redis
-	Length  int    // queue length
+	Length  int    // max queue length before blocking
 
-	QueueName, SetName string
+	QueueName, SetName string // the name suffix for storage (db key, redis key), "set" is for unique queue
 
 	BatchLength int
 	MaxWorkers  int
@@ -29,14 +29,15 @@ var queueSettingsDefault = QueueSettings{
 	Datadir: "queues/", // relative to AppDataPath
 	Length:  20,        // queue length before a channel queue will block
 
-	QueueName:   "_queue",  // the suffix of the default redis/disk queue name
-	SetName:     "_unique", // the suffix of the default redis/disk unique queue set name (for unique queue)
+	QueueName:   "_queue",
+	SetName:     "_unique",
 	BatchLength: 20,
 	MaxWorkers:  10,
 }
 
 func GetQueueSettings(rootCfg ConfigProvider, name string) (QueueSettings, error) {
 	cfg := queueSettingsDefault
+	cfg.Name = name
 	if sec, err := rootCfg.GetSection("queue"); err == nil {
 		if err = sec.MapTo(&cfg); err != nil {
 			log.Error("Failed to map queue common config for %q: %v", name, err)
@@ -78,8 +79,6 @@ func GetQueueSettings(rootCfg ConfigProvider, name string) (QueueSettings, error
 	return cfg, nil
 }
 
-// LoadQueueSettings sets up the default settings for Queues
-// This is exported for tests to be able to use the queue
 func LoadQueueSettings() {
 	loadQueueFrom(CfgProvider)
 }
@@ -95,8 +94,6 @@ func loadQueueFrom(rootCfg ConfigProvider) {
 	handleOldLengthConfiguration(rootCfg, "mirror", "repository", "MIRROR_QUEUE_LENGTH")
 }
 
-// handleOldLengthConfiguration allows fallback to older configuration. `[queue.name]` `LENGTH` will override this configuration, but
-// if that is left unset then we should fall back to the older configuration. (Except where the new length woul be <=0)
 func handleOldLengthConfiguration(rootCfg ConfigProvider, newQueueName, oldSection, oldKey string) {
 	if rootCfg.Section(oldSection).HasKey(oldKey) {
 		log.Fatal("Removed queue option:`[%s].%s`. Use new options in`[queue.%s]`", oldSection, oldKey, newQueueName)

services/pull/check_test.go

@@ -29,7 +29,7 @@ func TestPullRequest_AddToTaskQueue(t *testing.T) {
 		return nil
 	}
 	iniCfg := queue.IniConfig{Length: 10, BatchLength: 1, Workers: 1}
-	prPatchCheckerQueue = queue.NewWorkerPoolQueueByIniConfig("pr_patch_checker", iniCfg, testHandler, true)
+	prPatchCheckerQueue = queue.NewWorkerPoolQueueBySetting("pr_patch_checker", iniCfg, testHandler, true)
 
 	var queueShutdown []func()
 	var queueTerminate []func()