From 3509956bce649717b2f5d58fd7ef9a57f4f18e3d Mon Sep 17 00:00:00 2001 From: Max Date: Sat, 30 Jul 2022 21:18:05 +0800 Subject: [PATCH] [add] Task & Schedule --- cmd/start.go | 2 +- engine/load.go | 12 +++++++ go.mod | 1 + go.sum | 3 ++ main_test.go | 1 + schedule/schedule.go | 36 +++++++++++++++++++++ schedule/schedule_test.go | 21 ++++++++++++ service/service.go | 24 +++++++------- task/task.go | 36 +++++++++++++++++++++ task/task_test.go | 19 +++++++++++ tests/schedules/mail.sch.json | 6 ++++ tests/schedules/sendmail.sch.json | 6 ++++ tests/scripts/mail.js | 53 +++++++++++++++++++++++++++++++ tests/tasks/mail.task.json | 16 ++++++++++ 14 files changed, 224 insertions(+), 12 deletions(-) create mode 100644 schedule/schedule.go create mode 100644 schedule/schedule_test.go create mode 100644 task/task.go create mode 100644 task/task_test.go create mode 100644 tests/schedules/mail.sch.json create mode 100644 tests/schedules/sendmail.sch.json create mode 100644 tests/scripts/mail.js create mode 100644 tests/tasks/mail.task.json diff --git a/cmd/start.go b/cmd/start.go index 33a73f1e..516d6362 100644 --- a/cmd/start.go +++ b/cmd/start.go @@ -24,7 +24,7 @@ var startCmd = &cobra.Command{ Short: L("Start Engine"), Long: L("Start Engine"), Run: func(cmd *cobra.Command, args []string) { - defer service.Stop(func() { fmt.Println(L("Service stopped")) }) + // defer service.Stop(func() { fmt.Println(L("Service stopped")) }) Boot() if startDebug { // 强制 debug 模式启动 diff --git a/engine/load.go b/engine/load.go index 00e9b871..7adb9167 100644 --- a/engine/load.go +++ b/engine/load.go @@ -19,11 +19,13 @@ import ( "github.com/yaoapp/yao/page" "github.com/yaoapp/yao/plugin" "github.com/yaoapp/yao/query" + "github.com/yaoapp/yao/schedule" "github.com/yaoapp/yao/script" "github.com/yaoapp/yao/share" "github.com/yaoapp/yao/socket" "github.com/yaoapp/yao/store" "github.com/yaoapp/yao/table" + "github.com/yaoapp/yao/task" "github.com/yaoapp/yao/websocket" ) @@ -121,6 +123,16 @@ func Load(cfg config.Config) (err error) { log.Debug(err.Error()) } + err = task.Load(cfg) // Load tasks + if err != nil { + log.Debug(err.Error()) + } + + err = schedule.Load(cfg) // Load schedules + if err != nil { + log.Debug(err.Error()) + } + return nil } diff --git a/go.mod b/go.mod index 344c8cd8..ac673cbd 100644 --- a/go.mod +++ b/go.mod @@ -76,6 +76,7 @@ require ( github.com/tidwall/rtred v0.1.2 // indirect github.com/tidwall/tinyqueue v0.1.1 // indirect github.com/ugorji/go/codec v1.1.7 // indirect + github.com/robfig/cron/v3 v3.0.1 // indirect github.com/xuri/efp v0.0.0-20210322160811-ab561f5b45e3 // indirect golang.org/x/image v0.0.0-20210628002857-a66eb6448b8d // indirect golang.org/x/mod v0.4.2 // indirect diff --git a/go.sum b/go.sum index bcd66b44..5f1c8c17 100644 --- a/go.sum +++ b/go.sum @@ -326,7 +326,10 @@ github.com/richardlehane/mscfb v1.0.3 h1:rD8TBkYWkObWO0oLDFCbwMeZ4KoalxQy+QgniCj github.com/richardlehane/mscfb v1.0.3/go.mod h1:YzVpcZg9czvAuhk9T+a3avCpcFPMUWm7gK3DypaEsUk= github.com/richardlehane/msoleps v1.0.1 h1:RfrALnSNXzmXLbGct/P2b4xkFz4e8Gmj/0Vj9M9xC1o= github.com/richardlehane/msoleps v1.0.1/go.mod h1:BWev5JBpU9Ko2WAgmZEuiz4/u3ZYTKbjLycmwiWUfWg= +github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ= github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k= +github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= +github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ= github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= diff --git a/main_test.go b/main_test.go index e5c2a25a..8a00a058 100644 --- a/main_test.go +++ b/main_test.go @@ -122,6 +122,7 @@ func TestCommandStop(t *testing.T) { times++ res, err := request() if err != nil { + fmt.Println("REQUEST ERROR:", err) continue } assert.Equal(t, 1, any.Of(res.Get("id")).CInt()) diff --git a/schedule/schedule.go b/schedule/schedule.go new file mode 100644 index 00000000..70fe869e --- /dev/null +++ b/schedule/schedule.go @@ -0,0 +1,36 @@ +package schedule + +import ( + "fmt" + "path/filepath" + + "github.com/yaoapp/gou" + "github.com/yaoapp/kun/log" + "github.com/yaoapp/yao/config" + "github.com/yaoapp/yao/share" +) + +// Load load schedule +func Load(cfg config.Config) error { + var root = filepath.Join(cfg.Root, "schedules") + return LoadFrom(root, "") +} + +// LoadFrom load from dir +func LoadFrom(dir string, prefix string) error { + + if share.DirNotExists(dir) { + return fmt.Errorf("%s does not exists", dir) + } + + err := share.Walk(dir, ".json", func(root, filename string) { + name := prefix + share.SpecName(root, filename) + content := share.ReadFile(filename) + _, err := gou.LoadSchedule(string(content), name) + if err != nil { + log.With(log.F{"root": root, "file": filename}).Error(err.Error()) + } + }) + + return err +} diff --git a/schedule/schedule_test.go b/schedule/schedule_test.go new file mode 100644 index 00000000..d0f27e3e --- /dev/null +++ b/schedule/schedule_test.go @@ -0,0 +1,21 @@ +package schedule + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/yaoapp/gou" + "github.com/yaoapp/yao/config" + "github.com/yaoapp/yao/task" +) + +func TestLoad(t *testing.T) { + task.Load(config.Conf) + Load(config.Conf) + LoadFrom("not a path", "404.") + check(t) +} + +func check(t *testing.T) { + assert.Equal(t, 2, len(gou.Schedules)) +} diff --git a/service/service.go b/service/service.go index 31044161..5da3ca6d 100644 --- a/service/service.go +++ b/service/service.go @@ -6,8 +6,8 @@ import ( "github.com/yaoapp/yao/share" ) -var shutdown = make(chan bool) -var shutdownComplete = make(chan bool) +var shutdown = make(chan bool, 1) +var shutdownComplete = make(chan bool, 1) // Start 启动服务 func Start() error { @@ -25,7 +25,7 @@ func Start() error { Root: "/api", Allows: config.Conf.AllowFrom, }, - &shutdown, func(s gou.Server) { + shutdown, func(s gou.Server) { shutdownComplete <- true }, Middlewares...) @@ -44,7 +44,7 @@ func StartWithouttSession() { Root: "/api", Allows: config.Conf.AllowFrom, }, - &shutdown, func(s gou.Server) { + shutdown, func(s gou.Server) { shutdownComplete <- true }, Middlewares...) @@ -53,16 +53,18 @@ func StartWithouttSession() { // StopWithouttSession 关闭服务 func StopWithouttSession(onComplete func()) { shutdown <- true - <-shutdownComplete - gou.KillPlugins() - onComplete() + select { + case <-shutdownComplete: + onComplete() + } } // Stop 关闭服务 func Stop(onComplete func()) { shutdown <- true - <-shutdownComplete - share.SessionStop() - gou.KillPlugins() - onComplete() + select { + case <-shutdownComplete: + share.SessionStop() + onComplete() + } } diff --git a/task/task.go b/task/task.go new file mode 100644 index 00000000..07d36e7e --- /dev/null +++ b/task/task.go @@ -0,0 +1,36 @@ +package task + +import ( + "fmt" + "path/filepath" + + "github.com/yaoapp/gou" + "github.com/yaoapp/kun/log" + "github.com/yaoapp/yao/config" + "github.com/yaoapp/yao/share" +) + +// Load load task +func Load(cfg config.Config) error { + var root = filepath.Join(cfg.Root, "tasks") + return LoadFrom(root, "") +} + +// LoadFrom load from dir +func LoadFrom(dir string, prefix string) error { + + if share.DirNotExists(dir) { + return fmt.Errorf("%s does not exists", dir) + } + + err := share.Walk(dir, ".json", func(root, filename string) { + name := prefix + share.SpecName(root, filename) + content := share.ReadFile(filename) + _, err := gou.LoadTask(string(content), name) + if err != nil { + log.With(log.F{"root": root, "file": filename}).Error(err.Error()) + } + }) + + return err +} diff --git a/task/task_test.go b/task/task_test.go new file mode 100644 index 00000000..71f178a4 --- /dev/null +++ b/task/task_test.go @@ -0,0 +1,19 @@ +package task + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/yaoapp/gou/task" + "github.com/yaoapp/yao/config" +) + +func TestLoad(t *testing.T) { + Load(config.Conf) + LoadFrom("not a path", "404.") + check(t) +} + +func check(t *testing.T) { + assert.Equal(t, 1, len(task.Tasks)) +} diff --git a/tests/schedules/mail.sch.json b/tests/schedules/mail.sch.json new file mode 100644 index 00000000..fc9c1084 --- /dev/null +++ b/tests/schedules/mail.sch.json @@ -0,0 +1,6 @@ +{ + "name": "每分钟发送一封邮件", + "schedule": "*/1 * * * *", + "process": "scripts.mail.Send", + "args": [null, "$ENV.SEND_MAIL_TEST_MAIL"] +} diff --git a/tests/schedules/sendmail.sch.json b/tests/schedules/sendmail.sch.json new file mode 100644 index 00000000..b6d6e0bc --- /dev/null +++ b/tests/schedules/sendmail.sch.json @@ -0,0 +1,6 @@ +{ + "name": "每分钟发送一封邮件", + "schedule": "*/1 * * * *", + "task": "mail", + "args": ["$ENV.SEND_MAIL_TEST_MAIL"] +} diff --git a/tests/scripts/mail.js b/tests/scripts/mail.js new file mode 100644 index 00000000..978a5442 --- /dev/null +++ b/tests/scripts/mail.js @@ -0,0 +1,53 @@ +var id = 1024; + +/** + * Generate job id + * @returns + */ +function NextID() { + id = id + 1; + console.log(`NextID: ${id}`); + return id; +} + +function Send(id, mail, flag) { + for (var i = 1; i <= 3; i++) { + Process("xiang.system.Sleep", 200); + Process("tasks.mail.Progress", id, i, 3, "unit-test"); + } + if (flag) { + console.log(`flag: ${flag}`); + Process("xiang.system.Sleep", 2000); + } + console.log( + `Send: ${JSON.stringify({ foo: "bar", mail: mail, flag: flag || "-" })}` + ); + return { foo: "bar", mail: mail, flag: flag || "-" }; +} + +/** + * OnAdd add event + * @param {*} id + */ +function OnAdd(id) { + console.log(`OnAdd: #${id}`); +} + +/** + * OnProgress + * @param {*} id + * @param {*} current + * @param {*} total + * @param {*} message + */ +function OnProgress(id, current, total, message) { + console.log(`OnProgress: #${id} ${message} ${current}/${total} `); +} + +function OnSuccess(id, res) { + console.log(`OnSuccess: #${id} ${JSON.stringify(res)}`); +} + +function OnError(id, err) { + console.log(`OnError: #${id} ${err}`); +} diff --git a/tests/tasks/mail.task.json b/tests/tasks/mail.task.json new file mode 100644 index 00000000..5bf7bafd --- /dev/null +++ b/tests/tasks/mail.task.json @@ -0,0 +1,16 @@ +{ + "name": "发送邮件", + "worker_nums": "$ENV.SEND_MAIL_WORKER_NUMS", + "attempts": 3, + "attempt_after": 200, + "timeout": 2, + "size": 1000, + "process": "scripts.mail.Send", + "event": { + "next": "scripts.mail.NextID", + "add": "scripts.mail.OnAdd", + "success": "scripts.mail.OnSuccess", + "error": "scripts.mail.OnError", + "progress": "scripts.mail.OnProgress" + } +}