From ae599ec2e79f62b82e84d3326db3f352320a9298 Mon Sep 17 00:00:00 2001 From: Max Date: Mon, 12 Feb 2024 21:08:31 +0800 Subject: [PATCH] [add] pipe widget (dev) --- engine/load.go | 7 ++++ pipe/README.md | 5 +++ pipe/context.go | 76 +++++++++++++++++++++++++++++++++ pipe/interface.go | 1 + pipe/json.go | 105 ++++++++++++++++++++++++++++++++++++++++++++++ pipe/pipe.go | 87 ++++++++++++++++++++++++++++++++++++++ pipe/pipe_test.go | 42 +++++++++++++++++++ pipe/types.go | 77 ++++++++++++++++++++++++++++++++++ 8 files changed, 400 insertions(+) create mode 100644 pipe/README.md create mode 100644 pipe/context.go create mode 100644 pipe/interface.go create mode 100644 pipe/json.go create mode 100644 pipe/pipe.go create mode 100644 pipe/pipe_test.go create mode 100644 pipe/types.go diff --git a/engine/load.go b/engine/load.go index 134c3631..83cbd0b4 100644 --- a/engine/load.go +++ b/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 } diff --git a/pipe/README.md b/pipe/README.md new file mode 100644 index 00000000..7ff97648 --- /dev/null +++ b/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**. diff --git a/pipe/context.go b/pipe/context.go new file mode 100644 index 00000000..8544c59c --- /dev/null +++ b/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 +} diff --git a/pipe/interface.go b/pipe/interface.go new file mode 100644 index 00000000..a7bad566 --- /dev/null +++ b/pipe/interface.go @@ -0,0 +1 @@ +package pipe diff --git a/pipe/json.go b/pipe/json.go new file mode 100644 index 00000000..777c6836 --- /dev/null +++ b/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 + +} diff --git a/pipe/pipe.go b/pipe/pipe.go new file mode 100644 index 00000000..0aae8971 --- /dev/null +++ b/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, &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) +} diff --git a/pipe/pipe_test.go b/pipe/pipe_test.go new file mode 100644 index 00000000..7bcf1ce8 --- /dev/null +++ b/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) + } +} diff --git a/pipe/types.go b/pipe/types.go new file mode 100644 index 00000000..1254666a --- /dev/null +++ b/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{}