add taskLodVersions
This commit is contained in:
parent
dea3d4616a
commit
cdd8c9b03f
|
@ -30,7 +30,11 @@ func Attach(c *cron.Cron, cfg Config) error {
|
||||||
return fmt.Errorf("cfg.Queue cannot be nil, expected queue.Queue")
|
return fmt.Errorf("cfg.Queue cannot be nil, expected queue.Queue")
|
||||||
}
|
}
|
||||||
|
|
||||||
h := &handler{db: cfg.DB, maxConcurrentWorkers: cfg.MaxConcurrentWorkers}
|
h := &handler{
|
||||||
|
db: cfg.DB,
|
||||||
|
maxConcurrentWorkers: cfg.MaxConcurrentWorkers,
|
||||||
|
queue: cfg.Queue,
|
||||||
|
}
|
||||||
if err := h.init(); err != nil {
|
if err := h.init(); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
|
@ -68,7 +68,7 @@ func (t *taskLoadServers) execute(version *models.Version) error {
|
||||||
|
|
||||||
if _, err := t.db.Model(&models.Server{}).
|
if _, err := t.db.Model(&models.Server{}).
|
||||||
Set("status = ?", models.ServerStatusClosed).
|
Set("status = ?", models.ServerStatusClosed).
|
||||||
Where("key NOT IN (?) AND version_code =", pg.In(serverKeys), version.Code).
|
Where("key NOT IN (?) AND version_code = ?", pg.In(serverKeys), version.Code).
|
||||||
Update(); err != nil {
|
Update(); err != nil {
|
||||||
err = errors.Wrap(err, "couldn't update server statuses")
|
err = errors.Wrap(err, "couldn't update server statuses")
|
||||||
logrus.Fatal(err)
|
logrus.Fatal(err)
|
||||||
|
@ -92,6 +92,7 @@ func (t *taskLoadServers) getServers(version *models.Version) (map[string]string
|
||||||
return nil, errors.Wrapf(err, "%s: taskLoadServers.loadServers couldn't load servers", version.Host)
|
return nil, errors.Wrapf(err, "%s: taskLoadServers.loadServers couldn't load servers", version.Host)
|
||||||
}
|
}
|
||||||
defer resp.Body.Close()
|
defer resp.Body.Close()
|
||||||
|
|
||||||
bodyBytes, err := ioutil.ReadAll(resp.Body)
|
bodyBytes, err := ioutil.ReadAll(resp.Body)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, errors.Wrapf(err, "%s: taskLoadServers.loadServers couldn't read the body", version.Host)
|
return nil, errors.Wrapf(err, "%s: taskLoadServers.loadServers couldn't read the body", version.Host)
|
||||||
|
@ -100,5 +101,14 @@ func (t *taskLoadServers) getServers(version *models.Version) (map[string]string
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, errors.Wrapf(err, "%s: taskLoadServers.loadServers couldn't decode the body into the go value", version.Host)
|
return nil, errors.Wrapf(err, "%s: taskLoadServers.loadServers couldn't decode the body into the go value", version.Host)
|
||||||
}
|
}
|
||||||
return body.(map[string]string), nil
|
|
||||||
|
result := make(map[string]string)
|
||||||
|
for serverKey, url := range body.(map[interface{}]interface{}) {
|
||||||
|
serverKeyStr := serverKey.(string)
|
||||||
|
urlStr := url.(string)
|
||||||
|
if serverKeyStr != "" && urlStr != "" {
|
||||||
|
result[serverKeyStr] = urlStr
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return result, nil
|
||||||
}
|
}
|
||||||
|
|
28
cron/tasks/task_load_versions.go
Normal file
28
cron/tasks/task_load_versions.go
Normal file
|
@ -0,0 +1,28 @@
|
||||||
|
package tasks
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"github.com/pkg/errors"
|
||||||
|
"github.com/tribalwarshelp/shared/models"
|
||||||
|
|
||||||
|
"github.com/tribalwarshelp/cron/cron/queue"
|
||||||
|
)
|
||||||
|
|
||||||
|
type taskLoadVersions struct {
|
||||||
|
*task
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *taskLoadVersions) execute() error {
|
||||||
|
var versions []*models.Version
|
||||||
|
log.Debug("taskLoadVersions.execute: Loading versions...")
|
||||||
|
if err := t.db.Model(&versions).Relation("SpecialServers").Select(); err != nil {
|
||||||
|
err = errors.Wrap(err, "taskLoadVersions.execute: couldn't load versions")
|
||||||
|
log.Fatal(err)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
for _, version := range versions {
|
||||||
|
t.queue.Add(queue.MainQueue, Get(TaskNameLoadServers).WithArgs(context.Background(), version))
|
||||||
|
}
|
||||||
|
log.Debug("taskLoadVersions.execute: Versions have been loaded")
|
||||||
|
return nil
|
||||||
|
}
|
|
@ -13,7 +13,7 @@ const (
|
||||||
TaskNameLoadVersions = "loadVersions"
|
TaskNameLoadVersions = "loadVersions"
|
||||||
TaskNameLoadServers = "loadServers"
|
TaskNameLoadServers = "loadServers"
|
||||||
TaskNameUpdateServerData = "updateServerData"
|
TaskNameUpdateServerData = "updateServerData"
|
||||||
retryLimitLoadServers = 3
|
defaultRetryLimit = 3
|
||||||
)
|
)
|
||||||
|
|
||||||
var log = logrus.WithField("package", "tasks")
|
var log = logrus.WithField("package", "tasks")
|
||||||
|
@ -32,9 +32,14 @@ func RegisterTasks(cfg *Config) error {
|
||||||
db: cfg.DB,
|
db: cfg.DB,
|
||||||
queue: cfg.Queue,
|
queue: cfg.Queue,
|
||||||
}
|
}
|
||||||
|
taskq.RegisterTask(&taskq.TaskOptions{
|
||||||
|
Name: TaskNameLoadVersions,
|
||||||
|
RetryLimit: defaultRetryLimit,
|
||||||
|
Handler: (&taskLoadVersions{t}).execute,
|
||||||
|
})
|
||||||
taskq.RegisterTask(&taskq.TaskOptions{
|
taskq.RegisterTask(&taskq.TaskOptions{
|
||||||
Name: TaskNameLoadServers,
|
Name: TaskNameLoadServers,
|
||||||
RetryLimit: retryLimitLoadServers,
|
RetryLimit: defaultRetryLimit,
|
||||||
Handler: (&taskLoadServers{t}).execute,
|
Handler: (&taskLoadServers{t}).execute,
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
4
main.go
4
main.go
|
@ -57,9 +57,7 @@ func main() {
|
||||||
DB: conn,
|
DB: conn,
|
||||||
Queue: queue,
|
Queue: queue,
|
||||||
})
|
})
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
if err := queue.Start(context.Background()); err != nil {
|
||||||
defer cancel()
|
|
||||||
if err := queue.Start(ctx); err != nil {
|
|
||||||
logrus.Fatal(err)
|
logrus.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Reference in New Issue
Block a user