Merge pull request #575 from trheyi/main

[add] pipe widget (dev)

engine/load.go

@@ -22,6 +22,7 @@ import (
 	"github.com/yaoapp/yao/model"
 	"github.com/yaoapp/yao/neo"
 	"github.com/yaoapp/yao/pack"
+	"github.com/yaoapp/yao/pipe"
 	"github.com/yaoapp/yao/plugin"
 	"github.com/yaoapp/yao/query"
 	"github.com/yaoapp/yao/runtime"
@@ -210,6 +211,12 @@ func Load(cfg config.Config) (err error) {
 		printErr(cfg.Mode, "Moapi", err)
 	}
 
+	// Load Pipe
+	err = pipe.Load(cfg)
+	if err != nil {
+		printErr(cfg.Mode, "Pipe", err)
+	}
+
 	return nil
 }
 

pipe/README.md

@@ -0,0 +1,5 @@
+# Pipe
+
+**Warning: This function under development and is not yet publicly available.**
+
+A new workflow orchestration engine designed to solve complex workflow orchestration problems. This is an alternative to **Flow**.

pipe/context.go

@@ -0,0 +1,76 @@
+package pipe
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/google/uuid"
+	"github.com/yaoapp/kun/exception"
+)
+
+var contexts = sync.Map{}
+
+// Create create new context
+func (pipe *Pipe) Create() *Context {
+	id := uuid.NewString()
+	ctx := &Context{Pipe: pipe, id: uuid.NewString()}
+	ctx.current = 0
+	contexts.Store(id, ctx)
+	return ctx
+}
+
+// Open the context
+func Open(id string) *Context {
+	ctx, ok := contexts.Load(id)
+	if !ok {
+		exception.New("pipe: %s not found", 404, id).Throw()
+	}
+	return ctx.(*Context)
+}
+
+// Close the context
+func Close(id string) {
+	contexts.Delete(id)
+}
+
+// Run the pipe
+func (ctx *Context) Run(args ...any) any {
+	v, err := ctx.Exec(args...)
+	if err != nil {
+		exception.New("pipe: %s %s", 500, ctx.Name, err).Throw()
+	}
+	return v
+}
+
+// ID the context id
+func (ctx *Context) ID() string {
+	return ctx.id
+}
+
+// Exec and return error
+func (ctx *Context) Exec(args ...any) (any, error) {
+	fmt.Printf("name: %v\n", ctx.Name)
+	fmt.Printf("global: %v\n", ctx.global)
+	fmt.Printf("sid: %v\n", ctx.sid)
+	fmt.Printf("whitelist: %v\n", ctx.Whitelist)
+	return nil, nil
+}
+
+// With with the context
+func (ctx *Context) With(context context.Context) *Context {
+	ctx.context = context
+	return ctx
+}
+
+// WithGlobal with the global data
+func (ctx *Context) WithGlobal(data map[string]interface{}) *Context {
+	ctx.global = data
+	return ctx
+}
+
+// WithSid with the sid
+func (ctx *Context) WithSid(sid string) *Context {
+	ctx.sid = sid
+	return ctx
+}

pipe/interface.go

@@ -0,0 +1 @@
+package pipe

pipe/json.go

@@ -0,0 +1,105 @@
+package pipe
+
+import (
+	"fmt"
+
+	jsoniter "github.com/json-iterator/go"
+)
+
+// UnmarshalJSON Custom JSON unmarshal function
+func (whitelist *Whitelist) UnmarshalJSON(data []byte) error {
+
+	var list any
+	err := jsoniter.Unmarshal(data, &list)
+	if err != nil {
+		return err
+	}
+
+	switch v := list.(type) {
+	case []string:
+		list := map[string]bool{}
+		for _, name := range v {
+			list[name] = true
+		}
+		*whitelist = list
+
+	case []interface{}:
+		list := map[string]bool{}
+		for _, name := range v {
+			list[fmt.Sprint(name)] = true
+		}
+		*whitelist = list
+
+	case map[string]interface{}:
+		list := map[string]bool{}
+		for name := range v {
+			list[name] = true
+		}
+		*whitelist = list
+
+	default:
+		return fmt.Errorf("whitelist type error: %#v", v)
+	}
+
+	return nil
+}
+
+// UnmarshalJSON Custom JSON unmarshal function
+func (input Input) UnmarshalJSON(data []byte) error {
+
+	var res any
+	err := jsoniter.Unmarshal(data, &res)
+	if err != nil {
+		return err
+	}
+
+	switch v := res.(type) {
+	case []string:
+		input = []any{}
+		for _, name := range v {
+			input = append(input, name)
+		}
+
+	case []interface{}:
+		input = v
+
+	case string:
+		input = []any{v}
+
+	default:
+		return fmt.Errorf("input type error: %#v", v)
+	}
+
+	return nil
+
+}
+
+// UnmarshalJSON Custom JSON unmarshal function
+func (args Args) UnmarshalJSON(data []byte) error {
+
+	var res any
+	err := jsoniter.Unmarshal(data, &res)
+	if err != nil {
+		return err
+	}
+
+	switch v := res.(type) {
+	case []string:
+		args = []any{}
+		for _, name := range v {
+			args = append(args, name)
+		}
+
+	case []interface{}:
+		args = v
+
+	case string:
+		args = []any{v}
+
+	default:
+		return fmt.Errorf("input type error: %#v", v)
+	}
+
+	return nil
+
+}

pipe/pipe.go

@@ -0,0 +1,87 @@
+package pipe
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/yaoapp/gou/application"
+	"github.com/yaoapp/yao/config"
+	"github.com/yaoapp/yao/share"
+)
+
+var pipes = map[string]*Pipe{}
+
+// Load the pipe
+func Load(cfg config.Config) error {
+
+	exts := []string{"*.pip.yao", "*.pipe.yao"}
+	errs := []error{}
+	err := application.App.Walk("pipes", func(root, file string, isdir bool) error {
+		if isdir {
+			return nil
+		}
+
+		id := share.ID(root, file)
+		pipe, err := NewFile(file, root)
+		if err != nil {
+			errs = append(errs, err)
+			return err
+		}
+
+		Set(id, pipe)
+		return err
+	}, exts...)
+
+	if len(errs) > 0 {
+		return errors.Join(errs...)
+	}
+
+	return err
+}
+
+// New create Pipe
+func New(source []byte) (*Pipe, error) {
+	pipe := Pipe{}
+	err := application.Parse("<source>", source, &pipe)
+	if err != nil {
+		return nil, err
+	}
+	return &pipe, nil
+}
+
+// NewFile create pipe from file
+func NewFile(file string, root string) (*Pipe, error) {
+	source, err := application.App.Read(file)
+	if err != nil {
+		return nil, err
+	}
+
+	id := share.ID(root, file)
+	pipe := Pipe{ID: id}
+	err = application.Parse(file, source, &pipe)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pipe, nil
+}
+
+// Set pipe to
+func Set(id string, pipe *Pipe) {
+	pipes[id] = pipe
+}
+
+// Remove the pipe
+func Remove(id string) {
+	if _, has := pipes[id]; has {
+		delete(pipes, id)
+	}
+}
+
+// Get the pipe
+func Get(id string) (*Pipe, error) {
+	if pipe, has := pipes[id]; has {
+		return pipe, nil
+	}
+	return nil, fmt.Errorf("pipe %s not found", id)
+}

pipe/pipe_test.go

@@ -0,0 +1,42 @@
+package pipe
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/yaoapp/gou/session"
+	"github.com/yaoapp/yao/config"
+	"github.com/yaoapp/yao/test"
+)
+
+func TestRun(t *testing.T) {
+	prepare(t)
+	defer test.Clean()
+	translator, err := Get("translator")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sid := session.ID()
+	context, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	ctx := translator.
+		Create().
+		With(context).
+		WithGlobal(map[string]interface{}{"foo": "bar"}).
+		WithSid(sid)
+	defer Close(ctx.ID())
+
+	assert.NotPanics(t, func() { ctx.Run() })
+}
+
+func prepare(t *testing.T) {
+	test.Prepare(t, config.Conf)
+	err := Load(config.Conf)
+	if err != nil {
+		t.Fatal(err)
+	}
+}

pipe/types.go

@@ -0,0 +1,77 @@
+package pipe
+
+import "context"
+
+// Pipe the pipe
+type Pipe struct {
+	ID        string
+	Name      string    `json:"name"`
+	Nodes     []Node    `json:"nodes"`
+	Label     string    `json:"label,omitempty"`
+	Hooks     *Hooks    `json:"hooks,omitempty"`
+	Output    any       `json:"output,omitempty"`    // $output
+	Input     Input     `json:"input,omitempty"`     // $input
+	Whitelist Whitelist `json:"whitelist,omitempty"` // the process whitelist
+
+}
+
+// Context the Context
+type Context struct {
+	*Pipe
+	id      string
+	context context.Context
+	global  map[string]interface{} // $global
+	sid     string                 // $sid
+	current int                    // current position
+}
+
+// Hooks the Hooks
+type Hooks struct {
+	Progress string `json:"progress,omitempty"`
+}
+
+// Node the pip node
+type Node struct {
+	Name      string                 `json:"name"`
+	Type      string                 `json:"type,omitempty"`      // user-input, ai, process, switch, request
+	Label     string                 `json:"label,omitempty"`     // Display
+	Process   *Process               `json:"process,omitempty"`   // Yao Process
+	Prompts   []Prompt               `json:"prompts,omitempty"`   // AI prompts
+	Request   *Request               `json:"request,omitempty"`   // Http Request
+	Interface string                 `json:"interface,omitempty"` // User Interface command-line, web, app, wxapp ...
+	Case      map[string]CaseSection `json:"case,omitempty"`      // Switch
+	Input     Input                  `json:"input,omitempty"`     // $in
+	Output    any                    `json:"output,omitempty"`    // $out
+}
+
+// Whitelist the Whitelist
+type Whitelist map[string]bool
+
+// Input the input
+type Input []any
+
+// Args the args
+type Args []any
+
+// CaseSection the switch case section
+type CaseSection struct {
+	Input  Input  `json:"input,omitempty"`  // $in
+	Output any    `json:"output,omitempty"` // $out
+	Nodes  []Node `json:"nodes,omitempty"`  // $out
+	Goto   string `json:"goto,omitempty"`   // goto node name / EOF
+}
+
+// Prompt the switch
+type Prompt struct {
+	Role    string `json:"role,omitempty"`
+	Message string `json:"message,omitempty"`
+}
+
+// Process the switch
+type Process struct {
+	Name string `json:"name"`
+	Args Args   `json:"args,omitempty"`
+}
+
+// Request the request
+type Request struct{}