From 6b90d097e26078b1abecefd1420d5634e334d932 Mon Sep 17 00:00:00 2001 From: Max Date: Sat, 17 Feb 2024 21:12:28 +0800 Subject: [PATCH] Update pipe widget with new processes --- pipe/README.md | 20 +++++++++-- pipe/process.go | 83 +++++++++++++++++++++++++++++++++++++++++--- pipe/process_test.go | 63 +++++++++++++++++++++++++++++++++ 3 files changed, 160 insertions(+), 6 deletions(-) diff --git a/pipe/README.md b/pipe/README.md index 48df8b86..1db33817 100644 --- a/pipe/README.md +++ b/pipe/README.md @@ -47,7 +47,7 @@ If interrupted by user input interface, it returns a context ID for resuming exe Run Pipe, equivalent to `pipes.` ```bash -yao run pipe.run [args...] +yao run pipe.Run [args...] ``` ### pipe.Create @@ -55,7 +55,15 @@ yao run pipe.run [args...] Pass DSL text to create and run Pipe ```bash -yao run pipe.create [args...] +yao run pipe.Create [args...] +``` + +### pipe.CreateWith + +Pass DSL text to create and run Pipe + +```bash +yao run pipe.CreateWith '::{"foo":"bar"}' [args...] ``` ### pipe.Resume @@ -66,6 +74,14 @@ Resume execution, used for context restoration yao run pipe.Resume [args...] ``` +### pipe.ResumeWith + +Resume execution, used for context restoration + +```bash +yao run pipe.ResumeWith '::{"foo":"bar"}' [args...] +``` + ### pipe.Close Close Pipe diff --git a/pipe/process.go b/pipe/process.go index 38afd065..e1e7ff7d 100644 --- a/pipe/process.go +++ b/pipe/process.go @@ -8,10 +8,12 @@ import ( func init() { process.Register("pipes", processPipes) process.RegisterGroup("pipe", map[string]process.Handler{ - "run": processRun, - "create": processCreate, - "resume": processResume, - "close": processClose, + "run": processRun, + "create": processCreate, + "createwith": processCreateWith, // create with global data + "resume": processResume, + "resumewith": processResumeWith, // resume with global data + "close": processClose, }) } @@ -45,6 +47,41 @@ func processCreate(process *process.Process) interface{} { return ctx.Run(args...) } +// processCreateWith process the create pipe.createWith , [...args] +func processCreateWith(process *process.Process) interface{} { + process.ValidateArgNums(2) + dsl := process.ArgsString(0) + data := process.ArgsMap(1, map[string]any{}) + args := []any{} + if len(process.Args) > 2 { + args = process.Args[2:] + } + + // merge the global data + if process.Global != nil { + merge := map[string]any{} + global := process.Global + for k, v := range global { + merge[k] = v + } + + if data != nil { + for k, v := range data { + merge[k] = v + } + } + data = merge + } + + pipe, err := New([]byte(dsl)) + if err != nil { + exception.New(err.Error(), 500).Throw() + } + + ctx := pipe.Create().WithGlobal(data).WithSid(process.Sid) + return ctx.Run(args...) +} + // processRun process the resume pipe.run [...args] func processRun(process *process.Process) interface{} { process.ValidateArgNums(1) @@ -82,6 +119,44 @@ func processResume(process *process.Process) interface{} { Resume(id, args...) } +// processResumeWith process the resume pipe.resumeWith , [...args] +func processResumeWith(process *process.Process) interface{} { + process.ValidateArgNums(2) + id := process.ArgsString(0) + data := process.ArgsMap(1, map[string]any{}) + + args := []any{} + if len(process.Args) > 2 { + args = process.Args[2:] + } + + ctx, err := Open(id) + if err != nil { + exception.New("pipes.%s not found", 404, id).Throw() + } + + // merge the global data + if process.Global != nil { + merge := map[string]any{} + global := process.Global + for k, v := range global { + merge[k] = v + } + + if data != nil { + for k, v := range data { + merge[k] = v + } + } + data = merge + } + + return ctx. + WithGlobal(data). + WithSid(process.Sid). + Resume(id, args...) +} + // processClose process the close pipe.close func processClose(process *process.Process) interface{} { process.ValidateArgNums(1) diff --git a/pipe/process_test.go b/pipe/process_test.go index 424ee30f..dff4ceea 100644 --- a/pipe/process_test.go +++ b/pipe/process_test.go @@ -92,6 +92,40 @@ func TestProcessCreate(t *testing.T) { assert.Equal(t, "hello world", res.Get("input[0]")) } +func TestProcessCreateWith(t *testing.T) { + prepare(t) + defer test.Clean() + + dsl := `{ + "whitelist": ["utils.fmt.Print"], + "name": "test", + "label": "Test", + "input": "{{ $global.placeholder }}", + "nodes": [ + { + "name": "print", + "process": {"name":"utils.fmt.Print", "args": "{{ $in }}"}, + "output": "print" + } + ], + "output": {"input": "{{ $input }}" } + }` + + p, err := process.Of("pipe.CreateWith", dsl, map[string]interface{}{"placeholder": "hello 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.Equal(t, "hello world", res.Get("input[0]")) + +} + func TestProcessResume(t *testing.T) { prepare(t) defer test.Clean() @@ -120,6 +154,35 @@ func TestProcessResume(t *testing.T) { assert.Len(t, res.Get("switch"), 2) } +func TestProcessResumeWith(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.ResumeWith", resume.ID, map[string]interface{}{"foo": "bar"}, "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.Equal(t, "bar", res.Get("global.foo")) + assert.Len(t, res.Get("switch"), 2) +} + func TestProcessClose(t *testing.T) { prepare(t) defer test.Clean()