Chain is a package

chain/chain.go

@@ -0,0 +1,104 @@
+package chain
+
+import (
+	"bytes"
+	"sync"
+
+	"github.com/localhots/yeast/unit"
+)
+
+type (
+	Chain struct {
+		Flow  Flow
+		Links []unit.Caller
+	}
+)
+
+const (
+	LF = byte(10)
+)
+
+func New(name string) *Chain {
+	c, _ := chains[name]
+	return c
+}
+
+func (c *Chain) Call(data []byte) (resp []byte, err error) {
+	switch c.Flow {
+	case SequentialFlow:
+		return c.processSequentially(data)
+	case ParallelFlow:
+		return c.processInParallel(data)
+	case DelayedFlow:
+		return c.processDelayed(data)
+	default:
+		panic("Unreachable")
+	}
+}
+
+func (c *Chain) Units() []string {
+	// Collecting unique unit names using map
+	units := map[string]*struct{}{}
+	for _, caller := range c.Links {
+		for _, unit := range caller.Units() {
+			units[unit] = nil
+		}
+	}
+
+	// Extracting names to a slice
+	uniq := []string{}
+	for unit, _ := range units {
+		uniq = append(uniq, unit)
+	}
+
+	return uniq
+}
+
+func (c *Chain) processSequentially(data []byte) (resp []byte, err error) {
+	for _, caller := range c.Links {
+		if resp, err = caller.Call(data); err != nil {
+			return
+		}
+	}
+	return
+}
+
+func (c *Chain) processInParallel(data []byte) (resp []byte, err error) {
+	var (
+		inbox = make(chan []byte) // This channel must be unbuffered
+		buf   bytes.Buffer
+		wg    sync.WaitGroup
+	)
+
+	for _, caller := range c.Links {
+		wg.Add(1)
+		go func() {
+			if res, err := caller.Call(data); err == nil {
+				inbox <- res
+			}
+			wg.Done()
+		}()
+	}
+	go func() {
+		wg.Wait()
+		close(inbox)
+	}()
+
+	for {
+		if res, ok := <-inbox; ok {
+			buf.Write(res)
+			buf.WriteByte(LF) // Add linebreak
+		} else {
+			break
+		}
+	}
+
+	return buf.Bytes(), nil
+}
+
+func (c *Chain) processDelayed(data []byte) (resp []byte, err error) {
+	for _, caller := range c.Links {
+		go caller.Call(data)
+	}
+	return data, nil
+}

chain/parser.go

@@ -0,0 +1,72 @@
+package chain
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"reflect"
+
+	"github.com/localhots/yeast/unit"
+)
+
+var (
+	chains = map[string]*Chain{}
+)
+
+func LoadChains(path string) {
+	f, err := os.Open(path)
+	if err != nil {
+		panic("Failed to open chains config: " + path)
+	}
+	b, err := ioutil.ReadAll(f)
+	if err != nil {
+		panic("Failed to read chains config: " + path)
+	}
+
+	var schema map[string]interface{}
+	if err := json.Unmarshal(b, &schema); err != nil {
+		panic("Failed to parse chains config: " + path)
+	}
+
+	for name, c := range schema {
+		chains[name] = Parse(interface{}(c))
+	}
+}
+
+func Parse(conf interface{}) *Chain {
+	c := &Chain{
+		Links: []unit.Caller{},
+	}
+
+	for f, links := range conf.(map[string]interface{}) {
+		if flow := FlowOf(f); flow != UnknownFlow {
+			c.Flow = flow
+		} else {
+			panic("Unknown chain flow: " + f)
+		}
+
+		for _, link := range links.([]interface{}) {
+			val := reflect.ValueOf(link)
+
+			switch val.Kind() {
+			case reflect.Map:
+				subchain := Parse(interface{}(link))
+				if len(subchain.Links) > 0 {
+					c.Links = append(c.Links, unit.Caller(subchain))
+				}
+			case reflect.String:
+				name := link.(string)
+				if caller := unit.New(name); caller != nil {
+					c.Links = append(c.Links, caller)
+				} else {
+					fmt.Println("Unknown unit:", name)
+				}
+			default:
+				panic("Unexpected chain element: " + val.Kind().String())
+			}
+		}
+	}
+
+	return c
+}