From 24bf9c05a0c76032a6dacaed8179ac3aa6222c45 Mon Sep 17 00:00:00 2001 From: Max Date: Thu, 15 Feb 2024 10:48:44 +0800 Subject: [PATCH] pipe widget add process handlers --- pipe/context.go | 9 ++-- pipe/pipe_test.go | 6 ++- pipe/process.go | 50 +++++++++++++++++++++ pipe/process_test.go | 103 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 162 insertions(+), 6 deletions(-) create mode 100644 pipe/process_test.go diff --git a/pipe/context.go b/pipe/context.go index 124255f2..ecf1a308 100644 --- a/pipe/context.go +++ b/pipe/context.go @@ -36,12 +36,12 @@ func (pipe *Pipe) Create() *Context { } // Open the context -func Open(id string) *Context { +func Open(id string) (*Context, error) { ctx, ok := contexts.Load(id) if !ok { - exception.New("pipe: %s not found", 404, id).Throw() + return nil, fmt.Errorf("context %s not found", id) } - return ctx.(*Context) + return ctx.(*Context), nil } // Close the context @@ -50,8 +50,7 @@ func Close(id string) { } // Resume the context by id -func Resume(id string, args ...any) any { - ctx := Open(id) +func (ctx *Context) Resume(id string, args ...any) any { v, err := ctx.resume(args...) if err != nil { exception.New("pipe: %s %s", 500, ctx.Name, err).Throw() diff --git a/pipe/pipe_test.go b/pipe/pipe_test.go index 182fd8ce..90cc5e38 100644 --- a/pipe/pipe_test.go +++ b/pipe/pipe_test.go @@ -70,7 +70,11 @@ func TestRunWeb(t *testing.T) { resume := web.(ResumeContext) assert.Equal(t, Input{"hello web world"}, resume.Input) - output := Resume(resume.ID, "translate", "hello web world") + ctx, err = Open(resume.ID) + if err != nil { + t.Fatal(err) + } + output := ctx.Resume(resume.ID, "translate", "hello web world") res := any.Of(output).Map().MapStrAny.Dot() assert.True(t, res.Has("global")) diff --git a/pipe/process.go b/pipe/process.go index b088ac5f..c691465f 100644 --- a/pipe/process.go +++ b/pipe/process.go @@ -7,6 +7,11 @@ import ( func init() { process.Register("pipes", processPipes) + process.RegisterGroup("pipe", map[string]process.Handler{ + "run": processRun, + "resume": processResume, + "close": processClose, + }) } // processScripts @@ -26,3 +31,48 @@ func processPipes(process *process.Process) interface{} { return res } + +// processRun process the resume pipe.run [...args] +func processRun(process *process.Process) interface{} { + process.ValidateArgNums(1) + pid := process.ArgsString(0) + args := []any{} + if len(process.Args) > 1 { + args = process.Args[1:] + } + pipe, err := Get(pid) + if err != nil { + exception.New("pipes.%s not loaded", 404, process.ID).Throw() + } + + ctx := pipe.Create().WithGlobal(process.Global).WithSid(process.Sid) + return ctx.Run(args...) +} + +// processResume process the resume pipe.resume [...args] +func processResume(process *process.Process) interface{} { + process.ValidateArgNums(1) + id := process.ArgsString(0) + args := []any{} + if len(process.Args) > 1 { + args = process.Args[1:] + } + + ctx, err := Open(id) + if err != nil { + exception.New("pipes.%s not found", 404, id).Throw() + } + + return ctx. + WithGlobal(process.Global). + WithSid(process.Sid). + Resume(id, args...) +} + +// processClose process the close pipe.close +func processClose(process *process.Process) interface{} { + process.ValidateArgNums(1) + id := process.ArgsString(0) + Close(id) + return nil +} diff --git a/pipe/process_test.go b/pipe/process_test.go new file mode 100644 index 00000000..563cddc1 --- /dev/null +++ b/pipe/process_test.go @@ -0,0 +1,103 @@ +package pipe + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/yaoapp/gou/process" + "github.com/yaoapp/kun/any" + "github.com/yaoapp/yao/test" +) + +func TestProcessPipes(t *testing.T) { + prepare(t) + defer test.Clean() + + p, err := process.Of("pipes.cli.translator", map[string]interface{}{"placeholder": "translate\nhello world"}) + if err != nil { + t.Fatal(err) + } + + output, err := p.Exec() + res := any.Of(output).Map().MapStrAny.Dot() + assert.True(t, res.Has("global")) + assert.True(t, res.Has("input")) + assert.True(t, res.Has("output")) + assert.True(t, res.Has("sid")) + assert.True(t, res.Has("switch")) + assert.Equal(t, "translate\nhello world", res.Get("input[0].placeholder")) + assert.Len(t, res.Get("switch"), 2) +} + +func TestProcessRun(t *testing.T) { + prepare(t) + defer test.Clean() + + p, err := process.Of("pipe.Run", "cli.translator", map[string]interface{}{"placeholder": "translate\nhello world"}) + if err != nil { + t.Fatal(err) + } + + output, err := p.Exec() + if err != nil { + t.Fatal(err) + } + + res := any.Of(output).Map().MapStrAny.Dot() + assert.True(t, res.Has("global")) + assert.True(t, res.Has("input")) + assert.True(t, res.Has("output")) + assert.True(t, res.Has("sid")) + assert.True(t, res.Has("switch")) + assert.Equal(t, "translate\nhello world", res.Get("input[0].placeholder")) + assert.Len(t, res.Get("switch"), 2) +} + +func TestProcessResume(t *testing.T) { + prepare(t) + defer test.Clean() + + p, err := process.Of("pipe.Run", "web.translator", "hello web world") + if err != nil { + t.Fatal(err) + } + + web, err := p.Exec() + resume := web.(ResumeContext) + + p, err = process.Of("pipe.Resume", resume.ID, "translate", "hello web world") + output, err := p.Exec() + if err != nil { + t.Fatal(err) + } + + res := any.Of(output).Map().MapStrAny.Dot() + assert.True(t, res.Has("global")) + assert.True(t, res.Has("input")) + assert.True(t, res.Has("output")) + assert.True(t, res.Has("sid")) + assert.True(t, res.Has("switch")) + assert.Equal(t, "hello web world", res.Get("input[0]")) + assert.Len(t, res.Get("switch"), 2) +} + +func TestProcessClose(t *testing.T) { + prepare(t) + defer test.Clean() + + p, err := process.Of("pipe.Run", "web.translator", "hello web world") + if err != nil { + t.Fatal(err) + } + + web, err := p.Exec() + resume := web.(ResumeContext) + + p, err = process.Of("pipe.Close", resume.ID) + p.Exec() + + p, err = process.Of("pipe.Resume", resume.ID, "translate", "hello web world") + _, err = p.Exec() + assert.NotNil(t, err) + assert.Contains(t, err.Error(), "not found") +}