summaryrefslogtreecommitdiffstats
path: root/modules/task/queue_redis.go
diff options
context:
space:
mode:
authorLunny Xiao <xiaolunwen@gmail.com>2019-10-13 21:23:14 +0800
committerGitHub <noreply@github.com>2019-10-13 21:23:14 +0800
commitf2a3abc683ad4b2177b7c7c6160a2c0b4316120a (patch)
tree3b92f34b9bb9a015072f511dc5cf6340af18eda5 /modules/task/queue_redis.go
parent0a96e59884ca5c4fedc8c3d166d97f35b245ad6e (diff)
downloadgitea-f2a3abc683ad4b2177b7c7c6160a2c0b4316120a.tar.gz
gitea-f2a3abc683ad4b2177b7c7c6160a2c0b4316120a.zip
Move migrating repository from frontend to backend (#6200)
* move migrating to backend * add loading image when migrating and fix tests * fix format * fix lint * add redis task queue support and improve docs * add redis vendor * fix vet * add database migrations and fix app.ini sample * add comments for task section on app.ini.sample * Update models/migrations/v84.go Co-Authored-By: lunny <xiaolunwen@gmail.com> * Update models/repo.go Co-Authored-By: lunny <xiaolunwen@gmail.com> * move migrating to backend * add loading image when migrating and fix tests * fix fmt * add redis task queue support and improve docs * fix fixtures * fix fixtures * fix duplicate function on index.js * fix tests * rename repository statuses * check if repository is being create when SSH request * fix lint * fix template * some improvements * fix template * unified migrate options * fix lint * fix loading page * refactor * When gitea restart, don't restart the running tasks because we may have servel gitea instances, that may break the migration * fix js * Update models/repo.go Co-Authored-By: guillep2k <18600385+guillep2k@users.noreply.github.com> * Update docs/content/doc/advanced/config-cheat-sheet.en-us.md Co-Authored-By: guillep2k <18600385+guillep2k@users.noreply.github.com> * fix tests * rename ErrTaskIsNotExist to ErrTaskDoesNotExist * delete release after add one on tests to make it run happy * fix tests * fix tests * improve codes * fix lint * fix lint * fix migrations
Diffstat (limited to 'modules/task/queue_redis.go')
-rw-r--r--modules/task/queue_redis.go130
1 files changed, 130 insertions, 0 deletions
diff --git a/modules/task/queue_redis.go b/modules/task/queue_redis.go
new file mode 100644
index 0000000000..127de0cdbf
--- /dev/null
+++ b/modules/task/queue_redis.go
@@ -0,0 +1,130 @@
+// 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 task
+
+import (
+ "encoding/json"
+ "errors"
+ "strconv"
+ "strings"
+ "time"
+
+ "code.gitea.io/gitea/models"
+ "code.gitea.io/gitea/modules/log"
+
+ "github.com/go-redis/redis"
+)
+
+var (
+ _ Queue = &RedisQueue{}
+)
+
+type redisClient interface {
+ RPush(key string, args ...interface{}) *redis.IntCmd
+ LPop(key string) *redis.StringCmd
+ Ping() *redis.StatusCmd
+}
+
+// RedisQueue redis queue
+type RedisQueue struct {
+ client redisClient
+ queueName string
+ closeChan chan bool
+}
+
+func parseConnStr(connStr string) (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 "addrs":
+ addrs = items[1]
+ case "password":
+ password = items[1]
+ case "db":
+ dbIdx, err = strconv.Atoi(items[1])
+ if err != nil {
+ return
+ }
+ }
+ }
+ return
+}
+
+// NewRedisQueue creates single redis or cluster redis queue
+func NewRedisQueue(addrs string, password string, dbIdx int) (*RedisQueue, error) {
+ dbs := strings.Split(addrs, ",")
+ var queue = RedisQueue{
+ queueName: "task_queue",
+ closeChan: make(chan bool),
+ }
+ if len(dbs) == 0 {
+ return nil, errors.New("no redis host found")
+ } else if len(dbs) == 1 {
+ queue.client = redis.NewClient(&redis.Options{
+ Addr: strings.TrimSpace(dbs[0]), // use default Addr
+ Password: password, // no password set
+ DB: dbIdx, // use default DB
+ })
+ } else {
+ // cluster will ignore db
+ queue.client = redis.NewClusterClient(&redis.ClusterOptions{
+ Addrs: dbs,
+ Password: password,
+ })
+ }
+ if err := queue.client.Ping().Err(); err != nil {
+ return nil, err
+ }
+ return &queue, nil
+}
+
+// Run starts to run the queue
+func (r *RedisQueue) Run() error {
+ for {
+ select {
+ case <-r.closeChan:
+ return nil
+ case <-time.After(time.Millisecond * 100):
+ }
+
+ bs, err := r.client.LPop(r.queueName).Bytes()
+ if err != nil {
+ if err != redis.Nil {
+ log.Error("LPop failed: %v", err)
+ }
+ time.Sleep(time.Millisecond * 100)
+ continue
+ }
+
+ var task models.Task
+ err = json.Unmarshal(bs, &task)
+ if err != nil {
+ log.Error("Unmarshal task failed: %s", err.Error())
+ } else {
+ err = Run(&task)
+ if err != nil {
+ log.Error("Run task failed: %s", err.Error())
+ }
+ }
+ }
+}
+
+// Push implements Queue
+func (r *RedisQueue) Push(task *models.Task) error {
+ bs, err := json.Marshal(task)
+ if err != nil {
+ return err
+ }
+ return r.client.RPush(r.queueName, bs).Err()
+}
+
+// Stop stop the queue
+func (r *RedisQueue) Stop() {
+ r.closeChan <- true
+}