123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110 |
- // Copyright 2019 Gitea. 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"
- "fmt"
-
- "code.gitea.io/gitea/models"
- "code.gitea.io/gitea/modules/graceful"
- "code.gitea.io/gitea/modules/log"
- "code.gitea.io/gitea/modules/migrations/base"
- "code.gitea.io/gitea/modules/queue"
- repo_module "code.gitea.io/gitea/modules/repository"
- "code.gitea.io/gitea/modules/structs"
- "code.gitea.io/gitea/modules/timeutil"
- )
-
- // taskQueue is a global queue of tasks
- var taskQueue queue.Queue
-
- // Run a task
- func Run(t *models.Task) error {
- switch t.Type {
- case structs.TaskTypeMigrateRepo:
- return runMigrateTask(t)
- default:
- 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{})
-
- if taskQueue == nil {
- return fmt.Errorf("Unable to create Task Queue")
- }
-
- go graceful.GetManager().RunWithShutdownFns(taskQueue.Run)
-
- return nil
- }
-
- func handle(data ...queue.Data) {
- for _, datum := range data {
- task := datum.(*models.Task)
- if err := Run(task); err != nil {
- log.Error("Run task failed: %v", err)
- }
- }
- }
-
- // MigrateRepository add migration repository to task
- func MigrateRepository(doer, u *models.User, opts base.MigrateOptions) error {
- task, err := CreateMigrateTask(doer, u, opts)
- if err != nil {
- return err
- }
-
- return taskQueue.Push(task)
- }
-
- // CreateMigrateTask creates a migrate task
- func CreateMigrateTask(doer, u *models.User, opts base.MigrateOptions) (*models.Task, error) {
- bs, err := json.Marshal(&opts)
- if err != nil {
- return nil, err
- }
-
- var task = models.Task{
- DoerID: doer.ID,
- OwnerID: u.ID,
- Type: structs.TaskTypeMigrateRepo,
- Status: structs.TaskStatusQueue,
- PayloadContent: string(bs),
- }
-
- if err := models.CreateTask(&task); err != nil {
- return nil, err
- }
-
- repo, err := repo_module.CreateRepository(doer, u, models.CreateRepoOptions{
- Name: opts.RepoName,
- Description: opts.Description,
- OriginalURL: opts.OriginalURL,
- GitServiceType: opts.GitServiceType,
- IsPrivate: opts.Private,
- IsMirror: opts.Mirror,
- Status: models.RepositoryBeingMigrated,
- })
- if err != nil {
- task.EndTime = timeutil.TimeStampNow()
- task.Status = structs.TaskStatusFailed
- err2 := task.UpdateCols("end_time", "status")
- if err2 != nil {
- log.Error("UpdateCols Failed: %v", err2.Error())
- }
- return nil, err
- }
-
- task.RepoID = repo.ID
- if err = task.UpdateCols("repo_id"); err != nil {
- return nil, err
- }
-
- return &task, nil
- }
|