Portal implementation

browser/vue-flow/src/assets/doc/readme.md

@@ -111,6 +111,64 @@ registry.Describer(
 ![fig4](img/runningsplit.jpg)
 ![fig5](img/runningjoin.jpg)
 
+### Special Operations
+
+#### f.In
+
+Uses an argument passed through the process function
+
+```go
+op := f.In(0)
+res, err := op.Process("arg")
+// res will be "arg"
+```
+
+#### f.Var
+
+Variable that stores data in the flow (for future serialization) it can be
+restored
+
+```go
+f.Var("myvar", "initial value or operation")
+```
+
+#### f.SetVar
+
+Set var creates an operation that sets a variable to the result of the operation
+passed as the argument
+
+```go
+f.SetVar("myvar", operation)
+```
+
+### UI Special nodes
+
+#### Input
+
+Same as f.In where a property must be set to indicate which argument this
+node refers to
+
+#### Output
+
+Flow-UI only output is a special node that indicates the overall flow output,
+the UI contains a RUN button that will be visible if there is an output node and
+the flow will be executed from the output node.
+
+There can be only one output node
+
+#### Var
+
+Same as f.Var passing the variable name as a property of the node
+
+#### SetVar
+
+Same as f.SetVar passing the variable name as a property of the node
+
+#### Portal From
+
+UI Only: Portals are helper nodes that allows to connect areas of UI without
+crossing any links, right click in a node and choose create portal
+
 ---
 
 > WIP
@@ -124,7 +182,8 @@ registry.Describer(
 * UX/UI: Ability to group nodes into a single box exposing inputs and outputs
 * UX/UI: Implement touch
 * UX/UI: Drop link in node to link to the next compatible input
-* UX/UI: Portals to clean graph crossing
+* ~~UX/UI: Portals to clean graph crossing~~ (testing)
+* UX/UI: `Shift` key to merge group selections
 
 ### Packages
 
@@ -140,6 +199,7 @@ registry.Describer(
 ### Other
 
 * Collaboration: Better concurrent editing/message passing
+* Flow: Proper documentation
 
 ## Ideas
 
@@ -159,5 +219,13 @@ generate code based on a flow should be simple as we have the function signature
 
 ### Portals
 
+> Testing phase
+
 Named portal would connect a node to another without any link, this way we can
 have clean links without crossovers
+
+### Recursive registering
+
+Since an operation can be run with op.Process(inputs...) we can simple do
+registry.Add(op.Process) to register a flow as an operation, UI doesn't retain
+the registry/flow yet

browser/vue-flow/src/components/app-readme.vue

@@ -67,9 +67,9 @@ export default {
 
 .app-readme__menu {
   flex-grow:0;
-  min-width:200px;
+  min-width:250px;
   height:100vh;
-  padding-top:100px;
+  padding:80px 30px;
   border-right: solid 1px rgba(150,150,150,0.5);
   overflow-y:auto;
 }
@@ -102,7 +102,7 @@ export default {
 }
 
 .app-readme__menu ul li {
-  line-height:40px;
+  line-height:34px;
 }
 
 .app-readme__menu ul li a {

browser/vue-flow/src/components/flow/editor.js

@@ -156,7 +156,7 @@ export default {
     ...mapActions('flow', [
       'NOTIFICATION_ADD',
       'DOCUMENT_SYNC',
-      'NODE_RAISE', 'NODE_UPDATE', 'NODE_ADD', 'NODE_REMOVE', 'NODE_INSPECT', 'NODE_PROCESS',
+      'NODE_RAISE', 'NODE_UPDATE', 'NODE_ADD', 'NODE_REMOVE', 'NODE_INSPECT', 'NODE_PROCESS', 'NODE_TRAIN',
       'LINK_ADD', 'LINK_REMOVE',
       'TRIGGER_ADD', 'TRIGGER_REMOVE' ]),
 
@@ -526,7 +526,27 @@ export default {
       const n = this.nodeData.nodes.find(n => n.src === 'Output')
       this.NODE_PROCESS(n.id)
     },
+    createPortal (nodeID) {
+      // Find nodeID
+      let node = this.nodeData.nodes.find(n => n.id === nodeID)
+      if (!node) {
+        this.NOTIFICATION_ADD('invalid node ID' + nodeID)
+        return
+      }
+      // Special node
+      const portalNode = {
+        id: utils.guid(),
+        x: node.x + 10,
+        y: node.y + 100, // Downthere/improve this
+        defaultInputs: {},
+        label: node.label,
+        color: node.color,
+        prop: {'portal from': nodeID},
+        src: 'Portal From'
+      }
 
+      this.NODE_ADD(portalNode)
+    },
     // HELPERS depending on svg ref
     createSVGPoint (x, y) {
       const p = this.$refs.svg.createSVGPoint()

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

@@ -85,8 +85,11 @@
       <template slot-scope="d" >
         <div class="flow-node__context-menu">
           <div class="hover" @click="NODE_PROCESS(d.userData.id)">Run</div>
+          <div class="hover" @click="NODE_TRAIN(d.userData.id)">Train(temporary)</div>
           <div class="hover" @click="NODE_REMOVE([d.userData])">Delete</div>
           <hr>
+          <div class="hover" @click="createPortal(d.userData.id)">Create Portal</div>
+          <hr>
           <div class="hover" @click="nodeInspect(d.userData,true)">Inspect</div>
         </div>
       </template>

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

@@ -13,17 +13,41 @@
 
     <!-- shape -->
     <template>
-      <svg
-        v-if="style.shape == 'thing'"
+      <!--<svg
+        v-if="style.shape == 'portal'"
         ref="body"
         viewBox="0 0 100 100"
-        preserveAspectRatio="XMinYMin"
         class="flow-node__body"
         :class="{'flow-node__body--dragging':dragging}"
+        preserveAspectRation="xMinYMax"
         v-bind="bodyProps"
       >
-        <path d=" M 0 0 l 90 0 l 10 50 l -10 50 l -90 0 c 10 -20, 10 -80, 0 -100 Z " />
-      </svg>
+        <path d="
+        M 50 0
+        L 100 50
+        L 50 100
+        L 0 50
+        Z"/>
+        <!--<path d="
+      M 10 0
+      l 80 0
+      l 17 50
+      l -17 50
+      l -80 0
+      l -17 -50
+      Z " />-->
+      <!--c 10 -20, 10 -80, 0 -100-->
+      <!--</svg>-->
+      <rect
+        v-if="style.shape == 'portal'"
+        ref="body"
+        class="flow-node__body"
+        :class="{'flow-node__body--dragging':dragging}"
+        v-bind="bodyProps"
+        rx="25"
+        ry="25"
+      />
+
       <circle
         v-else-if="style.shape == 'circle'"
         ref="body"
@@ -234,6 +258,12 @@ export default {
       if (this.style.shape === 'circle') {
         rect.r = Math.max(width / 2, height / 2)
       }
+      if (this.style.shape === 'portal') {
+        /* rect.x -= 20
+        rect.y -= 20
+        rect.width += 40
+        rect.height += 40 */
+      }
       return rect
     },
     inputProps () {

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

@@ -108,7 +108,7 @@ export default {
     }
   },
   computed: {
-    ...mapGetters('flow', ['registry', 'activity']),
+    ...mapGetters('flow', ['registry', 'activity', 'nodeData']),
     nodeActivity () {
       return this.activity && this.activity.nodes && this.activity.nodes[this.nodeInspect.id]
     }
@@ -121,7 +121,35 @@ export default {
   methods: {
     ...mapActions('flow', ['NODE_UPDATE', 'DOCUMENT_SYNC', 'NODE_PROCESS']),
     localChange () {
-      this.NODE_UPDATE([JSON.parse(JSON.stringify(this.nodeInspect))])
+      let nodeUpdates = [JSON.parse(JSON.stringify(this.nodeInspect))]
+
+      // PORTAL related code
+      // Experimental find portals with same name
+      let n = this.nodeInspect
+
+      let theID = n.id
+      let relatedPortals = []
+      if (n.src === 'Portal From') {
+        let nodeFrom = this.nodeData.nodes.find(nn => nn.id === n.prop['portal from'])
+        relatedPortals.push(nodeFrom)
+        theID = nodeFrom.id
+      }
+      this.nodeData.nodes.forEach(nn => {
+        if (nn === n) { return }
+        if (nn.src !== 'Portal From') { return }
+        if (nn.prop['portal from'] !== theID) { return }
+        relatedPortals.push(nn)
+      })
+      // Update label and color perhaps
+      for (let nn of relatedPortals) {
+        let portal = JSON.parse(JSON.stringify(nn))
+        portal.label = n.label
+        portal.color = n.color
+        nodeUpdates.push(portal)
+      }
+      // Find the relative portal node
+
+      this.NODE_UPDATE(nodeUpdates)
       this.DOCUMENT_SYNC()
       // Seems that there might be browsers triggering the input before the v-model
       // so we defer the execution until we have nodeInspect updated

browser/vue-flow/src/services/flowservice.js

@@ -5,7 +5,8 @@ const methods = [
   'chatEvent', 'chatJoin', 'chatRename', // CHAT
   'linkAdd', 'linkUpdate', 'linkRemove', // LINK
   'triggerAdd', 'triggerUpdate', 'triggerRemove',
-  'nodeUpdate', 'nodeAdd', 'nodeRemove', 'nodeProcess' // NODE
+  'nodeUpdate', 'nodeAdd', 'nodeRemove', 'nodeProcess', // NODE
+  'nodeTrain' // XXX: experimental (should be removed)
 ]
 
 const debug = 0

browser/vue-flow/src/store/flow/actions.js

@@ -15,9 +15,9 @@ export default {
   [m.NODE_RAISE] ({commit}, nodes) {
     commit(m.NODE_RAISE, nodes)
   },
-  [m.NODE_UPDATE] ({commit}, nodes) {
+  [m.NODE_UPDATE] (ctx, nodes) {
     // WEBSOCKET
-    commit(m.NODE_UPDATE, nodes)
+    ctx.commit(m.NODE_UPDATE, nodes)
     flowService.nodeUpdate(nodes)
   },
 
@@ -37,6 +37,10 @@ export default {
   [m.NODE_PROCESS] (ctx, nodeId) {
     flowService.nodeProcess(nodeId)
   },
+  // XXX: Experimental
+  [m.NODE_TRAIN] (ctx, nodeId) {
+    flowService.nodeTrain(nodeId)
+  },
   [m.LINK_ADD] (ctx, link) {
     ctx.commit(m.LINK_ADD, link)
     ctx.dispatch(m.DOCUMENT_SYNC)

browser/vue-flow/src/store/flow/default-registry.js

@@ -20,11 +20,33 @@ export default{
   'SetVar': {
     categories: ['core'],
     inputs: [{type: 'interface {}', name: 'value'}],
-    output: {type: 'interface{}'},
+    output: {type: 'interface {}'},
     style: { color: '#88a', shape: 'circle' },
     props: {'variable name': ''}
   },
+  'Portal From': {
+    categories: ['core'],
+    output: {type: 'interface {}'},
+    style: { color: '#888', shape: 'portal' },
+    props: {'portal from': ''}
+  },
 
+  /* 'Portal In': {
+    categories: ['core'],
+    inputs: [{type: 'interface {}'}],
+    style: {color: '#ba5a00', shape: 'circle'},
+    props: {'portal name': ''}
+  },
+  'Portal Out': {
+    categories: ['core'],
+    output: {type: 'interface {}'},
+    style: {color: '#085798', shape: 'circle'},
+    props: {'portal name': ''}
+  }, */
+  'Comment': {
+    categories: ['core'],
+    style: {color: '#444'}
+  },
   'Notify': {
     categories: ['flow-web'],
     inputs: [{type: 'interface {}'}, {type: 'string', name: 'msg'}],

browser/vue-flow/src/store/flow/mutation-types.js

@@ -3,7 +3,7 @@ var actions = [
   'SESSID_UPDATE',
   'DOCUMENT_UPDATE', 'DOCUMENT_SYNC',
   'ACTIVITY_UPDATE',
-  'NODE_RAISE', 'NODE_UPDATE', 'NODE_ADD', 'NODE_REMOVE', 'NODE_INSPECT', 'NODE_PROCESS',
+  'NODE_RAISE', 'NODE_UPDATE', 'NODE_ADD', 'NODE_REMOVE', 'NODE_INSPECT', 'NODE_PROCESS', 'NODE_TRAIN',
   'LINK_ADD', 'LINK_REMOVE',
   'TRIGGER_ADD', 'TRIGGER_REMOVE',
   'NOTIFICATION_ADD', 'NOTIFICATION_CLEAR'

browser/vue-flow/webpack.config.js

@@ -93,7 +93,7 @@ module.exports = {
       },
       {
         test: /\.md$/,
-        loader: 'html-loader!highlight-loader!markdown-loader'
+        loader: 'html-loader!highlight-loader!markdown-loader?gfm=true'
       }
       /*,
       {

go/src/demos/cmd/xor/main.go

@@ -82,6 +82,7 @@ func main() {
 	dOutput := f.Op("matMulElem", networkError, slopeOutputLayer)
 	wOutAdj := f.Op("matMul", f.Op("matTranspose", hiddenLayerActivations), dOutput)
 	wOutAdj = f.Op("matScale", learningRate, wOutAdj)
+
 	// hidden weights
 	errorAtHiddenLayer := f.Op("matMul", dOutput, f.Op("matTranspose", wOut))
 	slopeHiddenLayer := f.Op("matSigmoidPrime", hiddenLayerActivations)

go/src/demos/ops/ml/gonumops.go

@@ -2,6 +2,7 @@
 package ml
 
 import (
+	"flow"
 	"flow/registry"
 	"math"
 	"math/rand"
@@ -32,6 +33,9 @@ func New() *registry.R {
 			matSigmoidPrime,
 			toFloatArr,
 		),
+		r.Add("train", func(a, b, c, d flow.Data) []flow.Data {
+			return []flow.Data{a, b, c, d}
+		}).Inputs("dummy", "dummy", "dummy", "dummy"),
 	).Description("gonum functions").
 		Tags("machine learning").
 		Extra("style", registry.M{"color": "#953"})

go/src/flow/flowserver/flowbuilder/builder.go

@@ -18,7 +18,7 @@ var ErrLoop = errors.New("Looping is disabled for now")
 // FlowBuilder builds a flow from flow-ui json data
 type FlowBuilder struct {
 	registry     *registry.R
-	doc          *FlowDocument
+	Doc          *FlowDocument
 	flow         *flow.Flow
 	OperationMap map[string]flow.Operation
 	nodeTrack    map[string]bool
@@ -57,7 +57,7 @@ func (fb *FlowBuilder) Load(rawData []byte) *FlowBuilder {
 		return fb
 	}
 
-	fb.doc = doc
+	fb.Doc = doc
 
 	return fb
 }
@@ -70,7 +70,7 @@ func (fb *FlowBuilder) Build(ID string) flow.Operation {
 	}
 	f := fb.flow
 	r := fb.registry
-	doc := fb.doc
+	doc := fb.Doc
 
 	if _, ok := fb.nodeTrack[ID]; ok {
 		fb.Err = ErrLoop //fmt.Errorf("[%v] Looping through nodes is disabled:", ID)
@@ -86,7 +86,7 @@ func (fb *FlowBuilder) Build(ID string) flow.Operation {
 		return op
 	}
 
-	node := fb.doc.fetchNodeByID(ID)
+	node := fb.Doc.FetchNodeByID(ID)
 	if node == nil {
 		op := fb.flow.ErrOp(fmt.Errorf("node not found [%v]", ID))
 		return op
@@ -96,6 +96,13 @@ func (fb *FlowBuilder) Build(ID string) flow.Operation {
 	var inputs []reflect.Type
 
 	switch node.Src {
+	case "Portal From":
+		nID := node.Prop["portal from"]
+		n := doc.FetchNodeByID(nID)
+		if n == nil {
+			return f.ErrOp(fmt.Errorf("Invalid portal, id: %v", nID))
+		}
+		return fb.Build(nID)
 	case "Input":
 		inputID, err := strconv.Atoi(node.Prop["input"])
 		if err != nil {
@@ -129,7 +136,7 @@ func (fb *FlowBuilder) Build(ID string) flow.Operation {
 	//// Build inputs ////
 	param := make([]flow.Data, len(inputs))
 	for i := range param {
-		l := doc.fetchLinkTo(node.ID, i)
+		l := doc.FetchLinkTo(node.ID, i)
 		if l == nil { // No link we fetch the value inserted
 			// Direct input entries
 			v, err := parseValue(inputs[i], node.DefaultInputs[i])
@@ -161,7 +168,7 @@ func (fb *FlowBuilder) Build(ID string) flow.Operation {
 
 func (fb *FlowBuilder) buildTriggersFor(node *Node, targetOp flow.Operation) error {
 	// Process triggers for this node
-	triggers := fb.doc.fetchTriggerFrom(node.ID)
+	triggers := fb.Doc.FetchTriggerFrom(node.ID)
 	log.Println("Operation has this triggers:", triggers)
 	for _, t := range triggers {
 		log.Println("will build for")

go/src/flow/flowserver/flowbuilder/model.go

@@ -1,7 +1,6 @@
 package flowbuilder
 
 // Node that will contain registry src
-import "log"
 
 type Node struct {
 	ID            string            `json:"id"`
@@ -32,7 +31,8 @@ type FlowDocument struct {
 	Triggers []Trigger `json:"triggers"`
 }
 
-func (fd *FlowDocument) fetchNodeByID(ID string) *Node {
+// FetchNodeByID retrieve a node by its ID
+func (fd *FlowDocument) FetchNodeByID(ID string) *Node {
 	for _, n := range fd.Nodes {
 		if n.ID == ID {
 			return &n
@@ -41,19 +41,31 @@ func (fd *FlowDocument) fetchNodeByID(ID string) *Node {
 	return nil
 }
 
-func (fd *FlowDocument) fetchTriggerFrom(ID string) []Trigger {
+// FetchNodeBySrc loops the nodes and filter src
+func (fd *FlowDocument) FetchNodeBySrc(src string) []Node {
+	ret := []Node{}
+	for _, n := range fd.Nodes {
+		if n.Src == src {
+			ret = append(ret, n)
+		}
+	}
+	return ret
+}
+
+// FetchTriggerFrom  fetch nodes where trigger comes from ID
+func (fd *FlowDocument) FetchTriggerFrom(ID string) []Trigger {
 	ret := []Trigger{}
 	for _, t := range fd.Triggers {
 		if t.From != ID {
 			continue
 		}
-		log.Println("Trigger is:", ID, "adding")
 		ret = append(ret, t)
 	}
 	return ret
 }
 
-func (fd *FlowDocument) fetchLinksTo(ID string) []Link {
+// FetchLinksTo fetch all links to node ID
+func (fd *FlowDocument) FetchLinksTo(ID string) []Link {
 	ret := []Link{}
 	for _, l := range fd.Links {
 		if l.To != ID {
@@ -63,7 +75,9 @@ func (fd *FlowDocument) fetchLinksTo(ID string) []Link {
 	}
 	return ret
 }
-func (fd *FlowDocument) fetchLinkTo(ID string, n int) *Link {
+
+// FetchLinkTo fetch a specific link to ID or return nil of none
+func (fd *FlowDocument) FetchLinkTo(ID string, n int) *Link {
 	for _, l := range fd.Links {
 		if l.To != ID || l.In != n {
 			continue
@@ -72,3 +86,13 @@ func (fd *FlowDocument) fetchLinkTo(ID string, n int) *Link {
 	}
 	return nil
 }
+
+// FetchNamedPortalIn fetch a portal in node, return nil of none
+func (fd *FlowDocument) FetchNamedPortalIn(name string) *Node {
+	for _, n := range fd.Nodes {
+		if n.Src == "Portal In" && n.Prop["portal name"] == name {
+			return &n
+		}
+	}
+	return nil
+}

go/src/flow/flowserver/session.go

@@ -40,7 +40,9 @@ type FlowSession struct {
 	RawDoc       []byte // Just share data
 	nodeActivity map[string]*NodeActivity
 
-	flow *flow.Flow
+	Data    map[interface{}]interface{}
+	flow    *flow.Flow
+	running bool
 }
 
 //NewSession creates and initializes a NewSession
@@ -69,6 +71,8 @@ func NewSession(fsm *FlowSessionManager, ID string) *FlowSession {
 		Chat:         ChatRoom{},
 		RawDoc:       rawDoc,
 		nodeActivity: map[string]*NodeActivity{},
+		// Experimental
+		Data: map[interface{}]interface{}{},
 
 		flow: nil,
 	}
@@ -197,15 +201,6 @@ func (s *FlowSession) NodeProcess(c *websocket.Conn, data []byte) error {
 		localR.Add("Log", func() io.Writer {
 			return s
 		})
-		// Special func
-		/*localR.Add("Variable", func(name string, initial flow.Data) flow.Data {
-			log.Println("Loading variable:", name)
-			_, ok := s.flow.Data[name]
-			if !ok {
-				s.flow.Data[name] = initial
-			}
-			return s.flow.Data[name]
-		})*/
 		localR.Add("Output", func(d interface{}) {
 
 			r := fmt.Sprint("Result:", d)
@@ -222,6 +217,11 @@ func (s *FlowSession) NodeProcess(c *websocket.Conn, data []byte) error {
 		s.flow = builder.Flow()
 		log.Println("Flow:", s.flow)
 
+		log.Println("Experimental: Loading data")
+		for k, v := range s.Data {
+			s.flow.Data.Store(k, v)
+		}
+
 		defer func() { // After routing gone
 			s.flow = nil
 		}()
@@ -277,11 +277,112 @@ func (s *FlowSession) NodeProcess(c *websocket.Conn, data []byte) error {
 			log.Println("Error operation", err)
 			return err
 		}
+		log.Println("Experimental storing data to session")
+		// Copy Data from flow
+		s.flow.Data.Range(func(k, v interface{}) bool {
+			s.Data[k] = v
+			return true
+		})
+		log.Println("Operation finish")
+		log.Println("Flow now:", s.flow)
+		return nil
+	}
+
+	go func() {
+		err := build()
+		if err != nil {
+			s.Notify(fmt.Sprint("ERR:", err))
+		}
+	}()
+
+	return nil
+}
+
+// NodeTrain temporary operation for repeating a node
+// this is for the demo purposes
+func (s *FlowSession) NodeTrain(c *websocket.Conn, data []byte) error {
+	ID := string(data[1 : len(data)-1]) // remove " instead of unmarshalling json
+	if s.flow != nil {
+		s.Notify("a node is already running")
+		return errors.New("node already running")
+	}
+
+	// Clear activity
+	s.nodeActivity = map[string]*NodeActivity{}
+	s.Broadcast(nil, s.activity()) // Ampty activity
+
+	build := func() error {
+		localR := s.manager.registry.Clone()
+		//Add our log func that is not in global registry
+		localR.Add("Notify", func(v flow.Data, msg string) flow.Data {
+			log.Println("Notify:", msg)
+			s.Notify(msg)
+			return v
+		})
+		localR.Add("Log", func() io.Writer {
+			return s
+		})
+		localR.Add("Output", func(d interface{}) {
+			r := fmt.Sprint("Result:", d)
+			// Do something
+			s.Notify(r)
+			s.Write([]byte(r))
+		})
+		builder := flowbuilder.New(localR)
+		builder.Load(s.RawDoc).Build(ID)
+		if builder.Err != nil {
+			return builder.Err
+		}
+
+		s.flow = builder.Flow()
+		log.Println("Flow:", s.flow)
+
+		// XXX: Possibly remove
+		log.Println("Experimental: Loading global data")
+		for k, v := range s.Data {
+			s.flow.Data.Store(k, v)
+		}
+
+		defer func() { // After routing gone
+			s.flow = nil
+		}()
+		// Flow activity
+
+		op, ok := builder.OperationMap[ID]
+		if !ok {
+			return fmt.Errorf("Operation not found %v", ID)
+		}
+		log.Println("Processing operation")
+
+		epochs := 5000
+		s.Notify(fmt.Sprintf("Training for %d epochs", epochs))
+		for i := 0; i < epochs; i++ {
+			res, err := op.Process()
+			if err != nil {
+				log.Println("Error operation", err)
+				return err
+			}
+			if i%1000 == 0 {
+				fmt.Fprintf(s, "Res: %v", res)
+				fmt.Fprintf(s, "Training... %d/%d", i, epochs)
+				outs := builder.Doc.FetchNodeBySrc("Output")
+				if len(outs) == 0 {
+					continue
+				}
+			}
+		}
+		fmt.Fprintf(s, "%v", s.flow)
+		// Copy Data from flow
+		s.flow.Data.Range(func(k, v interface{}) bool {
+			s.Data[k] = v
+			return true
+		})
 		log.Println("Operation finish")
 		log.Println("Flow now:", s.flow)
 		return nil
 	}
 
+	// Parallel building
 	go func() {
 		err := build()
 		if err != nil {
@@ -290,6 +391,7 @@ func (s *FlowSession) NodeProcess(c *websocket.Conn, data []byte) error {
 	}()
 
 	return nil
+
 }
 
 func (s *FlowSession) activity() *flowmsg.SendMessage {

go/src/flow/flowserver/sessionmgr.go

@@ -165,6 +165,11 @@ func (fsm *FlowSessionManager) ServeHTTP(w http.ResponseWriter, r *http.Request)
 					return errors.New("nodeRun: invalid session")
 				}
 				return sess.NodeProcess(c, m.Data)
+			case "nodeTrain":
+				if sess == nil {
+					return errors.New("nodeTrain: invalid session")
+				}
+				return sess.NodeTrain(c, m.Data)
 			////////////////////
 			// CHAT operations
 			/////////

go/src/flow/operation.go

@@ -102,6 +102,7 @@ func (f *Flow) SetVar(name string, data Data) Operation {
 		if err != nil {
 			return nil, err
 		}
+
 		f.Data.Store(name, res[0])
 		return res[0], nil
 	}