opencode/yao
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[add] Task & Schedule

Browse source
This commit is contained in:
Max 2022-07-30 21:18:05 +08:00
parent 5a92e4f7b5
commit 3509956bce
14 changed files with 224 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cmd/start.go
View file

@ -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 模式启动

12
engine/load.go
View file

@ -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
}

1
go.mod
View file

@ -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

3
go.sum
View file

@ -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=

1
main_test.go
View file

@ -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())

36
schedule/schedule.go Normal file
View file

@ -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
}

21
schedule/schedule_test.go Normal file
View file

@ -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))
}

24
service/service.go
View file

@ -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()
}
}

36
task/task.go Normal file
View file

@ -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
}

19
task/task_test.go Normal file
View file

@ -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))
}

6
tests/schedules/mail.sch.json Normal file
View file

@ -0,0 +1,6 @@
{
"name": "每分钟发送一封邮件",
"schedule": "*/1 * * * *",
"process": "scripts.mail.Send",
"args": [null, "$ENV.SEND_MAIL_TEST_MAIL"]
}

6
tests/schedules/sendmail.sch.json Normal file
View file

@ -0,0 +1,6 @@
{
"name": "每分钟发送一封邮件",
"schedule": "*/1 * * * *",
"task": "mail",
"args": ["$ENV.SEND_MAIL_TEST_MAIL"]
}

53
tests/scripts/mail.js Normal file
View file

@ -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}`);
}

16
tests/tasks/mail.task.json Normal file
View file

@ -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"
}
}