Merge pull request #580 from trheyi/main

Update pipe widget with new processes

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.<Widget.ID>`
 
 ```bash
-yao run pipe.run <Widget.ID> [args...]
+yao run pipe.Run <Widget.ID> [args...]
 ```
 
 ### pipe.Create
@@ -55,7 +55,15 @@ yao run pipe.run <Widget.ID> [args...]
 Pass DSL text to create and run Pipe
 
 ```bash
-yao run pipe.create <DSL> [args...]
+yao run pipe.Create <DSL> [args...]
+```
+
+### pipe.CreateWith
+
+Pass DSL text to create and run Pipe
+
+```bash
+yao run pipe.CreateWith <DSL> '::{"foo":"bar"}' [args...]
 ```
 
 ### pipe.Resume
@@ -66,6 +74,14 @@ Resume execution, used for context restoration
 yao run pipe.Resume <Context.ID> [args...]
 ```
 
+### pipe.ResumeWith
+
+Resume execution, used for context restoration
+
+```bash
+yao run pipe.ResumeWith <Context.ID> '::{"foo":"bar"}' [args...]
+```
+
 ### pipe.Close
 
 Close Pipe

pipe/process.go

@@ -10,7 +10,9 @@ func init() {
 	process.RegisterGroup("pipe", map[string]process.Handler{
 		"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 <pipe.id> <global>, [...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 <pipe.id> [...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 <id> <global>, [...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 <id>
 func processClose(process *process.Process) interface{} {
 	process.ValidateArgNums(1)

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