first commit

README.md

@@ -0,0 +1,3 @@
+# A simple golang workflow framework
+
+example codes at ./example directory.

config.go

@@ -0,0 +1,7 @@
+package goflow
+
+type Config struct {
+	Node     string   `json:"node,omitempty"`
+	Parallel []Config `json:"parallel,omitempty"`
+	Sequence []Config `json:"sequence,omitempty"`
+}

example/hello.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+
+	"github.com/zwkno1/goflow"
+)
+
+type Hello struct {
+	name string
+}
+
+func (h *Hello) Go(ctx context.Context) error {
+	fmt.Println("Hello " + h.name + "!")
+	return nil
+}
+
+type Factory struct {
+}
+
+func (f *Factory) Create(name string) (goflow.Node, error) {
+	return &Hello{
+		name: name,
+	}, nil
+}
+
+func main() {
+	data, err := ioutil.ReadFile(os.Args[1])
+	if err != nil {
+		fmt.Println("read config error: " + err.Error())
+		return
+	}
+	config := &goflow.Config{}
+	err = json.Unmarshal(data, config)
+	if err != nil {
+		fmt.Println("unmarshal config error: " + err.Error())
+		return
+	}
+
+	f, err := goflow.NewFlow(config, &Factory{})
+	if err != nil {
+		fmt.Println("create goflow error: " + err.Error())
+		return
+	}
+	if err := f.Go(nil); err != nil {
+		fmt.Println("go error: " + err.Error())
+	}
+}

example/mix.json

@@ -0,0 +1,23 @@
+{
+  "sequence": [
+    {
+      "node": "node1"
+    },
+    {
+      "parallel": [
+        {
+          "node": "node2-1"
+        },
+        {
+          "node": "node2-2"
+        },
+        {
+          "node": "node2-3"
+        }
+      ]
+    },
+    {
+      "node": "node3"
+    }
+  ]
+}

example/parallel.json

@@ -0,0 +1,13 @@
+{
+  "parallel": [
+    {
+      "node": "node1-1"
+    },
+    {
+      "node": "node1-2"
+    },
+    {
+      "node": "node1-3"
+    }
+  ]
+}

example/sequence.json

@@ -0,0 +1,13 @@
+{
+  "sequence": [
+    {
+      "node": "node1"
+    },
+    {
+      "node": "node2"
+    },
+    {
+      "node": "node3"
+    }
+  ]
+}

factory.go

@@ -0,0 +1,5 @@
+package goflow
+
+type Factory interface {
+	Create(name string) (Node, error)
+}

flow.go

@@ -0,0 +1,52 @@
+package goflow
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+)
+
+type Flow struct {
+	entrypoint Node
+}
+
+func NewFlow(config *Config, factory Factory) (*Flow, error) {
+	node, err := createNode(config, factory)
+	if err != nil {
+		return nil, err
+	}
+	return &Flow{entrypoint: node}, nil
+}
+
+func (f *Flow) Go(ctx context.Context) error {
+	return f.entrypoint.Go(ctx)
+}
+
+func createNode(config *Config, factory Factory) (Node, error) {
+	switch {
+	case len(config.Node) != 0:
+		return factory.Create(config.Node)
+	case len(config.Parallel) != 0 || len(config.Sequence) != 0:
+		configs := config.Parallel
+		if len(configs) == 0 {
+			configs = config.Sequence
+		}
+
+		nodes := make([]Node, 0, len(configs))
+		for _, c := range configs {
+			node, err := createNode(&c, factory)
+			if err != nil {
+				return nil, err
+			}
+			nodes = append(nodes, node)
+		}
+
+		if len(config.Parallel) != 0 {
+			return &Parallel{nodes: nodes}, nil
+		} else {
+			return &Sequence{nodes: nodes}, nil
+		}
+	default:
+		return nil, errors.New("empty node")
+	}
+}

go.mod

@@ -0,0 +1,5 @@
+module github.com/zwkno1/goflow
+
+go 1.17
+
+require github.com/pkg/errors v0.9.1

go.sum

@@ -0,0 +1,2 @@
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

node.go

@@ -0,0 +1,7 @@
+package goflow
+
+import "context"
+
+type Node interface {
+	Go(ctx context.Context) error
+}

parallel.go

@@ -0,0 +1,33 @@
+package goflow
+
+import (
+	"context"
+	"sync"
+)
+
+type Parallel struct {
+	nodes []Node
+}
+
+func (p *Parallel) Go(ctx context.Context) error {
+	var wg sync.WaitGroup
+	wg.Add(len(p.nodes))
+
+	errs := make([]error, len(p.nodes))
+
+	for i := range p.nodes {
+		err := &errs[i]
+		node := p.nodes[i]
+		go func() {
+			defer wg.Done()
+			*err = node.Go(ctx)
+		}()
+	}
+	wg.Wait()
+	for _, err := range errs {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

sequence.go

@@ -0,0 +1,16 @@
+package goflow
+
+import "context"
+
+type Sequence struct {
+	nodes []Node
+}
+
+func (s *Sequence) Go(ctx context.Context) error {
+	for _, node := range s.nodes {
+		if err := node.Go(ctx); err != nil {
+			return err
+		}
+	}
+	return nil
+}