mirror of
https://github.com/go-gitea/gitea
synced 2025-01-09 21:35:58 +01:00
215 lines
6.9 KiB
Go
215 lines
6.9 KiB
Go
// Copyright 2019 The Gitea Authors. All rights reserved.
|
|
// Use of this source code is governed by a MIT-style
|
|
// license that can be found in the LICENSE file.
|
|
|
|
package setting
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"path"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"code.gitea.io/gitea/modules/log"
|
|
"code.gitea.io/gitea/modules/queue"
|
|
)
|
|
|
|
type queueSettings struct {
|
|
DataDir string
|
|
Length int
|
|
BatchLength int
|
|
ConnectionString string
|
|
Type string
|
|
Network string
|
|
Addresses string
|
|
Password string
|
|
QueueName string
|
|
DBIndex int
|
|
WrapIfNecessary bool
|
|
MaxAttempts int
|
|
Timeout time.Duration
|
|
Workers int
|
|
MaxWorkers int
|
|
BlockTimeout time.Duration
|
|
BoostTimeout time.Duration
|
|
BoostWorkers int
|
|
}
|
|
|
|
// Queue settings
|
|
var Queue = queueSettings{}
|
|
|
|
// CreateQueue for name with provided handler and exemplar
|
|
func CreateQueue(name string, handle queue.HandlerFunc, exemplar interface{}) queue.Queue {
|
|
q := getQueueSettings(name)
|
|
opts := make(map[string]interface{})
|
|
opts["Name"] = name
|
|
opts["QueueLength"] = q.Length
|
|
opts["BatchLength"] = q.BatchLength
|
|
opts["DataDir"] = q.DataDir
|
|
opts["Addresses"] = q.Addresses
|
|
opts["Network"] = q.Network
|
|
opts["Password"] = q.Password
|
|
opts["DBIndex"] = q.DBIndex
|
|
opts["QueueName"] = q.QueueName
|
|
opts["Workers"] = q.Workers
|
|
opts["MaxWorkers"] = q.MaxWorkers
|
|
opts["BlockTimeout"] = q.BlockTimeout
|
|
opts["BoostTimeout"] = q.BoostTimeout
|
|
opts["BoostWorkers"] = q.BoostWorkers
|
|
|
|
cfg, err := json.Marshal(opts)
|
|
if err != nil {
|
|
log.Error("Unable to marshall generic options: %v Error: %v", opts, err)
|
|
log.Error("Unable to create queue for %s", name, err)
|
|
return nil
|
|
}
|
|
|
|
returnable, err := queue.CreateQueue(queue.Type(q.Type), handle, cfg, exemplar)
|
|
if q.WrapIfNecessary && err != nil {
|
|
log.Warn("Unable to create queue for %s: %v", name, err)
|
|
log.Warn("Attempting to create wrapped queue")
|
|
returnable, err = queue.CreateQueue(queue.WrappedQueueType, handle, queue.WrappedQueueConfiguration{
|
|
Underlying: queue.Type(q.Type),
|
|
Timeout: q.Timeout,
|
|
MaxAttempts: q.MaxAttempts,
|
|
Config: cfg,
|
|
QueueLength: q.Length,
|
|
}, exemplar)
|
|
}
|
|
if err != nil {
|
|
log.Error("Unable to create queue for %s: %v", name, err)
|
|
return nil
|
|
}
|
|
return returnable
|
|
}
|
|
|
|
func getQueueSettings(name string) queueSettings {
|
|
q := queueSettings{}
|
|
sec := Cfg.Section("queue." + name)
|
|
// DataDir is not directly inheritable
|
|
q.DataDir = path.Join(Queue.DataDir, name)
|
|
for _, key := range sec.Keys() {
|
|
switch key.Name() {
|
|
case "DATADIR":
|
|
q.DataDir = key.MustString(q.DataDir)
|
|
}
|
|
}
|
|
if !path.IsAbs(q.DataDir) {
|
|
q.DataDir = path.Join(AppDataPath, q.DataDir)
|
|
}
|
|
sec.Key("DATADIR").SetValue(q.DataDir)
|
|
// The rest are...
|
|
q.Length = sec.Key("LENGTH").MustInt(Queue.Length)
|
|
q.BatchLength = sec.Key("BATCH_LENGTH").MustInt(Queue.BatchLength)
|
|
q.ConnectionString = sec.Key("CONN_STR").MustString(Queue.ConnectionString)
|
|
validTypes := queue.RegisteredTypesAsString()
|
|
q.Type = sec.Key("TYPE").In(Queue.Type, validTypes)
|
|
q.WrapIfNecessary = sec.Key("WRAP_IF_NECESSARY").MustBool(Queue.WrapIfNecessary)
|
|
q.MaxAttempts = sec.Key("MAX_ATTEMPTS").MustInt(Queue.MaxAttempts)
|
|
q.Timeout = sec.Key("TIMEOUT").MustDuration(Queue.Timeout)
|
|
q.Workers = sec.Key("WORKERS").MustInt(Queue.Workers)
|
|
q.MaxWorkers = sec.Key("MAX_WORKERS").MustInt(Queue.MaxWorkers)
|
|
q.BlockTimeout = sec.Key("BLOCK_TIMEOUT").MustDuration(Queue.BlockTimeout)
|
|
q.BoostTimeout = sec.Key("BOOST_TIMEOUT").MustDuration(Queue.BoostTimeout)
|
|
q.BoostWorkers = sec.Key("BOOST_WORKERS").MustInt(Queue.BoostWorkers)
|
|
q.QueueName = sec.Key("QUEUE_NAME").MustString(Queue.QueueName)
|
|
|
|
q.Network, q.Addresses, q.Password, q.DBIndex, _ = ParseQueueConnStr(q.ConnectionString)
|
|
return q
|
|
}
|
|
|
|
// NewQueueService sets up the default settings for Queues
|
|
// This is exported for tests to be able to use the queue
|
|
func NewQueueService() {
|
|
sec := Cfg.Section("queue")
|
|
Queue.DataDir = sec.Key("DATADIR").MustString("queues/")
|
|
if !path.IsAbs(Queue.DataDir) {
|
|
Queue.DataDir = path.Join(AppDataPath, Queue.DataDir)
|
|
}
|
|
Queue.Length = sec.Key("LENGTH").MustInt(20)
|
|
Queue.BatchLength = sec.Key("BATCH_LENGTH").MustInt(20)
|
|
Queue.ConnectionString = sec.Key("CONN_STR").MustString(path.Join(AppDataPath, ""))
|
|
validTypes := queue.RegisteredTypesAsString()
|
|
Queue.Type = sec.Key("TYPE").In(string(queue.PersistableChannelQueueType), validTypes)
|
|
Queue.Network, Queue.Addresses, Queue.Password, Queue.DBIndex, _ = ParseQueueConnStr(Queue.ConnectionString)
|
|
Queue.WrapIfNecessary = sec.Key("WRAP_IF_NECESSARY").MustBool(true)
|
|
Queue.MaxAttempts = sec.Key("MAX_ATTEMPTS").MustInt(10)
|
|
Queue.Timeout = sec.Key("TIMEOUT").MustDuration(GracefulHammerTime + 30*time.Second)
|
|
Queue.Workers = sec.Key("WORKERS").MustInt(1)
|
|
Queue.MaxWorkers = sec.Key("MAX_WORKERS").MustInt(10)
|
|
Queue.BlockTimeout = sec.Key("BLOCK_TIMEOUT").MustDuration(1 * time.Second)
|
|
Queue.BoostTimeout = sec.Key("BOOST_TIMEOUT").MustDuration(5 * time.Minute)
|
|
Queue.BoostWorkers = sec.Key("BOOST_WORKERS").MustInt(5)
|
|
Queue.QueueName = sec.Key("QUEUE_NAME").MustString(Queue.QueueName)
|
|
|
|
hasWorkers := false
|
|
for _, key := range Cfg.Section("queue.notification").Keys() {
|
|
if key.Name() == "WORKERS" {
|
|
hasWorkers = true
|
|
break
|
|
}
|
|
}
|
|
if !hasWorkers {
|
|
Cfg.Section("queue.notification").Key("WORKERS").SetValue("5")
|
|
}
|
|
|
|
// Now handle the old issue_indexer configuration
|
|
section := Cfg.Section("queue.issue_indexer")
|
|
issueIndexerSectionMap := map[string]string{}
|
|
for _, key := range section.Keys() {
|
|
issueIndexerSectionMap[key.Name()] = key.Value()
|
|
}
|
|
if _, ok := issueIndexerSectionMap["TYPE"]; !ok {
|
|
switch Indexer.IssueQueueType {
|
|
case LevelQueueType:
|
|
section.Key("TYPE").SetValue("level")
|
|
case ChannelQueueType:
|
|
section.Key("TYPE").SetValue("persistable-channel")
|
|
case RedisQueueType:
|
|
section.Key("TYPE").SetValue("redis")
|
|
default:
|
|
log.Fatal("Unsupported indexer queue type: %v",
|
|
Indexer.IssueQueueType)
|
|
}
|
|
}
|
|
if _, ok := issueIndexerSectionMap["LENGTH"]; !ok {
|
|
section.Key("LENGTH").SetValue(fmt.Sprintf("%d", Indexer.UpdateQueueLength))
|
|
}
|
|
if _, ok := issueIndexerSectionMap["BATCH_LENGTH"]; !ok {
|
|
section.Key("BATCH_LENGTH").SetValue(fmt.Sprintf("%d", Indexer.IssueQueueBatchNumber))
|
|
}
|
|
if _, ok := issueIndexerSectionMap["DATADIR"]; !ok {
|
|
section.Key("DATADIR").SetValue(Indexer.IssueQueueDir)
|
|
}
|
|
if _, ok := issueIndexerSectionMap["CONN_STR"]; !ok {
|
|
section.Key("CONN_STR").SetValue(Indexer.IssueQueueConnStr)
|
|
}
|
|
}
|
|
|
|
// ParseQueueConnStr parses a queue connection string
|
|
func ParseQueueConnStr(connStr string) (network, addrs, password string, dbIdx int, err error) {
|
|
fields := strings.Fields(connStr)
|
|
for _, f := range fields {
|
|
items := strings.SplitN(f, "=", 2)
|
|
if len(items) < 2 {
|
|
continue
|
|
}
|
|
switch strings.ToLower(items[0]) {
|
|
case "network":
|
|
network = items[1]
|
|
case "addrs":
|
|
addrs = items[1]
|
|
case "password":
|
|
password = items[1]
|
|
case "db":
|
|
dbIdx, err = strconv.Atoi(items[1])
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
return
|
|
}
|