As per review

Signed-off-by: Andrew Thornton <[email protected]>

Author: zeripath

modules/doctor/queue.go

@@ -14,22 +14,6 @@ import (
 	"gitea.com/lunny/levelqueue"
 )
 
-var uniqueQueueNames = []string{
-	"code_indexer",
-	"repo_stats_update",
-	"mirror",
-	"pr_patch_checker",
-	"repo-archive",
-}
-
-var queueNames = []string{
-	"issue_indexer",
-	"notification-service",
-	"mail",
-	"push_update",
-	"task",
-}
-
 var levelqueueTypes = []string{
 	string(queue.PersistableChannelQueueType),
 	string(queue.PersistableChannelUniqueQueueType),
@@ -38,8 +22,8 @@ var levelqueueTypes = []string{
 }
 
 func checkUniqueQueues(ctx context.Context, logger log.Logger, autofix bool) error {
-	for _, name := range uniqueQueueNames {
-		q := setting.GetQueueSettings(name)
+	for _, name := range queue.KnownUniqueQueueNames {
+		q := setting.GetQueueSettings(string(name))
 		if q.Type == "" {
 			q.Type = string(queue.PersistableChannelQueueType)
 		}
@@ -62,7 +46,7 @@ func checkUniqueQueues(ctx context.Context, logger log.Logger, autofix bool) err
 
 		db, err := nosql.GetManager().GetLevelDB(connection)
 		if err != nil {
-			logger.Error("Queue: %s\nUnable to open DB connection %s: %v", q.Name, connection, err)
+			logger.Error("Queue: %s\nUnable to open DB connection %q: %v", q.Name, connection, err)
 			return err
 		}
 		defer db.Close()
@@ -138,8 +122,15 @@ func checkUniqueQueues(ctx context.Context, logger log.Logger, autofix bool) err
 
 func queueListDB(ctx context.Context, logger log.Logger, autofix bool) error {
 	connections := []string{}
+	queueNames := make([]string, 0, len(queue.KnownUniqueQueueNames)+len(queue.KnownQueueNames))
+	for _, name := range queue.KnownUniqueQueueNames {
+		queueNames = append(queueNames, string(name))
+	}
+	for _, name := range queue.KnownQueueNames {
+		queueNames = append(queueNames, string(name))
+	}
 
-	for _, name := range append(uniqueQueueNames, queueNames...) {
+	for _, name := range queueNames {
 		q := setting.GetQueueSettings(name)
 		if q.Type == "" {
 			q.Type = string(queue.PersistableChannelQueueType)
@@ -159,6 +150,7 @@ func queueListDB(ctx context.Context, logger log.Logger, autofix bool) error {
 			for _, connection := range connections {
 				if connection == q.ConnectionString {
 					found = true
+					break
 				}
 			}
 			if !found {
@@ -170,6 +162,7 @@ func queueListDB(ctx context.Context, logger log.Logger, autofix bool) error {
 		for _, connection := range connections {
 			if connection == q.DataDir {
 				found = true
+				break
 			}
 		}
 		if !found {
@@ -181,7 +174,7 @@ func queueListDB(ctx context.Context, logger log.Logger, autofix bool) error {
 		logger.Info("LevelDB: %s", connection)
 		db, err := nosql.GetManager().GetLevelDB(connection)
 		if err != nil {
-			logger.Error("Connection: %s Unable to open DB: %v", connection, err)
+			logger.Error("Connection: %q Unable to open DB: %v", connection, err)
 			return err
 		}
 		defer db.Close()
@@ -204,6 +197,7 @@ func init() {
 		SkipDatabaseInitialization: false,
 		Priority:                   1,
 	})
+
 	Register(&Check{
 		Title:                      "List all entries in leveldb",
 		Name:                       "queues-listdb",

modules/indexer/code/indexer.go

@@ -163,7 +163,7 @@ func Init() {
 			return unhandled
 		}
 
-		indexerQueue = queue.CreateUniqueQueue("code_indexer", handler, &IndexerData{})
+		indexerQueue = queue.CreateUniqueQueue(queue.CodeIndexerQueueName, handler, &IndexerData{})
 		if indexerQueue == nil {
 			log.Fatal("Unable to create codes indexer queue")
 		}

modules/indexer/issues/indexer.go

@@ -154,7 +154,7 @@ func InitIssueIndexer(syncReindex bool) {
 			return nil
 		}
 
-		issueIndexerQueue = queue.CreateQueue("issue_indexer", handler, &IndexerData{})
+		issueIndexerQueue = queue.CreateQueue(queue.IssueIndexerQueueName, handler, &IndexerData{})
 
 		if issueIndexerQueue == nil {
 			log.Fatal("Unable to create issue indexer queue")

modules/indexer/stats/queue.go

@@ -28,9 +28,9 @@ func handle(data ...queue.Data) []queue.Data {
 }
 
 func initStatsQueue() error {
-	statsQueue = queue.CreateUniqueQueue("repo_stats_update", handle, int64(0))
+	statsQueue = queue.CreateUniqueQueue(queue.RepoStatsUpdateQueueName, handle, int64(0))
 	if statsQueue == nil {
-		return fmt.Errorf("Unable to create repo_stats_update Queue")
+		return fmt.Errorf("unable to create repo_stats_update Queue")
 	}
 
 	go graceful.GetManager().RunWithShutdownFns(statsQueue.Run)

modules/notification/ui/ui.go

@@ -34,7 +34,7 @@ var _ base.Notifier = &notificationService{}
 // NewNotifier create a new notificationService notifier
 func NewNotifier() base.Notifier {
 	ns := &notificationService{}
-	ns.issueQueue = queue.CreateQueue("notification-service", ns.handle, issueNotificationOpts{})
+	ns.issueQueue = queue.CreateQueue(queue.NotificationQueueName, ns.handle, issueNotificationOpts{})
 	return ns
 }
 

modules/queue/setting.go

@@ -13,6 +13,48 @@ import (
 	"code.gitea.io/gitea/modules/setting"
 )
 
+// UniqueQueueName represents an expected name for an UniqueQueue
+type UniqueQueueName string
+
+// list of all expected UniqueQueues
+const (
+	CodeIndexerQueueName     UniqueQueueName = "code_indexer"
+	RepoStatsUpdateQueueName UniqueQueueName = "repo_stats_update"
+	MirrorQueueName          UniqueQueueName = "mirror"
+	PRPatchQueueName         UniqueQueueName = "pr_patch_checker"
+	RepoArchiveQueueName     UniqueQueueName = "repo-archive"
+)
+
+// KnownUniqueQueueNames represents the list of expected unique queues
+var KnownUniqueQueueNames = []UniqueQueueName{
+	CodeIndexerQueueName,
+	RepoStatsUpdateQueueName,
+	MirrorQueueName,
+	PRPatchQueueName,
+	RepoArchiveQueueName,
+}
+
+// QueueName represents an expected name for Queue
+type QueueName string
+
+// list of all expected Queues
+const (
+	IssueIndexerQueueName QueueName = "issue_indexer"
+	NotificationQueueName QueueName = "notification-service"
+	MailerQueueName       QueueName = "mail"
+	PushUpdateQueueName   QueueName = "push_update"
+	TaskQueueName         QueueName = "task"
+)
+
+// KnownQueueNames represents the list of expected queues
+var KnownQueueNames = []QueueName{
+	IssueIndexerQueueName,
+	NotificationQueueName,
+	MailerQueueName,
+	PushUpdateQueueName,
+	TaskQueueName,
+}
+
 func validType(t string) (Type, error) {
 	if len(t) == 0 {
 		return PersistableChannelQueueType, nil
@@ -37,8 +79,19 @@ func getQueueSettings(name string) (setting.QueueSettings, []byte) {
 }
 
 // CreateQueue for name with provided handler and exemplar
-func CreateQueue(name string, handle HandlerFunc, exemplar interface{}) Queue {
-	q, cfg := getQueueSettings(name)
+func CreateQueue(name QueueName, handle HandlerFunc, exemplar interface{}) Queue {
+	found := false
+	for _, expected := range KnownQueueNames {
+		if name == expected {
+			found = true
+			break
+		}
+	}
+	if !found {
+		log.Warn("%s is not an expected name for an Queue", name)
+	}
+
+	q, cfg := getQueueSettings(string(name))
 	if len(cfg) == 0 {
 		return nil
 	}
@@ -58,7 +111,7 @@ func CreateQueue(name string, handle HandlerFunc, exemplar interface{}) Queue {
 			MaxAttempts: q.MaxAttempts,
 			Config:      cfg,
 			QueueLength: q.QueueLength,
-			Name:        name,
+			Name:        string(name),
 		}, exemplar)
 	}
 	if err != nil {
@@ -79,8 +132,19 @@ func CreateQueue(name string, handle HandlerFunc, exemplar interface{}) Queue {
 }
 
 // CreateUniqueQueue for name with provided handler and exemplar
-func CreateUniqueQueue(name string, handle HandlerFunc, exemplar interface{}) UniqueQueue {
-	q, cfg := getQueueSettings(name)
+func CreateUniqueQueue(name UniqueQueueName, handle HandlerFunc, exemplar interface{}) UniqueQueue {
+	found := false
+	for _, expected := range KnownUniqueQueueNames {
+		if name == expected {
+			found = true
+			break
+		}
+	}
+	if !found {
+		log.Warn("%s is not an expected name for an UniqueQueue", name)
+	}
+
+	q, cfg := getQueueSettings(string(name))
 	if len(cfg) == 0 {
 		return nil
 	}
@@ -107,6 +171,7 @@ func CreateUniqueQueue(name string, handle HandlerFunc, exemplar interface{}) Un
 			MaxAttempts: q.MaxAttempts,
 			Config:      cfg,
 			QueueLength: q.QueueLength,
+			Name:        string(name),
 		}, exemplar)
 	}
 	if err != nil {

services/mailer/mailer.go

@@ -346,7 +346,7 @@ func NewContext() {
 		Sender = &dummySender{}
 	}
 
-	mailQueue = queue.CreateQueue("mail", func(data ...queue.Data) []queue.Data {
+	mailQueue = queue.CreateQueue(queue.MailerQueueName, func(data ...queue.Data) []queue.Data {
 		for _, datum := range data {
 			msg := datum.(*Message)
 			gomailMsg := msg.ToMessage()

services/mirror/mirror.go

@@ -151,7 +151,7 @@ func InitSyncMirrors() {
 	if !setting.Mirror.Enabled {
 		return
 	}
-	mirrorQueue = queue.CreateUniqueQueue("mirror", queueHandle, new(SyncRequest))
+	mirrorQueue = queue.CreateUniqueQueue(queue.MirrorQueueName, queueHandle, new(SyncRequest))
 
 	go graceful.GetManager().RunWithShutdownFns(mirrorQueue.Run)
 }

services/pull/check.go

@@ -278,10 +278,10 @@ func CheckPrsForBaseBranch(baseRepo *repo_model.Repository, baseBranchName strin
 
 // Init runs the task queue to test all the checking status pull requests
 func Init() error {
-	prQueue = queue.CreateUniqueQueue("pr_patch_checker", handle, "")
+	prQueue = queue.CreateUniqueQueue(queue.PRPatchQueueName, handle, "")
 
 	if prQueue == nil {
-		return fmt.Errorf("Unable to create pr_patch_checker Queue")
+		return fmt.Errorf("unable to create pr_patch_checker Queue")
 	}
 
 	go graceful.GetManager().RunWithShutdownFns(prQueue.Run)

services/repository/archiver/archiver.go

@@ -261,9 +261,9 @@ func Init() error {
 		return nil
 	}
 
-	archiverQueue = queue.CreateUniqueQueue("repo-archive", handler, new(ArchiveRequest))
+	archiverQueue = queue.CreateUniqueQueue(queue.RepoArchiveQueueName, handler, new(ArchiveRequest))
 	if archiverQueue == nil {
-		return errors.New("unable to create codes indexer queue")
+		return errors.New("unable to create repo archiver queue")
 	}
 
 	go graceful.GetManager().RunWithShutdownFns(archiverQueue.Run)

services/repository/push.go

@@ -44,7 +44,7 @@ func handle(data ...queue.Data) []queue.Data {
 }
 
 func initPushQueue() error {
-	pushQueue = queue.CreateQueue("push_update", handle, []*repo_module.PushUpdateOptions{})
+	pushQueue = queue.CreateQueue(queue.PushUpdateQueueName, handle, []*repo_module.PushUpdateOptions{})
 	if pushQueue == nil {
 		return errors.New("unable to create push_update Queue")
 	}

services/task/task.go

@@ -32,16 +32,16 @@ func Run(t *models.Task) error {
 	case structs.TaskTypeMigrateRepo:
 		return runMigrateTask(t)
 	default:
-		return fmt.Errorf("Unknown task type: %d", t.Type)
+		return fmt.Errorf("unknown task type: %d", t.Type)
 	}
 }
 
 // Init will start the service to get all unfinished tasks and run them
 func Init() error {
-	taskQueue = queue.CreateQueue("task", handle, &models.Task{})
+	taskQueue = queue.CreateQueue(queue.TaskQueueName, handle, &models.Task{})
 
 	if taskQueue == nil {
-		return fmt.Errorf("Unable to create Task Queue")
+		return fmt.Errorf("unable to create Task Queue")
 	}
 
 	go graceful.GetManager().RunWithShutdownFns(taskQueue.Run)