142 lines
3.1 KiB
Go
142 lines
3.1 KiB
Go
package scheduler
|
|
|
|
import (
|
|
"encoding/json"
|
|
"reflect"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.loafle.net/commons/util-go/time/scheduler/storage"
|
|
"git.loafle.net/commons/util-go/time/scheduler/task"
|
|
)
|
|
|
|
type storageAdapter struct {
|
|
taskStorage storage.TaskStorage
|
|
taskTargetRegistry *task.TaskTargetRegistry
|
|
}
|
|
|
|
func (a *storageAdapter) Add(task *task.Task) error {
|
|
ta, err := a.getTaskAttribute(task)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return a.taskStorage.Add(ta)
|
|
}
|
|
|
|
func (a *storageAdapter) Fetch() ([]*task.Task, error) {
|
|
tas, err := a.taskStorage.Fetch()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var tasks []*task.Task
|
|
for _, ta := range tas {
|
|
lastRun, err := time.Parse(time.RFC3339, ta.LastRun)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
nextRun, err := time.Parse(time.RFC3339, ta.NextRun)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
duration, err := time.ParseDuration(ta.Duration)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
isRecurring, err := strconv.Atoi(ta.IsRecurring)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
taskTarget, err := a.taskTargetRegistry.Get(ta.Name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
params, err := paramsFromString(taskTarget, ta.Params)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
t := task.NewWithSchedule(taskTarget, params, &task.TaskSchedule{
|
|
IsRecurring: isRecurring == 1,
|
|
Duration: time.Duration(duration),
|
|
LastRun: lastRun,
|
|
NextRun: nextRun,
|
|
})
|
|
tasks = append(tasks, t)
|
|
}
|
|
return tasks, nil
|
|
}
|
|
|
|
func (a *storageAdapter) Remove(task *task.Task) error {
|
|
ta, err := a.getTaskAttribute(task)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return a.taskStorage.Remove(ta)
|
|
}
|
|
|
|
func (a *storageAdapter) getTaskAttribute(task *task.Task) (*storage.TaskAttribute, error) {
|
|
params, err := paramsToString(task.Params)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
isRecurring := 0
|
|
if task.IsRecurring {
|
|
isRecurring = 1
|
|
}
|
|
|
|
return &storage.TaskAttribute{
|
|
Hash: string(task.Hash()),
|
|
Name: task.TaskTarget.Name,
|
|
LastRun: task.LastRun.Format(time.RFC3339),
|
|
NextRun: task.NextRun.Format(time.RFC3339),
|
|
Duration: task.Duration.String(),
|
|
IsRecurring: strconv.Itoa(isRecurring),
|
|
Params: params,
|
|
}, nil
|
|
}
|
|
|
|
func paramsToString(params []interface{}) (string, error) {
|
|
var paramsList []string
|
|
for _, param := range params {
|
|
paramStr, err := json.Marshal(param)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
paramsList = append(paramsList, string(paramStr))
|
|
}
|
|
data, err := json.Marshal(paramsList)
|
|
return string(data), err
|
|
}
|
|
|
|
func paramsFromString(taskTarget *task.TaskTarget, payload string) ([]interface{}, error) {
|
|
var params []interface{}
|
|
if strings.TrimSpace(payload) == "" {
|
|
return params, nil
|
|
}
|
|
paramTypes := taskTarget.Params()
|
|
var paramsStrings []string
|
|
err := json.Unmarshal([]byte(payload), ¶msStrings)
|
|
if err != nil {
|
|
return params, err
|
|
}
|
|
for i, paramStr := range paramsStrings {
|
|
paramType := paramTypes[i]
|
|
target := reflect.New(paramType)
|
|
err := json.Unmarshal([]byte(paramStr), target.Interface())
|
|
if err != nil {
|
|
return params, err
|
|
}
|
|
param := reflect.Indirect(target).Interface()
|
|
params = append(params, param)
|
|
}
|
|
|
|
return params, nil
|
|
}
|