pipe widget add process handlers

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 {
+func (ctx *Context) Resume(id string, args ...any) any {
-	ctx := Open(id)
 	v, err := ctx.resume(args...)
 	if err != nil {
 		exception.New("pipe: %s %s", 500, ctx.Name, err).Throw()

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

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 <pipe.id> [...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 <id> [...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 <id>
+func processClose(process *process.Process) interface{} {
+	process.ValidateArgNums(1)
+	id := process.ArgsString(0)
+	Close(id)
+	return nil
+}

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")
+}