flow registry helpers, preparing demo

browser/vue-flow/src/assets/style.css

@@ -82,6 +82,14 @@ input {
   padding: 10px;
 }
 
+textarea {
+  outline: none;
+  background: #fff;
+  width: 100%;
+  border: none;
+  padding: 10px;
+}
+
 /* Let's get this party started */
 ::-webkit-scrollbar {
   padding-top: 10px;

browser/vue-flow/src/components/flow/manager.vue

@@ -76,7 +76,7 @@
 import FlowNode from './node'
 import FlowLink from './link'
 import FlowPanZoom from './panzoom'
-import SvgDefs from './svgdefwrapper.vue'
+import SvgDefs from './svgdefswrapper'
 import utils from '@/utils/utils'
 
 export default {
@@ -117,6 +117,10 @@ export default {
           }
         }
         const nodeClass = this.registry[node.src]
+        if (!nodeClass) {
+          this.nodeRemove(node)
+          return
+        }
         return {
           transform: `translate(${node.x} ${node.y})`,
           id: node.id,
@@ -139,6 +143,9 @@ export default {
 
         const refFrom = this.$refs.nodes.find(n => n.id === link.from)
         const refTo = this.$refs.nodes.find(n => n.id === link.to)
+        if (!refFrom) { // delete link
+          return {}
+        }
 
         const fromOutput = refFrom.outputPos(0) // only 1 output
         const toInput = refTo.inputPos(link.in)
@@ -324,7 +331,7 @@ export default {
           const output = this.registry[nodeFrom.src].output
           const input = this.registry[nodeTo.src].inputs[link.in]
           // Type checking
-          if (!(output === 'any' || output === input || input === 'any')) {
+          if (!(output === 'interface {}' || output === input || input === 'interface {}')) {
             console.error('LINK: Invalid type')
             return
           }
@@ -414,7 +421,6 @@ export default {
       this.sendDocumentUpdate()
     },
     linkPointerClick (ev, link) {
-      console.log('Link click')
       ev.preventDefault()
       this.linkRemove(link)
     },

browser/vue-flow/src/components/flow/node-activity.vue

@@ -67,7 +67,6 @@ export default {
     }
   },
   mounted () {
-    console.log('Activity', this.activity)
     this._timeOut = setTimeout(this.updateTime, 999)
   },
   beforeDestroy () {

browser/vue-flow/src/components/flow/node.vue

@@ -72,7 +72,7 @@
       v-bind="inputProps(i)"
       :data-nodeid="id"
       :data-in="i"
-      :class="{'flow-node__socket--match': match.in && (inp == match.in || match.in == 'any' || inp=='any')}"
+      :class="{'flow-node__socket--match': match.in && (inp == match.in || match.in == 'interface {}' || inp=='interface {}')}"
       :key="'in'+i"
       @mousedown.stop.prevent="socketPointerDown($event, {in:i})"
     />
@@ -83,7 +83,7 @@
       :data-nodeid="id"
       :data-out="0"
       :key="'out'+0"
-      :class="{ 'flow-node__socket--match': match.out && (output == match.out || match.out == 'any' || output == 'any'), }"
+      :class="{ 'flow-node__socket--match': match.out && (output == match.out || match.out == 'interface {}' || output == 'interface {}'), }"
       v-bind="outputProps(0)"
       @mousedown.stop.prevent="socketPointerDown($event, {out:0})"
     />

browser/vue-flow/src/components/flow/svgdefs.svg

@@ -1,4 +1,4 @@
-  <defs>
+<defs>
     <marker id="head" class="flow-link__head" orient="auto" markerWidth="4" markerHeight="8" refX="0.1" refY="4">
       <path d="M0,0 V8 L4,4 Z"/>
     </marker>
@@ -34,5 +34,4 @@
     <filter id="drag-shadow" x="-100%" y="-100%" width="300%" height="300%">
       <feDropShadow dx="0" dy="2" stdDeviation="3"/>
     </filter>
-
   </defs>

browser/vue-flow/src/components/flow/svgdefwrapper.vue

@@ -1,2 +1 @@
-<!-- Simple because -->
 <template src="./svgdefs.svg"/>

browser/vue-flow/src/components/main.vue

@@ -155,23 +155,22 @@ import '@/assets/style.css'
 const defRegistry = {
   'Input': {
     categories: ['core'],
-    output: 'any',
+    output: 'interface {}',
     style: { color: '#686', textColor: '#fff', shape: 'circle' },
     props: {} // should be sent in the node
   },
   'Variable': {
     categories: ['core'],
-    output: 'any',
+    output: 'interface {}',
     style: { color: '#88a', textColor: '#000' },
     props: {init: ''}
   },
   'Const': {
     categories: ['core'],
-    output: 'any',
+    output: 'interface {}',
     style: {
       color: '#555',
-      textColor: '#333',
-      shape: 'circle'
+      textColor: '#333'
     },
     props: {value: ''}
   }

browser/vue-flow/src/components/panel.vue

@@ -78,7 +78,7 @@ export default {
       return (g) => {
         const ret = Object.keys(this.registry).filter(v => this.registry[v].categories.includes(g))
           .map(v => {
-            return { src: v, label: v }
+            return { src: v, label: v.split('.').join('<br/>') }
           })
         if (!this.search) {
           return ret

go/src/flow/flow.go

@@ -11,13 +11,6 @@ import (
 	"sync"
 )
 
-// Global
-var (
-	Global       = registry.New()
-	Register     = Global.Register
-	Descriptions = Global.Descriptions
-)
-
 // Data interface
 type Data = interface{}
 
@@ -32,7 +25,7 @@ type opEntry struct {
 // We could Create a single array of operations
 // refs would only mean id, types would be embed in operation
 type Flow struct {
-	registry   *registry.Registry
+	registry   *registry.R
 	consts     map[string]Data
 	data       map[string]Data // Should be named, to fetch later
 	operations sync.Map
@@ -48,7 +41,7 @@ type Flow struct {
 // New create a new flow
 func New() *Flow {
 	return &Flow{
-		registry: Global,
+		registry: registry.Global,
 		// Data
 		consts:     map[string]Data{},
 		data:       map[string]Data{},
@@ -63,7 +56,7 @@ func (f *Flow) Err() error {
 }
 
 //SetRegistry use the registry specified
-func (f *Flow) SetRegistry(r *registry.Registry) *Flow {
+func (f *Flow) SetRegistry(r *registry.R) *Flow {
 	f.registry = r
 	// chain
 	return f

go/src/flow/operation.go

@@ -54,7 +54,12 @@ func (o *operation) Process(params ...Data) Data {
 
 // Every single one is run with this internally
 func (o *operation) processWithCtx(ctx OpCtx, params ...Data) Data {
-	entry, _ := o.flow.getOp(fmt.Sprint(o.id))
+	return o.process(ctx, params...)
+	/*entry, _ := o.flow.getOp(fmt.Sprint(o.id))
+	if entry == nil {
+		log.Println("Entry is nil for id:", o.id, ", why??")
+		return nil
+	}
 	entry.Lock()
 	defer entry.Unlock()
 
@@ -62,14 +67,13 @@ func (o *operation) processWithCtx(ctx OpCtx, params ...Data) Data {
 		return nil
 	}
 	if ctx == nil { // No cache/Context
-		return o.process(ctx, params...)
 	}
 	if v, ok := ctx.Load(o.id); ok {
 		return v
 	}
 	res := o.process(ctx, params...)
 	ctx.Store(o.id, res)
-	return res
+	return res*/
 }
 
 // Set setter for certain operations (Var)
@@ -108,6 +112,8 @@ func opConst(f *Flow, id string) *operation {
 		},
 	}
 }
+
+// Debug type
 func opFunc(f *Flow, id string) *operation {
 	return &operation{
 		flow: f,
@@ -123,6 +129,18 @@ func opFunc(f *Flow, id string) *operation {
 				return nil
 			}
 
+			if f.err != nil {
+				return nil
+			}
+			op.Lock()
+			defer op.Unlock()
+			if ctx != nil {
+				if v, ok := ctx.Load(id); ok {
+					return v
+				}
+			}
+
+			// Check inputs
 			fnval := reflect.ValueOf(op.executor)
 			if fnval.Type().NumIn() != len(op.inputs) {
 				f.err = fmt.Errorf("expect %d inputs got %d", fnval.Type().NumIn(), len(op.inputs))
@@ -137,6 +155,7 @@ func opFunc(f *Flow, id string) *operation {
 
 			callParam := make([]reflect.Value, len(op.inputs))
 
+			// Parallel processing if inputs
 			wg := sync.WaitGroup{}
 			wg.Add(len(op.inputs))
 			for i, in := range op.inputs {
@@ -147,27 +166,39 @@ func opFunc(f *Flow, id string) *operation {
 				}(i, in)
 			}
 			wg.Wait()
-			// Check params
-			for _, p := range callParam {
-				if !p.IsValid() {
-					f.err = fmt.Errorf("Input failed", p)
-					log.Println("Flow err:", f.err)
+
+			f.hooks.start(id)
+
+			// Check params results
+			for i, p := range callParam {
+				f.err = func() error {
+					if !p.IsValid() {
+						return fmt.Errorf("Input is not valid %v", p)
+					}
+					if !p.Type().AssignableTo(fnval.Type().In(i)) {
+						return fmt.Errorf("Input not assignable to %v", p)
+					}
+					return nil
+				}()
+				if f.err != nil {
 					f.hooks.error(id, f.err)
 					return nil
 				}
 			}
-			f.hooks.start(id)
+
 			fnret := fnval.Call(callParam)
-			if len(fnret) > 1 {
-				if err := fnret[1].Interface().(error); err != nil {
-					f.err = err
-					log.Println("Flow err:", f.err)
-					f.hooks.error(id, f.err)
-					return nil
-				}
+			if len(fnret) > 1 && (fnret[1].Interface().(error) != nil) {
+				f.err = fnret[1].Interface().(error)
+				log.Println("Flow err:", f.err)
+				f.hooks.error(id, f.err)
+				return nil
 			}
-			// check fnret[1] for error
+
+			// THE RESULT
 			ret := fnret[0].Interface()
+			if ctx != nil {
+				ctx.Store(id, ret)
+			}
 			f.hooks.finish(id, ret)
 			return ret
 		},

go/src/flow/registry/entry.go

@@ -20,14 +20,16 @@ type Description struct {
 
 // Entry contains a function description params etc
 type Entry struct {
-	registry    *Registry
+	registry    *R
 	fn          interface{}
+	Inputs      []reflect.Type
+	Output      reflect.Type
 	Description *Description
 	err         error
 }
 
 // NewEntry creates and describes a New Entry
-func NewEntry(r *Registry, fn interface{}) *Entry {
+func NewEntry(r *R, fn interface{}) *Entry {
 	e := &Entry{registry: r, fn: fn}
 
 	fntyp := reflect.TypeOf(e.fn)
@@ -44,9 +46,11 @@ func NewEntry(r *Registry, fn interface{}) *Entry {
 	Inputs := []string{}
 	nInputs := fnTyp.NumIn()
 	for i := 0; i < nInputs; i++ {
+		e.Inputs = append(e.Inputs, fnTyp.In(i))
 		Inputs = append(Inputs, fmt.Sprint(fnTyp.In(i)))
 	}
 	Output := fmt.Sprint(fnTyp.Out(0))
+	e.Output = fnTyp.Out(0)
 
 	e.Description = &Description{
 		Categories: []string{"generic"},
@@ -72,8 +76,8 @@ func (e *Entry) Categories(cat ...string) *Entry {
 	return e
 }
 
-// Input description for Input
-func (e *Entry) Input(i int, desc string) *Entry {
+// DescInput description for Input
+func (e *Entry) DescInput(i int, desc string) *Entry {
 	if e.err != nil {
 		return e
 	}
@@ -81,8 +85,8 @@ func (e *Entry) Input(i int, desc string) *Entry {
 	return e
 }
 
-// Output description for Input
-func (e *Entry) Output(desc string) *Entry {
+// DescOutput description for Input
+func (e *Entry) DescOutput(desc string) *Entry {
 	if e.err != nil {
 		return e
 	}

go/src/flow/registry/entry_test.go

@@ -28,11 +28,11 @@ func TestDescription(t *testing.T) {
 	a.Eq(e.Err(), nil, "should not fail setting categories")
 	a.Eq(len(e.Description.Categories), 2, "should have 2 categories")
 
-	e.Input(0, "input doc")
+	e.DescInput(0, "input doc")
 	a.Eq(e.Err(), nil, "should not fail setting input description")
 	a.Eq(len(e.Description.InputDesc), 1, "should have 1 input description")
 
-	e.Output("output desc")
+	e.DescOutput("output desc")
 	a.Eq(e.Err(), nil, "should not fail setting input description")
 	a.Eq(e.Description.OutputDesc, "output desc", "output description should be the same")
 
@@ -47,9 +47,9 @@ func TestDescriptionError(t *testing.T) {
 	a.Eq(e.Err(), registry.ErrNotAFunc, "entry is not a function")
 	e.Categories("a", "b")
 	a.Eq(e.Err(), registry.ErrNotAFunc, "entry is not a function")
-	e.Input(0, "input doc")
+	e.DescInput(0, "input doc")
 	a.Eq(e.Err(), registry.ErrNotAFunc, "entry is not a function")
-	e.Output("output desc")
+	e.DescOutput("output desc")
 	a.Eq(e.Err(), registry.ErrNotAFunc, "entry is not a function")
 	e.Extra("test", 123)
 	a.Eq(e.Err(), registry.ErrNotAFunc, "entry is not a function")

go/src/flow/registry/registry.go

@@ -1,22 +1,45 @@
 package registry
 
 import (
+	"path"
 	"reflect"
+	"runtime"
 )
 
-// Registry function registry
-type Registry struct {
+// Global
+var (
+	Global       = New()
+	Register     = Global.Register
+	Descriptions = Global.Descriptions
+	GetEntry     = Global.Entry
+	Add          = Global.Add
+)
+
+// R the function registry
+type R struct {
 	data map[string]*Entry
 }
 
 // New creates a new registry
-func New() *Registry {
-	return &Registry{map[string]*Entry{}}
+func New() *R {
+	return &R{map[string]*Entry{}}
 }
 
-//Register should be a func only register
-func (r *Registry) Register(name string, v interface{}) *Entry {
+// Add unnamed function
+func (r *R) Add(fns ...interface{}) Batch {
+
+	b := Batch{}
+	for _, fn := range fns {
+		name := runtime.FuncForPC(reflect.ValueOf(fn).Pointer()).Name()
+		name = path.Base(name)
+		b = append(b, r.Register(name, fn))
+	}
+
+	return b
+}
 
+//Register should be a func only register
+func (r *R) Register(name string, v interface{}) *Entry {
 	e := NewEntry(r, v)
 	if e.err != nil {
 		return e
@@ -26,7 +49,7 @@ func (r *Registry) Register(name string, v interface{}) *Entry {
 }
 
 // Get an entry
-func (r *Registry) Get(name string, params ...interface{}) (interface{}, error) {
+func (r *R) Get(name string, params ...interface{}) (interface{}, error) {
 	e, ok := r.data[name]
 	if !ok {
 		return nil, ErrNotFound
@@ -51,7 +74,7 @@ func (r *Registry) Get(name string, params ...interface{}) (interface{}, error)
 }
 
 // Entry fetches entries from the register
-func (r *Registry) Entry(name string) (*Entry, error) {
+func (r *R) Entry(name string) (*Entry, error) {
 	e, ok := r.data[name]
 	if !ok {
 		return nil, ErrNotFound
@@ -62,7 +85,7 @@ func (r *Registry) Entry(name string) (*Entry, error) {
 //Describe named fn
 
 // Descriptions Description list
-func (r *Registry) Descriptions() (map[string]Description, error) {
+func (r *R) Descriptions() (map[string]Description, error) {
 	ret := map[string]Description{}
 	for k, e := range r.data {
 		ret[k] = *e.Description

go/src/flowserver/cmd/flowserver/floatops/floatops.go

@@ -0,0 +1,29 @@
+package floatops
+
+import (
+	"errors"
+	"flow/registry"
+	"math"
+)
+
+// Register into registry
+func Register(r *registry.R) {
+
+	r.Add(sum, sub, div, mul).Categories("floatops")
+}
+func sum(a, b float32) float32 {
+	return a + b
+}
+func sub(a, b float32) float32 {
+	return a - b
+}
+func div(a, b float32) (float32, error) {
+	if b == 0 {
+		return float32(math.NaN()), errors.New("Division by 0")
+	}
+	return a / b, nil
+}
+
+func mul(a, b float32) float32 {
+	return a * b
+}

go/src/flowserver/cmd/flowserver/main.go

@@ -1,14 +1,13 @@
 package main
 
 import (
-	"errors"
 	"flow"
 	"flow/registry"
 	"flowserver"
+	"flowserver/cmd/flowserver/floatops"
 	"fmt"
+	"io"
 	"log"
-	"math/rand"
-	"strings"
 	"time"
 
 	"github.com/gohxs/prettylog"
@@ -16,46 +15,19 @@ import (
 
 func main() {
 	prettylog.Global()
-
 	log.Println("Running version:", flowserver.Version)
 
-	registry.Batch{
-		flow.Register("delay.cat", delayCat),
-		flow.Register("delay.reverse", delayReverse),
-		flow.Register("delay.split", delaySplit),
-		flow.Register("delay.join", delayJoin),
-		flow.Register("addduration", duration),
-	}.Categories("string", "test").
-		Extra("style", map[string]string{
-			"color": "#a55",
-		})
+	floatops.Register(registry.Global)
 
-	registry.Batch{
-		flow.Register("longduration", duration(10*time.Second)),
-		flow.Register("mediumduration", duration(5*time.Second)),
-		flow.Register("randomduration", randduration(10*time.Second)),
-	}.Categories("time simulation")
+	registry.Add(customStruct, setName, setURL).Categories("test-struct")
 
-	registry.Batch{
-		flow.Register("multiplex4", multiplex4),
-		flow.Register("multiplex6", multiplex6),
-		flow.Register("error", func(s string) (string, error) {
-			time.Sleep(4 * time.Second)
-			return "", errors.New("Some error")
-		}),
-	}.Categories("test").
-		Extra("style", map[string]string{
-			"color": "#959",
-		})
+	registry.Batch{registry.Register("stream", stream)}.Categories("streamer")
 
 	registry.Batch{
-		flow.Register("str.cat", strCat),
-		flow.Register("str.reverse", strReverse),
-		flow.Register("str.split", strings.Split),
-		flow.Register("str.join", strings.Join),
-	}.Categories("string").
+		registry.Register("wait", wait),
+	}.Categories("time-test").
 		Extra("style", map[string]string{
-			"color": "#aa5",
+			"color": "#8a5",
 		})
 
 	addr := ":2015"
@@ -65,65 +37,30 @@ func main() {
 	f.ListenAndServe(addr)
 }
 
-func multiplex4(a1, a2, a3, a4 string) string {
-	return strings.Join([]string{a1, a2, a3, a4}, " ")
-}
-func multiplex6(a1, a2, a3, a4, a5, a6 string) string {
-	return strings.Join([]string{a1, a2, a3, a4, a5, a6}, " ")
+func wait(data flow.Data, n int) flow.Data {
+	time.Sleep(time.Duration(n) * time.Second) // Simulate
+	return data
 }
 
-/* Sample funcs */
-func strCat(a1, a2 string) string {
-	return a1 + " " + a2
+func stream(w io.Writer, val interface{}) interface{} {
+	fmt.Fprint(w, val)
+	return val
 }
 
-func strReverse(s string) string {
-	n := len(s)
-	runes := make([]rune, n)
-	for _, r := range s {
-		n--
-		runes[n] = r
-	}
-	return string(runes[n:])
+// CustomStruct testing custom struct passing
+type CustomStruct struct {
+	Name string
+	Url  string
 }
 
-func delayCat(a1, a2 string) string {
-	time.Sleep(time.Duration(2+rand.Intn(10)) * time.Second) // Simulate
-	return strCat(a1, a2)
+func customStruct() CustomStruct {
+	return CustomStruct{}
 }
-
-func delayReverse(s string) string {
-	time.Sleep(time.Duration(2+rand.Intn(10)) * time.Second) // Simulate
-	return strReverse(s)
-
-}
-func delaySplit(s, sep string) []string {
-	time.Sleep(time.Duration(2+rand.Intn(10)) * time.Second) // Simulate
-	return strings.Split(s, sep)
-}
-func delayJoin(s []string, join string) string {
-	time.Sleep(time.Duration(2+rand.Intn(10)) * time.Second) // Simulate
-	return strings.Join(s, join)
-}
-
-func longduration() string {
-	mark := time.Now()
-	time.Sleep(25 * time.Second) // Simulate
-	now := time.Now()
-
-	return " took: " + fmt.Sprint(now.Sub(mark))
-
-}
-
-func duration(n time.Duration) func() string {
-	return func() string {
-		time.Sleep(n)
-		return fmt.Sprint("I waited:", n)
-	}
+func setName(a CustomStruct, name string) CustomStruct {
+	a.Name = name
+	return a
 }
-func randduration(n time.Duration) func() string {
-	return func() string {
-		time.Sleep(time.Duration(2 + rand.Intn(int(n)))) // Simulate
-		return fmt.Sprint("I waited:", n)
-	}
+func setURL(a CustomStruct, url string) CustomStruct {
+	a.Url = url
+	return a
 }

go/src/flowserver/flowbuilder.go

@@ -3,6 +3,9 @@ package flowserver
 import (
 	"encoding/json"
 	"flow"
+	"flow/registry"
+	"log"
+	"reflect"
 )
 
 // Node that will contain registry src
@@ -42,16 +45,22 @@ func FlowBuild(rawData []byte) (*flow.Flow, error) {
 
 	ninput := 0
 	for _, n := range doc.Nodes {
+		if n.Src == "Input" || n.Src == "Variable" || n.Src == "Const" {
+			continue
+		}
+
 		// Find link refered as To
-		param := make([]interface{}, 10) // 10 is temporary to test out operations
-		lastParamID := -1
+		entry, err := registry.GetEntry(n.Src)
+		if err != nil {
+			return nil, err
+		}
+
+		param := make([]flow.Data, len(entry.Inputs))
+		// Find links
 		for _, l := range doc.Links {
 			if l.To != n.ID {
 				continue
 			}
-			if l.In > lastParamID {
-				lastParamID = l.In
-			}
 			// Define operators here
 			from := nodeMap[l.From]
 			switch from.Src {
@@ -66,15 +75,23 @@ func FlowBuild(rawData []byte) (*flow.Flow, error) {
 			case "Variable":
 				param[l.In] = f.Var(from.ID, from.Prop["init"])
 			case "Const":
-				param[l.In] = f.Const(from.Prop["value"])
+				// XXX: Automate this in a func
+				newVal := reflect.New(entry.Inputs[l.In])
+				raw := from.Prop["value"]
+				if _, ok := newVal.Interface().(*string); ok {
+					raw = "\"" + from.Prop["value"] + "\""
+				}
+				err := json.Unmarshal([]byte(raw), newVal.Interface())
+				if err != nil {
+					log.Println("Error", err)
+					param[l.In] = nil
+					continue
+				}
+				param[l.In] = f.Const(newVal.Elem().Interface())
 			default:
 				param[l.In] = f.Res(from.ID)
 			}
 		}
-		param = param[:lastParamID+1]
-		if n.Src == "Input" || n.Src == "Variable" || n.Src == "Const" {
-			continue
-		}
 		f.DefOp(n.ID, n.Src, param...)
 	}
 	return f, nil

go/src/flowserver/session.go

@@ -4,7 +4,9 @@ import (
 	"encoding/json"
 	"errors"
 	"flow"
+	"flow/registry"
 	"flowserver/flowmsg"
+	"fmt"
 	"io/ioutil"
 	"log"
 	"os"
@@ -25,7 +27,7 @@ type NodeActivity struct {
 	StartTime time.Time `json:"startTime"`
 	EndTime   time.Time `json:"endTime"`
 	Data      flow.Data `json:"data"`
-	Error     error     `json:"error"`
+	Error     string    `json:"error"`
 }
 
 // FlowSession Create a session and link clients
@@ -84,7 +86,7 @@ func (s *FlowSession) ClientAdd(c *websocket.Conn) error {
 	if err != nil {
 		return err
 	}
-	desc, err := flow.Descriptions()
+	desc, err := registry.Descriptions()
 	if err != nil {
 		return err
 	}
@@ -102,6 +104,7 @@ func (s *FlowSession) ClientAdd(c *websocket.Conn) error {
 		return err
 	}
 
+	// Sending activity
 	return c.WriteJSON(flowmsg.SendMessage{OP: "nodeActivity", Data: s.nodeActivity})
 
 	// Send registry
@@ -186,6 +189,7 @@ func (s *FlowSession) NodeRun(c *websocket.Conn, data []byte) error {
 		s.flow, err = FlowBuild(s.RawDoc)
 		if err != nil {
 			log.Println("Flow error:", err)
+			return
 		}
 		defer func() { // After routing gone
 			s.flow = nil
@@ -214,7 +218,7 @@ func (s *FlowSession) NodeRun(c *websocket.Conn, data []byte) error {
 				case "Error":
 					status = "error"
 					act.EndTime = triggerTime
-					act.Error = extra[0].(error)
+					act.Error = fmt.Sprint(extra[0])
 				}
 				if act.Status == status {
 					return