mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

jobstore (#15250)

Просмотр исходного кода 
* Migration completed

* Modify test case

* Update en.json

* Fix layers

* Lint: remove unnecessary use of sprint

* trigger CI

* fix error

* Fixes

* fix test

Co-authored-by: Rodrigo Villablanca <villa061004@gmail.com>
Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
Этот коммит содержится в:
Agniva De Sarker
2020-08-20 19:37:19 +05:30
коммит произвёл GitHub
родитель b451b3cf86
Коммит 78766625df
12 изменённых файлов: 474 добавлений и 251 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

70
jobs/jobs.go
Просмотреть файл

@@ -5,12 +5,13 @@ package jobs
import (
"context"
"time"
"errors"
"net/http"
"time"
"github.com/mattermost/mattermost-server/v5/mlog"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/store"
)
const (
@@ -31,18 +32,34 @@ func (srv *JobServer) CreateJob(jobType string, jobData map[string]string) (*mod
}
if _, err := srv.Store.Job().Save(&job); err != nil {
return nil, err
return nil, model.NewAppError("CreateJob", "app.job.save.app_error", nil, err.Error(), http.StatusInternalServerError)
}
return &job, nil
}
func (srv *JobServer) GetJob(id string) (*model.Job, *model.AppError) {
return srv.Store.Job().Get(id)
job, err := srv.Store.Job().Get(id)
if err != nil {
var nfErr *store.ErrNotFound
switch {
case errors.As(err, &nfErr):
return nil, model.NewAppError("GetJob", "app.job.get.app_error", nil, nfErr.Error(), http.StatusNotFound)
default:
return nil, model.NewAppError("GetJob", "app.job.get.app_error", nil, err.Error(), http.StatusInternalServerError)
}
}
return job, nil
}
func (srv *JobServer) ClaimJob(job *model.Job) (bool, *model.AppError) {
return srv.Store.Job().UpdateStatusOptimistically(job.Id, model.JOB_STATUS_PENDING, model.JOB_STATUS_IN_PROGRESS)
updated, err := srv.Store.Job().UpdateStatusOptimistically(job.Id, model.JOB_STATUS_PENDING, model.JOB_STATUS_IN_PROGRESS)
if err != nil {
return false, model.NewAppError("ClaimJob", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
return updated, nil
}
func (srv *JobServer) SetJobProgress(job *model.Job, progress int64) *model.AppError {
@@ -50,21 +67,21 @@ func (srv *JobServer) SetJobProgress(job *model.Job, progress int64) *model.AppE
job.Progress = progress
if _, err := srv.Store.Job().UpdateOptimistically(job, model.JOB_STATUS_IN_PROGRESS); err != nil {
return err
return model.NewAppError("SetJobProgress", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
return nil
}
func (srv *JobServer) SetJobWarning(job *model.Job) *model.AppError {
if _, err := srv.Store.Job().UpdateStatus(job.Id, model.JOB_STATUS_WARNING); err != nil {
return err
return model.NewAppError("SetJobWarning", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
return nil
}
func (srv *JobServer) SetJobSuccess(job *model.Job) *model.AppError {
if _, err := srv.Store.Job().UpdateStatus(job.Id, model.JOB_STATUS_SUCCESS); err != nil {
return err
return model.NewAppError("SetJobSuccess", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
return nil
}
@@ -72,7 +89,11 @@ func (srv *JobServer) SetJobSuccess(job *model.Job) *model.AppError {
func (srv *JobServer) SetJobError(job *model.Job, jobError *model.AppError) *model.AppError {
if jobError == nil {
_, err := srv.Store.Job().UpdateStatus(job.Id, model.JOB_STATUS_ERROR)
return err
if err != nil {
return model.NewAppError("SetJobError", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
return nil
}
job.Status = model.JOB_STATUS_ERROR
@@ -84,16 +105,16 @@ func (srv *JobServer) SetJobError(job *model.Job, jobError *model.AppError) *mod
updated, err := srv.Store.Job().UpdateOptimistically(job, model.JOB_STATUS_IN_PROGRESS)
if err != nil {
return err
return model.NewAppError("SetJobError", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
if !updated {
updated, err = srv.Store.Job().UpdateOptimistically(job, model.JOB_STATUS_CANCEL_REQUESTED)
if err != nil {
return err
return model.NewAppError("SetJobError", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
if !updated {
return model.NewAppError("Jobs.SetJobError", "jobs.set_job_error.update.error", nil, "id="+job.Id, http.StatusInternalServerError)
return model.NewAppError("SetJobError", "jobs.set_job_error.update.error", nil, "id="+job.Id, http.StatusInternalServerError)
}
}
@@ -102,7 +123,7 @@ func (srv *JobServer) SetJobError(job *model.Job, jobError *model.AppError) *mod
func (srv *JobServer) SetJobCanceled(job *model.Job) *model.AppError {
if _, err := srv.Store.Job().UpdateStatus(job.Id, model.JOB_STATUS_CANCELED); err != nil {
return err
return model.NewAppError("SetJobCanceled", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
return nil
}
@@ -111,7 +132,7 @@ func (srv *JobServer) UpdateInProgressJobData(job *model.Job) *model.AppError {
job.Status = model.JOB_STATUS_IN_PROGRESS
job.LastActivityAt = model.GetMillis()
if _, err := srv.Store.Job().UpdateOptimistically(job, model.JOB_STATUS_IN_PROGRESS); err != nil {
return err
return model.NewAppError("UpdateInProgressJobData", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
return nil
}
@@ -119,7 +140,7 @@ func (srv *JobServer) UpdateInProgressJobData(job *model.Job) *model.AppError {
func (srv *JobServer) RequestCancellation(jobId string) *model.AppError {
updated, err := srv.Store.Job().UpdateStatusOptimistically(jobId, model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED)
if err != nil {
return err
return model.NewAppError("RequestCancellation", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
if updated {
return nil
@@ -127,14 +148,14 @@ func (srv *JobServer) RequestCancellation(jobId string) *model.AppError {
updated, err = srv.Store.Job().UpdateStatusOptimistically(jobId, model.JOB_STATUS_IN_PROGRESS, model.JOB_STATUS_CANCEL_REQUESTED)
if err != nil {
return err
return model.NewAppError("RequestCancellation", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
}
if updated {
return nil
}
return model.NewAppError("Jobs.RequestCancellation", "jobs.request_cancellation.status.error", nil, "id="+jobId, http.StatusInternalServerError)
return model.NewAppError("RequestCancellation", "jobs.request_cancellation.status.error", nil, "id="+jobId, http.StatusInternalServerError)
}
func (srv *JobServer) CancellationWatcher(ctx context.Context, jobId string, cancelChan chan interface{}) {
@@ -168,7 +189,7 @@ func GenerateNextStartDateTime(now time.Time, nextStartTime time.Time) *time.Tim
func (srv *JobServer) CheckForPendingJobsByType(jobType string) (bool, *model.AppError) {
count, err := srv.Store.Job().GetCountByStatusAndType(model.JOB_STATUS_PENDING, jobType)
if err != nil {
return false, err
return false, model.NewAppError("CheckForPendingJobsByType", "app.job.get_count_by_status_and_type.app_error", nil, err.Error(), http.StatusInternalServerError)
}
return count > 0, nil
}
@@ -178,5 +199,16 @@ func (srv *JobServer) GetLastSuccessfulJobByType(jobType string) (*model.Job, *m
if jobType == model.JOB_TYPE_MESSAGE_EXPORT {
statuses = []string{model.JOB_STATUS_WARNING, model.JOB_STATUS_SUCCESS}
}
return srv.Store.Job().GetNewestJobByStatusesAndType(statuses, jobType)
job, err := srv.Store.Job().GetNewestJobByStatusesAndType(statuses, jobType)
if err != nil {
var nfErr *store.ErrNotFound
switch {
case errors.As(err, &nfErr):
return nil, model.NewAppError("GetLastSuccessfulJobByType", "app.job.get_newest_job_by_status_and_type.app_error", nil, nfErr.Error(), http.StatusNotFound)
default:
return nil, model.NewAppError("GetLastSuccessfulJobByType", "app.job.get_newest_job_by_status_and_type.app_error", nil, err.Error(), http.StatusInternalServerError)
}
}
return job, nil
}