diff options
author | Lunny Xiao <xiaolunwen@gmail.com> | 2019-10-13 21:23:14 +0800 |
---|---|---|
committer | GitHub <noreply@github.com> | 2019-10-13 21:23:14 +0800 |
commit | f2a3abc683ad4b2177b7c7c6160a2c0b4316120a (patch) | |
tree | 3b92f34b9bb9a015072f511dc5cf6340af18eda5 /modules/task/queue_redis.go | |
parent | 0a96e59884ca5c4fedc8c3d166d97f35b245ad6e (diff) | |
download | gitea-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.go | 130 |
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 +} |