Home Explore Help
Sign In
stdio
/
Flow
1
0
Fork 0
Files Issues 0 Pull Requests 0 Wiki
Browse Source

Flow restructuring

luis 7 years ago
parent
d850a94a50
commit
737c654e43
22 changed files with 344 additions and 5738 deletions
Unified View Show Diff Stats
  1. 9 5273
      go/src/demos/cmd/demo1/assets/assets.go
  2. 5 5
      go/src/demos/cmd/demo1/main.go
  3. 0 40
      go/src/demos/cmd/demo1/static/c1.html
  4. 0 45
      go/src/demos/cmd/demo1/static/c2.html
  5. BIN
      go/src/demos/cmd/demo1/static/c2.jpg
  6. 0 67
      go/src/demos/cmd/demo1/static/c3.html
  7. BIN
      go/src/demos/cmd/demo1/static/c3.jpg
  8. 1 1
      go/src/demos/cmd/demo1/static/index.html
  9. 7 2
      go/src/demos/cmd/noui/flow.go
  10. 0 0
      go/src/demos/ops/defaultops/defaultops.go
  11. 0 0
      go/src/demos/ops/devops/devops.go
  12. 2 1
      go/src/demos/cmd/demo1/gonumops/gonumops.go
  13. 7 0
      go/src/flow/README.md
  14. 37 156
      go/src/flow/flow.go
  15. 5 3
      go/src/flow/flow_test.go
  16. 40 26
      go/src/flow/flowserver/flowbuilder/builder.go
  17. 5 3
      go/src/flow/flowserver/session.go
  18. 15 15
      go/src/flow/hook.go
  19. 125 94
      go/src/flow/operation.go
  20. 14 7
      go/src/flow/registry/registry.go
  21. 17 0
      go/src/flow/registry/registry_test.go
  22. 55 0
      go/src/flow/session.go

File diff suppressed because it is too large
+ 9 - 5273
go/src/demos/cmd/demo1/assets/assets.go


+ 5 - 5
go/src/demos/cmd/demo1/main.go
View File 

@@ -2,9 +2,9 @@ package main
 
 
 
 import (
 
 import (
 
 	"demos/cmd/demo1/assets"
 
 	"demos/cmd/demo1/assets"
 
-	"demos/cmd/demo1/defaultops"
 
-	"demos/cmd/demo1/devops"
 
-	"demos/cmd/demo1/gonumops"
 
+	"demos/ops/defaultops"
 
+	"demos/ops/devops"
 
+	"demos/ops/ml"
 
 	"flow/flowserver"
 
 	"flow/flowserver"
 
 	"log"
 
 	"log"
 
 	"mime"
 
 	"mime"
 
@@ -35,8 +35,8 @@ func main() {
 
 	mux.Handle("/default/", c.Build(
 
 	mux.Handle("/default/", c.Build(
 
 		http.StripPrefix("/default", flowserver.New(defaultops.New(), "default")),
 
 		http.StripPrefix("/default", flowserver.New(defaultops.New(), "default")),
 
 	))
 
 	))
 
-	mux.Handle("/gonumops/", c.Build(
 
-		http.StripPrefix("/gonumops", flowserver.New(gonumops.New(), "gonumops")),
 
+	mux.Handle("/machinelearning/", c.Build(
 
+		http.StripPrefix("/machinelearning", flowserver.New(ml.New(), "ml")),
 
 	))
 
 	))
 
 
 
 	mux.Handle("/devops/", c.Build(
 
 	mux.Handle("/devops/", c.Build(

+ 0 - 40
go/src/demos/cmd/demo1/static/c1.html
View File 

@@ -1,40 +0,0 @@
 
-<html>
 
-<head>
 
-<style type="text/css">
 
-pre { font-family: monospace; color: #d0d0d0; background-color: #121212; }
 
-* { font-size: 1em; }
 
-.Statement { color: #d7005f; font-weight: bold; }
 
-.String { color: #afaf87; }
 
-.Function { color: #87ff00; }
 
-.Operator { color: #d7005f; }
 
-.LineNr { color: #bcbcbc; background-color: #303030; padding-bottom: 1px; }
 
-.Keyword { color: #d7005f; font-weight: bold; }
 
-.Type { color: #5fd7ff; }
 
--->
 
-</style>
 
-
 
-</head>
 
-<body >
 
-<pre id='vimCodeElement'>
 
-<span id="L1" class="LineNr"> 1 </span><span class="Statement">package</span> main
 
-<span id="L2" class="LineNr"> 2 </span>
 
-<span id="L3" class="LineNr"> 3 </span><span class="Statement">import</span> (
 
-<span id="L4" class="LineNr"> 4 </span>  <span class="String">&quot;flow/registry&quot;</span>
 
-<span id="L5" class="LineNr"> 5 </span>  <span class="String">&quot;flowserver&quot;</span>
 
-<span id="L6" class="LineNr"> 6 </span>  <span class="String">&quot;net/http&quot;</span>
 
-<span id="L7" class="LineNr"> 7 </span>)
 
-<span id="L8" class="LineNr"> 8 </span>
 
-<span id="L9" class="LineNr"> 9 </span><span class="Keyword">func</span> <span class="Function">main</span>() {
 
-<span id="L10" class="LineNr">10 </span>
 
-<span id="L11" class="LineNr">11 </span>  r <span class="Operator">:=</span> registry.New()
 
-<span id="L12" class="LineNr">12 </span>  r.Register(<span class="String">&quot;hello&quot;</span>, <span class="Keyword">func</span>() <span class="Type">string</span> {
 
-<span id="L13" class="LineNr">13 </span>    <span class="Statement">return</span> <span class="String">&quot;hello world&quot;</span>
 
-<span id="L14" class="LineNr">14 </span>  })
 
-<span id="L15" class="LineNr">15 </span>
 
-<span id="L16" class="LineNr">16 </span>  http.ListenAndServe(<span class="String">&quot;:5000&quot;</span>, flowserver.New(r, <span class="String">&quot;storename&quot;</span>))
 
-<span id="L17" class="LineNr">17 </span>
 
-<span id="L18" class="LineNr">18 </span>}
 
-</pre>
 
-</body>
 
-</html>
 
-<!-- vim: set foldmethod=manual : -->

+ 0 - 45
go/src/demos/cmd/demo1/static/c2.html
View File 

@@ -1,45 +0,0 @@
 
-<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01//EN" "http://www.w3.org/TR/html4/strict.dtd">
 
-<html>
 
-<head>
 
-<style type="text/css">
 
-pre { font-family: monospace; color: #d0d0d0; background-color: #121212; }
 
-* { font-size: 1em; }
 
-.Statement { color: #d7005f; font-weight: bold; }
 
-.String { color: #afaf87; }
 
-.Function { color: #87ff00; }
 
-.Operator { color: #d7005f; }
 
-.LineNr { color: #bcbcbc; background-color: #303030; padding-bottom: 1px; }
 
-.Keyword { color: #d7005f; font-weight: bold; }
 
-.Type { color: #5fd7ff; }
 
-</style>
 
-</head>
 
-<body>
 
-
 
-<pre id='vimCodeElement'>
 
-<span id="L1" class="LineNr"> 1 </span><span class="Statement">package</span> main
 
-<span id="L2" class="LineNr"> 2 </span>
 
-<span id="L3" class="LineNr"> 3 </span><span class="Statement">import</span> (
 
-<span id="L4" class="LineNr"> 4 </span>  <span class="String">&quot;flow/registry&quot;</span>
 
-<span id="L5" class="LineNr"> 5 </span>  <span class="String">&quot;flowserver&quot;</span>
 
-<span id="L6" class="LineNr"> 6 </span>  <span class="String">&quot;net/http&quot;</span>
 
-<span id="L7" class="LineNr"> 7 </span>  <span class="String">&quot;strings&quot;</span>
 
-<span id="L8" class="LineNr"> 8 </span>)
 
-<span id="L9" class="LineNr"> 9 </span>
 
-<span id="L10" class="LineNr">10 </span><span class="Keyword">func</span> <span class="Function">main</span>() {
 
-<span id="L11" class="LineNr">11 </span>
 
-<span id="L12" class="LineNr">12 </span>  r <span class="Operator">:=</span> registry.New()
 
-<span id="L13" class="LineNr">13 </span>  r.Register(<span class="String">&quot;hello&quot;</span>, <span class="Keyword">func</span>() <span class="Type">string</span> {
 
-<span id="L14" class="LineNr">14 </span>    <span class="Statement">return</span> <span class="String">&quot;hello world&quot;</span>
 
-<span id="L15" class="LineNr">15 </span>  })
 
-<span id="L16" class="LineNr">16 </span>  r.Add(strings.Split, strings.Join)
 
-<span id="L17" class="LineNr">17 </span>
 
-<span id="L18" class="LineNr">18 </span>  http.ListenAndServe(<span class="String">&quot;:5000&quot;</span>, flowserver.New(r, <span class="String">&quot;storename&quot;</span>))
 
-<span id="L19" class="LineNr">19 </span>
 
-<span id="L20" class="LineNr">20 </span>}
 
-</pre>
 
-<img src="c2.jpg">
 
-
 
-</body>
 
-</html>
 
-
 
-<!-- vim: set foldmethod=manual : -->

BIN
go/src/demos/cmd/demo1/static/c2.jpg
View File


+ 0 - 67
go/src/demos/cmd/demo1/static/c3.html
View File 

@@ -1,67 +0,0 @@
 
-<html>
 
-<head>
 
-<style type="text/css">
 
-<!--
 
-pre { font-family: monospace; color: #d0d0d0; background-color: #121212; }
 
-* { font-size: 1em; }
 
-.Statement { color: #d7005f; font-weight: bold; }
 
-.String { color: #afaf87; }
 
-.Conditional { color: #d7005f; font-weight: bold; }
 
-.Function { color: #87ff00; }
 
-.Boolean { color: #af5fff; }
 
-.Operator { color: #d7005f; }
 
-.LineNr { color: #bcbcbc; background-color: #303030; padding-bottom: 1px; }
 
-.Keyword { color: #d7005f; font-weight: bold; }
 
-.Comment { color: #5f5f5f; }
 
-.Type { color: #5fd7ff; }
 
--->
 
-</style>
 
-</head>
 
-<pre id='vimCodeElement'>
 
-<span id="L1" class="LineNr"> 1 </span><span class="Statement">package</span> main
 
-<span id="L2" class="LineNr"> 2 </span>
 
-<span id="L3" class="LineNr"> 3 </span><span class="Statement">import</span> (
 
-<span id="L4" class="LineNr"> 4 </span>  <span class="String">&quot;flow/registry&quot;</span>
 
-<span id="L5" class="LineNr"> 5 </span>  <span class="String">&quot;flowserver&quot;</span>
 
-<span id="L6" class="LineNr"> 6 </span>  <span class="String">&quot;io/ioutil&quot;</span>
 
-<span id="L7" class="LineNr"> 7 </span>  <span class="String">&quot;log&quot;</span>
 
-<span id="L8" class="LineNr"> 8 </span>  <span class="String">&quot;math&quot;</span>
 
-<span id="L9" class="LineNr"> 9 </span>  <span class="String">&quot;net/http&quot;</span>
 
-<span id="L10" class="LineNr">10 </span>  <span class="String">&quot;strings&quot;</span>
 
-<span id="L11" class="LineNr">11 </span>)
 
-<span id="L12" class="LineNr">12 </span>
 
-<span id="L13" class="LineNr">13 </span><span class="Keyword">func</span> <span class="Function">main</span>() {
 
-<span id="L14" class="LineNr">14 </span>
 
-<span id="L15" class="LineNr">15 </span>  r <span class="Operator">:=</span> registry.New()
 
-<span id="L16" class="LineNr">16 </span>  r.Register(<span class="String">&quot;hello&quot;</span>, <span class="Keyword">func</span>() <span class="Type">string</span> {
 
-<span id="L17" class="LineNr">17 </span>    <span class="Statement">return</span> <span class="String">&quot;hello world&quot;</span>
 
-<span id="L18" class="LineNr">18 </span>  })
 
-<span id="L19" class="LineNr">19 </span>  r.Add(strings.Split, strings.Join)
 
-<span id="L20" class="LineNr">20 </span>
 
-<span id="L21" class="LineNr">21 </span>  fns, err <span class="Operator">:=</span> r.Add(math.Exp, math.Sin, math.Cos)
 
-<span id="L22" class="LineNr">22 </span>  <span class="Conditional">if</span> err <span class="Operator">!=</span> <span class="Boolean">nil</span> {
 
-<span id="L23" class="LineNr">23 </span>    log.Fatal(err)
 
-<span id="L24" class="LineNr">24 </span>  }
 
-<span id="L25" class="LineNr">25 </span><span class="Comment">  //Describing groups of functions</span>
 
-<span id="L26" class="LineNr">26 </span>  fns.Tags(<span class="String">&quot;math&quot;</span>)
 
-<span id="L27" class="LineNr">27 </span>  fns.Extra(<span class="String">&quot;style&quot;</span>, registry.M{<span class="String">&quot;color&quot;</span>: <span class="String">&quot;#F00&quot;</span>})
 
-<span id="L28" class="LineNr">28 </span>
 
-<span id="L29" class="LineNr">29 </span><span class="Comment">  // Describing a function</span>
 
-<span id="L30" class="LineNr">30 </span>  registry.Describer(
 
-<span id="L31" class="LineNr">31 </span>    r.MustAdd(http.Get).Inputs(<span class="String">&quot;url&quot;</span>),
 
-<span id="L32" class="LineNr">32 </span>    r.MustAdd(responseReader).Inputs(<span class="String">&quot;response&quot;</span>),
 
-<span id="L33" class="LineNr">33 </span>  ).Tags(<span class="String">&quot;http&quot;</span>).Extra(<span class="String">&quot;style&quot;</span>, registry.M{<span class="String">&quot;color&quot;</span>: <span class="String">&quot;#00F&quot;</span>})
 
-<span id="L34" class="LineNr">34 </span>
 
-<span id="L35" class="LineNr">35 </span>  http.ListenAndServe(<span class="String">&quot;:5000&quot;</span>, flowserver.New(r, <span class="String">&quot;storename&quot;</span>))
 
-<span id="L36" class="LineNr">36 </span>
 
-<span id="L37" class="LineNr">37 </span>}
 
-<span id="L38" class="LineNr">38 </span><span class="Keyword">func</span> <span class="Function">responseReader</span>(r <span class="Operator">*</span>http.Response) (<span class="Type">string</span>, <span class="Type">error</span>) {
 
-<span id="L39" class="LineNr">39 </span>  <span class="Statement">defer</span> r.Body.Close()
 
-<span id="L40" class="LineNr">40 </span>  res, err <span class="Operator">:=</span> ioutil.ReadAll(r.Body)
 
-<span id="L41" class="LineNr">41 </span>  <span class="Statement">return</span> <span class="Type">string</span>(res), err
 
-<span id="L42" class="LineNr">42 </span>}
 
-</pre>
 
-<img src="c3.jpg">
 
-</body>
 
-</html
 
-

BIN
go/src/demos/cmd/demo1/static/c3.jpg
View File


+ 1 - 1
go/src/demos/cmd/demo1/static/index.html
View File 

@@ -20,7 +20,7 @@
 
     <h3> Sample ideas using flow: </h3>
 
     <h3> Sample ideas using flow: </h3>
 
     <div class="app-selector-items">
 
     <div class="app-selector-items">
 
 			[<a href="/default">Default thing</a>]
 
 			[<a href="/default">Default thing</a>]
 
-			[<a href="/gonumops">Gonum</a>]
 
+			[<a href="/machinelearning">Machine Learning</a>]
 
 			[<a href="/devops">Devops sample</a>]
 
 			[<a href="/devops">Devops sample</a>]
 
 			[<a href="/testops">testing registry</a>]
 
 			[<a href="/testops">testing registry</a>]
 
     </div>
 
     </div>

+ 7 - 2
go/src/demos/cmd/noui/flow.go
View File 

@@ -4,6 +4,7 @@ import (
 
 	"flow"
 
 	"flow"
 
 	"flow/registry"
 
 	"flow/registry"
 
 	"log"
 
 	"log"
 
+	"os"
 
 	"strings"
 
 	"strings"
 
 )
 
 )
 
 
 
@@ -23,11 +24,15 @@ func main() {
 
 	// Find a way to process this
 
 	// Find a way to process this
 
 	v := f.Var("x", first)
 
 	v := f.Var("x", first)
 
 
 
-	opSplit := f.Op("Split", v, " ")
 
+	res, err := v.Process()
 
+	log.Println("P:", res, err)
 
 
 
+	opSplit := f.Op("Split", v, " ")
 
 	opJoin := f.Op("Join", opSplit, ",")
 
 	opJoin := f.Op("Join", opSplit, ",")
 
 
 
-	res, err := opJoin.Process()
 
+	log.Println("Flow:", f)
 
+	f.Analyse(os.Stdout)
 
+	res, err = opJoin.Process()
 
 	if err != nil {
 
 	if err != nil {
 
 		log.Fatal(err)
 
 		log.Fatal(err)
 
 	}
 
 	}

go/src/demos/cmd/demo1/defaultops/defaultops.go → go/src/demos/ops/defaultops/defaultops.go
View File


go/src/demos/cmd/demo1/devops/devops.go → go/src/demos/ops/devops/devops.go
View File


+ 2 - 1
go/src/demos/cmd/demo1/gonumops/gonumops.go
View File 

@@ -1,4 +1,5 @@
 
-package gonumops
 
+// Package ml machine learning operations for flow
 
+package ml
 
 
 
 import (
 
 import (
 
 	"flow"
 
 	"flow"

+ 7 - 0
go/src/flow/README.md
View File 

@@ -1,5 +1,12 @@
 
 # Extensible Flow Engine
 
 # Extensible Flow Engine
 
 
 
+## TODO
 
+
 
+### Session
 
+
 
+Importance of session, is to be able to serialize different sets of data with
 
+different internal values, right now variables are being stored in session
 
+
 
 ## Features
 
 ## Features
 
 
 
 * Develop new Operators
 
 * Develop new Operators

+ 37 - 156
go/src/flow/flow.go
View File 

@@ -2,13 +2,10 @@ package flow
 
 
 
 import (
 
 import (
 
 	"bytes"
 
 	"bytes"
 
-	"encoding/json"
 
-	"errors"
 
 	"flow/registry"
 
 	"flow/registry"
 
 	"fmt"
 
 	"fmt"
 
 	"io"
 
 	"io"
 
 	"os"
 
 	"os"
 
-	"sync"
 
 )
 
 )
 
 
 
 // Data interface
 
 // Data interface
 
@@ -18,14 +15,11 @@ type Data = interface{}
 
 // We could Create a single array of operations
 
 // We could Create a single array of operations
 
 // refs would only mean id, types would be embed in operation
 
 // refs would only mean id, types would be embed in operation
 
 type Flow struct {
 
 type Flow struct {
 
-	sync.Mutex
 
-
 
+	//sync.Mutex // Needed?
 
 	registry   *registry.R
 
 	registry   *registry.R
 
-	idGen      func() string
 
-	consts     map[string]Data
 
 	Data       map[string]Data // Should be named, to fetch later
 
 	Data       map[string]Data // Should be named, to fetch later
 
-	operations sync.Map
 
-	runID      int
 
+	consts     []Data
 
+	operations []*operation
 
 
 
 	// Experimental run Event
 
 	// Experimental run Event
 
 	hooks Hooks
 
 	hooks Hooks
 
@@ -34,13 +28,10 @@ type Flow struct {
 
 // New create a new flow
 
 // New create a new flow
 
 func New() *Flow {
 
 func New() *Flow {
 
 	return &Flow{
 
 	return &Flow{
 
-		registry: registry.Global,
 
-		idGen:    func() string { return RandString(8) },
 
-		// Data
 
-		consts:     map[string]Data{},
 
+		registry:   registry.Global,
 
 		Data:       map[string]Data{},
 
 		Data:       map[string]Data{},
 
-		operations: sync.Map{},
 
-		//map[string]opEntry{},
 
+		operations: []*operation{},
 
+		consts:     []Data{},
 
 	}
 
 	}
 
 }
 
 }
 
 
 
@@ -51,90 +42,17 @@ func (f *Flow) UseRegistry(r *registry.R) *Flow {
 
 	return f
 
 	return f
 
 }
 
 }
 
 
 
-//SetIDGen set the id generator that will generate
 
-//ID for new nodes
 
-func (f *Flow) SetIDGen(idGen func() string) {
 
-	f.idGen = idGen
 
-}
 
-
 
-// Auto ID generation
 
-
 
-// Op return an function operator
 
-//  name - a previous registered function
 
-//  params - the function inputs
 
-func (f *Flow) Op(name string, params ...interface{}) Operation {
 
-	var op Operation
 
-	var err error
 
-	allocErr := f.allocID(func(id string) error {
 
-		op, err = f.DefOp(id, name, params...)
 
-		if err != nil {
 
-			op = f.DefErrOp(id, err)
 
-		}
 
-		// Err op
 
-		return err
 
-	})
 
-	if allocErr != nil {
 
-		return nil
 
-	}
 
-	return op
 
-}
 
-
 
-// ErrOp error operation with generated ID
 
-func (f *Flow) ErrOp(operr error) Operation {
 
-	var op Operation
 
-
 
-	f.allocID(func(id string) error {
 
-		op = f.DefErrOp(id, operr)
 
-		return nil
 
-	})
 
-	return op
 
-}
 
-
 
-// Const returns a const operation with generated ID
 
-func (f *Flow) Const(value Data) Operation {
 
-	var op Operation
 
-	f.allocID(func(id string) error {
 
-		op = f.DefConst(id, value)
 
-		return nil
 
-	})
 
-	return op
 
-}
 
-
 
-// Var operation
 
-func (f *Flow) Var(name string, initial ...Data) Operation {
 
-	var op Operation
 
-	err := f.allocID(func(id string) error {
 
-		op = f.DefVar(id, name, initial...)
 
-		return nil
 
-	})
 
-	if err != nil {
 
-		return nil
 
-	}
 
-	return op
 
-}
 
-
 
-// In input operation
 
-func (f *Flow) In(paramID int) Operation {
 
-	var op Operation
 
-	f.allocID(func(id string) error {
 
-		op = f.DefIn(id, paramID)
 
-		return nil
 
-	})
 
-	return op
 
-
 
-}
 
-
 
 // Analyse every operations
 
 // Analyse every operations
 
 func (f *Flow) Analyse(w io.Writer, params ...Data) {
 
 func (f *Flow) Analyse(w io.Writer, params ...Data) {
 
 	if w == nil {
 
 	if w == nil {
 
 		w = os.Stdout
 
 		w = os.Stdout
 
 	}
 
 	}
 
 	fmt.Fprintf(w, "Ops analysis:\n")
 
 	fmt.Fprintf(w, "Ops analysis:\n")
 
-	f.operations.Range(func(pk, po interface{}) bool {
 
-		k, op := pk.(string), po.(*operation)
 
+
 
+	for k, op := range f.operations {
 
 		fw := bytes.NewBuffer(nil)
 
 		fw := bytes.NewBuffer(nil)
 
 		//fmt.Fprintf(w, "  [%s] (%v)", k, op.name)
 
 		//fmt.Fprintf(w, "  [%s] (%v)", k, op.name)
 
-		fmt.Fprintf(fw, "  [%s] %s(", pk, op.name)
 
+		fmt.Fprintf(fw, "  [%v] %s(", k, op.name)
 
 		for j, in := range op.inputs {
 
 		for j, in := range op.inputs {
 
 			//ref := in.(Op)
 
 			//ref := in.(Op)
 
 			if j != 0 {
 
 			if j != 0 {
 
@@ -143,23 +61,21 @@ func (f *Flow) Analyse(w io.Writer, params ...Data) {
 
 			ires, err := in.Process(params...)
 
 			ires, err := in.Process(params...)
 
 			if err != nil {
 
 			if err != nil {
 
 				fmt.Fprintf(w, "Operator: %s error#%s\n", op.name, err)
 
 				fmt.Fprintf(w, "Operator: %s error#%s\n", op.name, err)
 
-				return false
 
+				break
 
 			}
 
 			}
 
-			fmt.Fprintf(fw, " %s[%v](%v)", op.kind, op.id, ires)
 
+			fmt.Fprintf(fw, " %s[%v](%v)", op.kind, k, ires)
 
 		}
 
 		}
 
 		fmt.Fprintf(fw, ") - ")
 
 		fmt.Fprintf(fw, ") - ")
 
 		// Create OpProcessor and execute
 
 		// Create OpProcessor and execute
 
 		//
 
 		//
 
-		opfn := f.GetOp(k)
 
-		res, err := opfn.Process(params...)
 
+		res, err := op.Process(params...)
 
 		if err != nil {
 
 		if err != nil {
 
 			fmt.Fprintf(fw, "ERR\n")
 
 			fmt.Fprintf(fw, "ERR\n")
 
 		}
 
 		}
 
 		fmt.Fprintf(fw, "%v\n", res)
 
 		fmt.Fprintf(fw, "%v\n", res)
 
 
 
 		fmt.Fprintf(w, "%s", fw.String())
 
 		fmt.Fprintf(w, "%s", fw.String())
 
-		return true
 
-	})
 
+	}
 
 }
 
 }
 
 
 
 /////////////////////////////
 
 /////////////////////////////
 
@@ -172,7 +88,7 @@ func (f *Flow) String() string {
 
 	// consts
 
 	// consts
 
 	fmt.Fprintf(ret, "consts:\n")
 
 	fmt.Fprintf(ret, "consts:\n")
 
 	for i, v := range f.consts {
 
 	for i, v := range f.consts {
 
-		fmt.Fprintf(ret, "  [%s] %v\n", i, v)
 
+		fmt.Fprintf(ret, "  [%v] %v\n", i, v)
 
 	}
 
 	}
 
 	fmt.Fprintf(ret, "data:\n") // Or variable
 
 	fmt.Fprintf(ret, "data:\n") // Or variable
 
 	for k, v := range f.Data {
 
 	for k, v := range f.Data {
 
@@ -180,94 +96,59 @@ func (f *Flow) String() string {
 
 	}
 
 	}
 
 
 
 	fmt.Fprintf(ret, "operations:\n")
 
 	fmt.Fprintf(ret, "operations:\n")
 
-	f.operations.Range(func(pk, pv interface{}) bool {
 
-		k, v := pk.(string), pv.(*operation)
 
-
 
-		fmt.Fprintf(ret, "  [%s] %s(", k, v.name)
 
-		for j, in := range v.inputs {
 
+	for k, op := range f.operations {
 
+		fmt.Fprintf(ret, "  [%v] %s(", k, op.name)
 
+		for j, in := range op.inputs {
 
 			if j != 0 {
 
 			if j != 0 {
 
 				fmt.Fprintf(ret, ", ")
 
 				fmt.Fprintf(ret, ", ")
 
 			}
 
 			}
 
-			fmt.Fprintf(ret, "%s[%v]", "func", in.ID())
 
+			if in.kind == "const" {
 
+				v, _ := in.Process()
 
+				fmt.Fprintf(ret, "%s[%v](%v)", in.kind, j, v)
 
+				continue
 
+			}
 
+			// Find operation index
 
+			for t := range f.operations {
 
+				if f.operations[t] == in {
 
+					fmt.Fprintf(ret, "%s[%v]", in.kind, t)
 
+					break
 
+				}
 
+			}
 
 		}
 
 		}
 
 		fmt.Fprintf(ret, ")\n")
 
 		fmt.Fprintf(ret, ")\n")
 
-		return true
 
-	})
 
+	}
 
 
 
 	return ret.String()
 
 	return ret.String()
 
 }
 
 }
 
 
 
 // MarshalJSON implementation
 
 // MarshalJSON implementation
 
-func (f *Flow) MarshalJSON() ([]byte, error) {
 
+/*func (f *Flow) MarshalJSON() ([]byte, error) {
 
 	data := map[string]interface{}{}
 
 	data := map[string]interface{}{}
 
 	type opMarshal struct {
 
 	type opMarshal struct {
 
 		Name  string
 
 		Name  string
 
 		Input []map[string]interface{}
 
 		Input []map[string]interface{}
 
 	}
 
 	}
 
-	operations := map[string]opMarshal{}
 
-	f.operations.Range(func(pk, po interface{}) bool {
 
-		k, o := pk.(string), po.(*operation)
 
+	operations := map[int]opMarshal{}
 
+	for k, op := range f.operations {
 
 		refs := []map[string]interface{}{}
 
 		refs := []map[string]interface{}{}
 
-		for _, in := range o.inputs { // Switch type?
 
+		for _, in := range op.inputs { // Switch type?
 
 			refs = append(refs, map[string]interface{}{
 
 			refs = append(refs, map[string]interface{}{
 
 				"type": in.kind,
 
 				"type": in.kind,
 
 				"id":   in.id,
 
 				"id":   in.id,
 
 			})
 
 			})
 
 		}
 
 		}
 
 		operations[k] = opMarshal{o.name, refs}
 
 		operations[k] = opMarshal{o.name, refs}
 
-		return true
 
-	})
 
+	}
 
 
 
 	data["operations"] = operations
 
 	data["operations"] = operations
 
 	data["data"] = f.Data
 
 	data["data"] = f.Data
 
 	data["consts"] = f.consts
 
 	data["consts"] = f.consts
 
 
 
 	return json.Marshal(data)
 
 	return json.Marshal(data)
 
-}
 
-
 
-func (f *Flow) allocID(fn func(id string) error) error {
 
-
 
-	genID := func() (string, error) {
 
-		f.Lock()
 
-		defer f.Unlock()
 
-
 
-		var id string
 
-		// generate ID
 
-		for i := 0; i < 10; i++ {
 
-			id = f.idGen()
 
-			// Load or store
 
-			if _, ok := f.operations.LoadOrStore(id, nil); !ok {
 
-				return id, nil
 
-			}
 
-		}
 
-		return "", errors.New("ID Exausted")
 
-	}
 
-
 
-	// Safe generate an ID
 
-	id, err := genID()
 
-	if err != nil {
 
-		return err
 
-	}
 
-
 
-	err = fn(id)
 
-	if err != nil {
 
-		f.operations.Delete(id)
 
-	}
 
-	return nil
 
-
 
-}
 
-
 
-// GetOp Return an existing operation or return notfound error
 
-func (f *Flow) GetOp(id string) *operation {
 
-	op, ok := f.operations.Load(id)
 
-	if !ok {
 
-		return nil
 
-	}
 
-	return op.(*operation)
 
-}
 
+}*/
 
 
 
 //////////////////////////////////////////////
 
 //////////////////////////////////////////////
 
-// Experimental event
 
+// Experimental event hooks
 
 ////////////////
 
 ////////////////
 
 
 
 // Hook attach the node event hooks
 
 // Hook attach the node event hooks

+ 5 - 3
go/src/flow/flow_test.go
View File 

@@ -73,7 +73,7 @@ func TestGetOp(t *testing.T) {
 
 
 
 }
 
 }
 
 
 
-func TestIDGen(t *testing.T) {
 
+/*func TestIDGen(t *testing.T) {
 
 	a := assert.A(t)
 
 	a := assert.A(t)
 
 	idTable := []string{"2", "1", "1"}
 
 	idTable := []string{"2", "1", "1"}
 
 
 
@@ -102,7 +102,7 @@ func TestIDGen(t *testing.T) {
 
 	o = f.Op("vecadd", i1, i2)
 
 	o = f.Op("vecadd", i1, i2)
 
 	a.Eq(o, nil, "Should be nil, id generation exausted")
 
 	a.Eq(o, nil, "Should be nil, id generation exausted")
 
 
 
-}
 
+}*/
 
 
 
 func TestSerialize(t *testing.T) {
 
 func TestSerialize(t *testing.T) {
 
 	// Does not text yet
 
 	// Does not text yet
 
@@ -169,6 +169,8 @@ func TestOp(t *testing.T) {
 
 	a.Eq(test, res)
 
 	a.Eq(test, res)
 
 }
 
 }
 
 
 
+/*
 
+* TODO: Create variable test
 
 func TestVariable(t *testing.T) {
 
 func TestVariable(t *testing.T) {
 
 	a := assert.A(t)
 
 	a := assert.A(t)
 
 	f := flow.New()
 
 	f := flow.New()
 
@@ -182,7 +184,7 @@ func TestVariable(t *testing.T) {
 
 	res, err = v.Process()
 
 	res, err = v.Process()
 
 	a.Eq(err, nil)
 
 	a.Eq(err, nil)
 
 	a.Eq(res, 2)
 
 	a.Eq(res, 2)
 
-}
 
+}*/
 
 
 
 // Test context
 
 // Test context
 
 func TestCache(t *testing.T) {
 
 func TestCache(t *testing.T) {

+ 40 - 26
go/src/flow/flowserver/flowbuilder/builder.go
View File 

@@ -17,21 +17,33 @@ var ErrLoop = errors.New("Looping is disabled for now")
 
 
 
 // FlowBuilder builds a flow from flow-ui json data
 
 // FlowBuilder builds a flow from flow-ui json data
 
 type FlowBuilder struct {
 
 type FlowBuilder struct {
 
-	registry  *registry.R
 
-	doc       *FlowDocument
 
-	flow      *flow.Flow
 
-	nodeTrack map[string]bool
 
-	Err       error
 
+	registry     *registry.R
 
+	doc          *FlowDocument
 
+	flow         *flow.Flow
 
+	OperationMap map[string]flow.Operation
 
+	nodeTrack    map[string]bool
 
+	Err          error
 
 }
 
 }
 
 
 
 // New creates a New builder
 
 // New creates a New builder
 
 func New(r *registry.R) *FlowBuilder {
 
 func New(r *registry.R) *FlowBuilder {
 
 	return &FlowBuilder{
 
 	return &FlowBuilder{
 
-		registry:  r,
 
-		nodeTrack: map[string]bool{},
 
+		registry:     r,
 
+		OperationMap: map[string]flow.Operation{},
 
+		nodeTrack:    map[string]bool{},
 
 	}
 
 	}
 
 }
 
 }
 
 
 
+// GetOpID fetches operation ID
 
+func (fb *FlowBuilder) GetOpID(op flow.Operation) string {
 
+	for k, v := range fb.OperationMap {
 
+		if op == v {
 
+			return k
 
+		}
 
+	}
 
+	return ""
 
+}
 
+
 
 // Load document from json into builder
 
 // Load document from json into builder
 
 func (fb *FlowBuilder) Load(rawData []byte) *FlowBuilder {
 
 func (fb *FlowBuilder) Load(rawData []byte) *FlowBuilder {
 
 	fb.flow = flow.New()
 
 	fb.flow = flow.New()
 
@@ -53,7 +65,7 @@ func (fb *FlowBuilder) Load(rawData []byte) *FlowBuilder {
 
 // Build a flow starting from node
 
 // Build a flow starting from node
 
 func (fb *FlowBuilder) Build(ID string) flow.Operation {
 
 func (fb *FlowBuilder) Build(ID string) flow.Operation {
 
 	if fb.Err != nil {
 
 	if fb.Err != nil {
 
-		op := fb.flow.DefErrOp(ID, fb.Err)
 
+		op := fb.flow.ErrOp(fb.Err)
 
 		return op
 
 		return op
 
 	}
 
 	}
 
 	f := fb.flow
 
 	f := fb.flow
 
@@ -62,40 +74,45 @@ func (fb *FlowBuilder) Build(ID string) flow.Operation {
 
 
 
 	if _, ok := fb.nodeTrack[ID]; ok {
 
 	if _, ok := fb.nodeTrack[ID]; ok {
 
 		fb.Err = ErrLoop //fmt.Errorf("[%v] Looping through nodes is disabled:", ID)
 
 		fb.Err = ErrLoop //fmt.Errorf("[%v] Looping through nodes is disabled:", ID)
 
-		op := fb.flow.DefErrOp(ID, fb.Err)
 
+		op := fb.flow.ErrOp(fb.Err)
 
 		return op
 
 		return op
 
 	}
 
 	}
 
+	// loop detector
 
 	fb.nodeTrack[ID] = true
 
 	fb.nodeTrack[ID] = true
 
 	defer delete(fb.nodeTrack, ID)
 
 	defer delete(fb.nodeTrack, ID)
 
 
 
 	// If flow already has ID just return
 
 	// If flow already has ID just return
 
-	if op := f.GetOp(ID); op != nil {
 
+	if op, ok := fb.OperationMap[ID]; ok {
 
 		return op
 
 		return op
 
 	}
 
 	}
 
 
 
 	node := fb.doc.fetchNodeByID(ID)
 
 	node := fb.doc.fetchNodeByID(ID)
 
 	if node == nil {
 
 	if node == nil {
 
-		op := fb.flow.DefErrOp(ID, fmt.Errorf("node not found [%v]", ID))
 
+		op := fb.flow.ErrOp(fmt.Errorf("node not found [%v]", ID))
 
 		return op
 
 		return op
 
 	}
 
 	}
 
 
 
 	var op flow.Operation
 
 	var op flow.Operation
 
-	var err error
 
 	var inputs []reflect.Type
 
 	var inputs []reflect.Type
 
 
 
 	switch node.Src {
 
 	switch node.Src {
 
 	case "Input":
 
 	case "Input":
 
 		inputID, err := strconv.Atoi(node.Prop["input"])
 
 		inputID, err := strconv.Atoi(node.Prop["input"])
 
 		if err != nil {
 
 		if err != nil {
 
-			return f.DefErrOp(node.ID, errors.New("Invalid inputID value, must be a number"))
 
+			op := f.ErrOp(errors.New("Invalid inputID value, must be a number"))
 
+			fb.OperationMap[node.ID] = op
 
+			return op
 
 		}
 
 		}
 
-		return f.In(inputID) // By id perhaps
 
+		op := f.In(inputID) // By id perhaps
 
+		fb.OperationMap[node.ID] = op
 
+		return op
 
 	default:
 
 	default:
 
 		log.Println("Loading entry:", node.Src)
 
 		log.Println("Loading entry:", node.Src)
 
 		entry, err := r.Entry(node.Src)
 
 		entry, err := r.Entry(node.Src)
 
 		if err != nil {
 
 		if err != nil {
 
-			op = f.DefErrOp(node.ID, err)
 
-			return nil
 
+			op = f.ErrOp(err)
 
+			fb.OperationMap[node.ID] = op
 
+			return op
 
 		}
 
 		}
 
 		inputs = entry.Inputs
 
 		inputs = entry.Inputs
 
 
 
@@ -135,29 +152,26 @@ func (fb *FlowBuilder) Build(ID string) flow.Operation {
 
 		param[i] = fb.Build(l.From)
 
 		param[i] = fb.Build(l.From)
 
 	}
 
 	}
 
 
 
-	op, err = f.DefOp(node.ID, node.Src, param...)
 
-	if err != nil {
 
-		op := f.DefErrOp(node.ID, err)
 
-		return op
 
-	}
 
-	fb.addTriggersTo(node)
 
+	op = f.Op(node.Src, param...)
 
+	fb.OperationMap[node.ID] = op
 
+	fb.addTriggersTo(node, op)
 
 
 
 	return op
 
 	return op
 
 }
 
 }
 
 
 
-func (fb *FlowBuilder) addTriggersTo(node *Node) error {
 
+func (fb *FlowBuilder) addTriggersTo(node *Node, op flow.Operation) error {
 
 	// Process triggers for this node
 
 	// Process triggers for this node
 
 	triggers := fb.doc.fetchTriggerFrom(node.ID)
 
 	triggers := fb.doc.fetchTriggerFrom(node.ID)
 
 	for _, t := range triggers {
 
 	for _, t := range triggers {
 
 		op := fb.Build(t.To)
 
 		op := fb.Build(t.To)
 
 		// Register the thing here
 
 		// Register the thing here
 
 		fb.flow.Hook(flow.Hook{
 
 		fb.flow.Hook(flow.Hook{
 
-			Any: func(name string, ID string, triggerTime time.Time, extra ...interface{}) {
 
+			Any: func(name string, triggerOp flow.Operation, triggerTime time.Time, extra ...interface{}) {
 
 				if name != "Error" && name != "Finish" {
 
 				if name != "Error" && name != "Finish" {
 
 					return
 
 					return
 
 				}
 
 				}
 
-				if ID != t.From {
 
-					log.Printf("ID[%v] triggered [%v], I'm t.From: %v", ID, name, t.From)
 
+				if triggerOp != op {
 
+					log.Printf("ID triggered [%v], I'm t.From: %v", name, t.From)
 
 					return
 
 					return
 
 				}
 
 				}
 
 				exec := false
 
 				exec := false

+ 5 - 3
go/src/flow/flowserver/session.go
View File 

@@ -225,10 +225,12 @@ func (s *FlowSession) NodeProcess(c *websocket.Conn, data []byte) error {
 
 		defer func() { // After routing gone
 
 		defer func() { // After routing gone
 
 			s.flow = nil
 
 			s.flow = nil
 
 		}()
 
 		}()
 
+		// Flow activity
 
 		s.flow.Hook(flow.Hook{
 
 		s.flow.Hook(flow.Hook{
 
-			Any: func(name string, nodeID string, triggerTime time.Time, extra ...flow.Data) {
 
+			Any: func(name string, triggerOp flow.Operation, triggerTime time.Time, extra ...flow.Data) {
 
 				s.Lock()
 
 				s.Lock()
 
 				defer s.Unlock()
 
 				defer s.Unlock()
 
+				nodeID := builder.GetOpID(triggerOp)
 
 
 
 				act, ok := s.nodeActivity[nodeID]
 
 				act, ok := s.nodeActivity[nodeID]
 
 				if !ok {
 
 				if !ok {
 
@@ -265,8 +267,8 @@ func (s *FlowSession) NodeProcess(c *websocket.Conn, data []byte) error {
 
 			},
 
 			},
 
 		})
 
 		})
 
 
 
-		op := s.flow.GetOp(ID)
 
-		if op == nil {
 
+		op, ok := builder.OperationMap[ID]
 
+		if !ok {
 
 			return fmt.Errorf("Operation not found %v", ID)
 
 			return fmt.Errorf("Operation not found %v", ID)
 
 		}
 
 		}
 
 		log.Println("Processing operation")
 
 		log.Println("Processing operation")

+ 15 - 15
go/src/flow/hook.go
View File 

@@ -13,47 +13,47 @@ type Hooks struct {
 
 
 
 // Hook funcs to handle certain events on the flow
 
 // Hook funcs to handle certain events on the flow
 
 type Hook struct {
 
 type Hook struct {
 
-	Wait   func(ID string, triggerTime time.Time)
 
-	Start  func(ID string, triggerTime time.Time)
 
-	Finish func(ID string, triggerTime time.Time, res interface{})
 
-	Error  func(ID string, triggerTime time.Time, err error)
 
-	Any    func(name string, ID string, triggerTime time.Time, extra ...interface{})
 
+	Wait   func(op Operation, triggerTime time.Time)
 
+	Start  func(op Operation, triggerTime time.Time)
 
+	Finish func(op Operation, triggerTime time.Time, res interface{})
 
+	Error  func(op Operation, triggerTime time.Time, err error)
 
+	Any    func(name string, op Operation, triggerTime time.Time, extra ...interface{})
 
 }
 
 }
 
 
 
 // Trigger a hook
 
 // Trigger a hook
 
-func (hs *Hooks) Trigger(name string, ID string, extra ...Data) {
 
+func (hs *Hooks) Trigger(name string, op Operation, extra ...Data) {
 
 	hs.Lock()
 
 	hs.Lock()
 
 	defer hs.Unlock()
 
 	defer hs.Unlock()
 
 
 
 	for _, h := range hs.hooks {
 
 	for _, h := range hs.hooks {
 
 		if h.Any != nil {
 
 		if h.Any != nil {
 
-			h.Any(name, ID, time.Now(), extra...)
 
+			h.Any(name, op, time.Now(), extra...)
 
 		}
 
 		}
 
 		switch name {
 
 		switch name {
 
 		case "Wait":
 
 		case "Wait":
 
 			if h.Wait != nil {
 
 			if h.Wait != nil {
 
-				h.Wait(ID, time.Now())
 
+				h.Wait(op, time.Now())
 
 			}
 
 			}
 
 		case "Start":
 
 		case "Start":
 
 			if h.Start != nil {
 
 			if h.Start != nil {
 
-				h.Start(ID, time.Now())
 
+				h.Start(op, time.Now())
 
 			}
 
 			}
 
 		case "Finish":
 
 		case "Finish":
 
 			if h.Finish != nil {
 
 			if h.Finish != nil {
 
-				h.Finish(ID, time.Now(), extra[0])
 
+				h.Finish(op, time.Now(), extra[0])
 
 			}
 
 			}
 
 		case "Error":
 
 		case "Error":
 
 			if h.Error != nil {
 
 			if h.Error != nil {
 
-				h.Error(ID, time.Now(), extra[0].(error))
 
+				h.Error(op, time.Now(), extra[0].(error))
 
 			}
 
 			}
 
 		}
 
 		}
 
 	}
 
 	}
 
 }
 
 }
 
 
 
-func (hs *Hooks) wait(ID string)             { hs.Trigger("Wait", ID) }
 
-func (hs *Hooks) start(ID string)            { hs.Trigger("Start", ID) }
 
-func (hs *Hooks) finish(ID string, res Data) { hs.Trigger("Finish", ID, res) }
 
-func (hs *Hooks) error(ID string, err error) { hs.Trigger("Error", ID, err) }
 
+func (hs *Hooks) wait(op Operation)             { hs.Trigger("Wait", op) }
 
+func (hs *Hooks) start(op Operation)            { hs.Trigger("Start", op) }
 
+func (hs *Hooks) finish(op Operation, res Data) { hs.Trigger("Finish", op, res) }
 
+func (hs *Hooks) error(op Operation, err error) { hs.Trigger("Error", op, err) }
 
 
 
 // Attach attach a hook
 
 // Attach attach a hook
 
 func (hs *Hooks) Attach(h Hook) {
 
 func (hs *Hooks) Attach(h Hook) {

+ 125 - 94
go/src/flow/operation.go
View File 

@@ -18,20 +18,21 @@ type executorFunc func(OpCtx, ...Data) (Data, error)
 
 
 
 // Operation interface
 
 // Operation interface
 
 type Operation interface { // Id perhaps?
 
 type Operation interface { // Id perhaps?
 
-	ID() string
 
-	Set(input Data) // Special var method
 
+	//ID() string
 
+	//Set(input Data) // Special var method
 
 	Process(params ...Data) (Data, error)
 
 	Process(params ...Data) (Data, error)
 
 }
 
 }
 
 
 
-// Will be named operation
 
 type operation struct {
 
 type operation struct {
 
 	sync.Mutex
 
 	sync.Mutex
 
-	flow     *Flow
 
-	id       string
 
-	name     string
 
-	kind     string
 
-	inputs   []*operation // still figuring, might be Operation
 
-	setter   func(Data)
 
+	flow *Flow
 
+	name string
 
+	kind string
 
+
 
+	// Could be a simple ID for operation, but it will depend on flow
 
+	inputs []*operation // still figuring, might be Operation
 
+
 
+	fn       interface{} // Any func
 
 	executor executorFunc
 
 	executor executorFunc
 
 }
 
 }
 
 
 
@@ -43,25 +44,19 @@ func newOpCtx() OpCtx {
 
 	return &sync.Map{}
 
 	return &sync.Map{}
 
 }
 
 }
 
 
 
-func (o *operation) ID() string { return o.id }
 
+//func (o *operation) ID() string { return o.id }
 
 
 
 func (o *operation) Process(params ...Data) (Data, error) {
 
 func (o *operation) Process(params ...Data) (Data, error) {
 
 	// Create CTX
 
 	// Create CTX
 
 	ctx := newOpCtx()
 
 	ctx := newOpCtx()
 
 	return o.executor(ctx, params...)
 
 	return o.executor(ctx, params...)
 
 
 
-	//return executeOP(o.flow, o.id, ctx, params...)
 
-}
 
-func (o *operation) Set(data Data) {
 
-	if o.setter != nil {
 
-		o.setter(data)
 
-	}
 
 }
 
 }
 
 
 
 // make Executor for func
 
 // make Executor for func
 
-func (f *Flow) asTrigger(id string, fn executorFunc) executorFunc {
 
+func (f *Flow) asTrigger(op *operation, fn executorFunc) executorFunc {
 
 	return func(ctx OpCtx, params ...Data) (Data, error) {
 
 	return func(ctx OpCtx, params ...Data) (Data, error) {
 
-		f.hooks.start(id)
 
+		f.hooks.start(op)
 
 
 
 		//panic recoverer, since nodes are not our functions
 
 		//panic recoverer, since nodes are not our functions
 
 		var err error
 
 		var err error
 
@@ -78,45 +73,45 @@ func (f *Flow) asTrigger(id string, fn executorFunc) executorFunc {
 
 		}()
 
 		}()
 
 
 
 		if err != nil {
 
 		if err != nil {
 
-			f.hooks.error(id, err)
 
+			f.hooks.error(op, err)
 
 		} else {
 
 		} else {
 
-			f.hooks.finish(id, res)
 
+			f.hooks.finish(op, res)
 
 		}
 
 		}
 
 		return res, err
 
 		return res, err
 
 
 
 	}
 
 	}
 
 }
 
 }
 
 
 
-// save makeExecutor with a bunch of type checks
 
+// safe makeExecutor with a bunch of type checks
 
 // we will create a less safer functions to get performance
 
 // we will create a less safer functions to get performance
 
-func (f *Flow) makeExecutor(id string, fn interface{}) executorFunc {
 
+func (f *Flow) makeExecutor(op *operation, fn interface{}) executorFunc {
 
+	// ExecutorFunc
 
 	return func(ctx OpCtx, params ...Data) (Data, error) {
 
 	return func(ctx OpCtx, params ...Data) (Data, error) {
 
 
 
-		op := f.GetOp(id)
 
+		/*op := f.GetOp(id)
 
 		if op == nil {
 
 		if op == nil {
 
 			return nil, fmt.Errorf("invalid operation '%s'", id)
 
 			return nil, fmt.Errorf("invalid operation '%s'", id)
 
-		}
 
+		}*/
 
 		op.Lock()
 
 		op.Lock()
 
 		defer op.Unlock()
 
 		defer op.Unlock()
 
 
 
 		// Load from cache if any
 
 		// Load from cache if any
 
 		if ctx != nil {
 
 		if ctx != nil {
 
-			if v, ok := ctx.Load(id); ok {
 
+			if v, ok := ctx.Load(op); ok {
 
 				return v, nil
 
 				return v, nil
 
 			}
 
 			}
 
 		}
 
 		}
 
-		// Change to wait
 
-		f.hooks.wait(id)
 
+		// Change to wait to wait for the inputs
 
+		f.hooks.wait(op)
 
 
 
 		fnval := reflect.ValueOf(fn)
 
 		fnval := reflect.ValueOf(fn)
 
 		callParam, err := f.processInputs(ctx, op, fnval, params...)
 
 		callParam, err := f.processInputs(ctx, op, fnval, params...)
 
 		if err != nil {
 
 		if err != nil {
 
-			//log.Println("ERR", err)
 
 			return nil, err
 
 			return nil, err
 
 		}
 
 		}
 
 
 
-		// Start again
 
-		f.hooks.start(id)
 
+		// Start again and execute function
 
+		f.hooks.start(op)
 
 		fnret := fnval.Call(callParam)
 
 		fnret := fnval.Call(callParam)
 
 		if len(fnret) == 0 {
 
 		if len(fnret) == 0 {
 
 			return nil, nil
 
 			return nil, nil
 
@@ -134,15 +129,15 @@ func (f *Flow) makeExecutor(id string, fn interface{}) executorFunc {
 
 		ret := fnret[0].Interface()
 
 		ret := fnret[0].Interface()
 
 		// Store in the cache
 
 		// Store in the cache
 
 		if ctx != nil {
 
 		if ctx != nil {
 
-			ctx.Store(id, ret)
 
+			ctx.Store(op, ret)
 
 		}
 
 		}
 
 		return ret, nil
 
 		return ret, nil
 
 	}
 
 	}
 
 }
 
 }
 
 
 
-/////////////////////////////
 
+// processInputs will run a list of operations and return reflect values
 
+// to be processed next
 
 // NEW PARALLEL PROCESSING
 
 // NEW PARALLEL PROCESSING
 
-///////////
 
 func (f *Flow) processInputs(ctx OpCtx, op *operation, fnval reflect.Value, params ...Data) ([]reflect.Value, error) {
 
 func (f *Flow) processInputs(ctx OpCtx, op *operation, fnval reflect.Value, params ...Data) ([]reflect.Value, error) {
 
 	nInputs := fnval.Type().NumIn()
 
 	nInputs := fnval.Type().NumIn()
 
 
 
@@ -200,6 +195,8 @@ func (f *Flow) processInputs(ctx OpCtx, op *operation, fnval reflect.Value, para
 
 		}(i, in)
 
 		}(i, in)
 
 	}
 
 	}
 
 	wg.Wait()
 
 	wg.Wait()
 
+
 
+	// Check for any error
 
 	if callErrors != "" {
 
 	if callErrors != "" {
 
 		log.Println("Call errors:", callErrors)
 
 		log.Println("Call errors:", callErrors)
 
 		return nil, errors.New(callErrors)
 
 		return nil, errors.New(callErrors)
 
@@ -208,8 +205,40 @@ func (f *Flow) processInputs(ctx OpCtx, op *operation, fnval reflect.Value, para
 
 	return callParam, nil
 
 	return callParam, nil
 
 }
 
 }
 
 
 
-// DefVar define var operation with optional initial
 
-func (f *Flow) DefVar(id string, name string, initial ...Data) Operation {
 
+// Var create a operation
 
+func (f *Flow) Var(name string, initial Data) Operation {
 
+	// Input from params
 
+	var input *operation
 
+	switch v := initial.(type) {
 
+	case *operation:
 
+		input = v
 
+	default:
 
+		c := f.Const(v)
 
+		input = c.(*operation)
 
+	}
 
+	op := &operation{
 
+		Mutex:  sync.Mutex{},
 
+		flow:   f,
 
+		name:   fmt.Sprintf("(var)<%s>", name),
 
+		kind:   "var",
 
+		inputs: []*operation{input},
 
+	}
 
+
 
+	op.executor = f.makeExecutor(op, func(initial Data) Data {
 
+		val, ok := f.Data[name]
 
+		if !ok {
 
+			val = initial
 
+			f.Data[name] = val
 
+		}
 
+		return val
 
+	})
 
+
 
+	return op
 
+
 
+}
 
+
 
+// Var define var operation with optional initial
 
+/*func (f *Flow) Var(name string, initial ...Data) Operation {
 
 	// Unique
 
 	// Unique
 
 	if _, ok := f.Data[name]; !ok {
 
 	if _, ok := f.Data[name]; !ok {
 
 		var v interface{}
 
 		var v interface{}
 
@@ -218,27 +247,26 @@ func (f *Flow) DefVar(id string, name string, initial ...Data) Operation {
 
 		}
 
 		}
 
 		f.Data[name] = v
 
 		f.Data[name] = v
 
 	}
 
 	}
 
-	setter := func(v Data) { f.Data[name] = v }
 
 
 
-	opEntry := &operation{
 
+	op := &operation{
 
 		Mutex:  sync.Mutex{},
 
 		Mutex:  sync.Mutex{},
 
-		id:     id,
 
 		flow:   f,
 
 		flow:   f,
 
 		name:   fmt.Sprintf("(var)<%s>", name),
 
 		name:   fmt.Sprintf("(var)<%s>", name),
 
 		kind:   "var",
 
 		kind:   "var",
 
 		inputs: nil,
 
 		inputs: nil,
 
-		setter: setter,
 
-		executor: f.asTrigger(id, func(OpCtx, ...Data) (Data, error) {
 
-			// if f.data == nil we set from the init operation
 
-			return f.Data[name], nil
 
-		}),
 
 	}
 
 	}
 
-	f.operations.Store(id, opEntry)
 
-	return opEntry
 
-}
 
+	op.executor = f.asTrigger(op, func(OpCtx, ...Data) (Data, error) {
 
+		// if f.data == nil we set from the init operation
 
+		return f.Data[name], nil
 
+	})
 
+	f.operations = append(f.operations, op)
 
+	//f.operations.Store(id, op)
 
+	return op
 
+}*/
 
 
 
-// DefOp Manual tag an Operation
 
-func (f *Flow) DefOp(id string, name string, params ...interface{}) (Operation, error) {
 
+// Op Manual tag an Operation
 
+// Define operation for ID
 
+func (f *Flow) Op(name string, params ...interface{}) Operation {
 
 	inputs := make([]*operation, len(params))
 
 	inputs := make([]*operation, len(params))
 
 	for i, p := range params {
 
 	for i, p := range params {
 
 		switch v := p.(type) {
 
 		switch v := p.(type) {
 
@@ -254,78 +282,81 @@ func (f *Flow) DefOp(id string, name string, params ...interface{}) (Operation,
 
 	// Grab executor here
 
 	// Grab executor here
 
 	registryFn, err := f.registry.Get(name)
 
 	registryFn, err := f.registry.Get(name)
 
 	if err != nil {
 
 	if err != nil {
 
-		return nil, err
 
+		return f.ErrOp(err)
 
 	}
 
 	}
 
-	executor := f.makeExecutor(id, registryFn)
 
 	op := &operation{
 
 	op := &operation{
 
-		Mutex:    sync.Mutex{},
 
-		id:       id,
 
-		flow:     f,
 
-		name:     name,
 
-		kind:     "func",
 
-		inputs:   inputs,
 
-		setter:   nil, // No set
 
-		executor: f.asTrigger(id, executor),
 
+		Mutex:  sync.Mutex{},
 
+		flow:   f,
 
+		name:   name,
 
+		kind:   "func",
 
+		inputs: inputs,
 
 	}
 
 	}
 
-	f.operations.Store(id, op)
 
+	executor := f.makeExecutor(op, registryFn)
 
+	op.executor = f.asTrigger(op, executor)
 
+	f.operations = append(f.operations, op)
 
 
 
-	return op, nil
 
+	return op
 
 }
 
 }
 
 
 
-// DefErrOp define a nil operation that will return error
 
+// ErrOp define a nil operation that will return error
 
 // Usefull for builders
 
 // Usefull for builders
 
-func (f *Flow) DefErrOp(id string, err error) Operation {
 
+func (f *Flow) ErrOp(err error) Operation {
 
 	op := &operation{
 
 	op := &operation{
 
-		Mutex:    sync.Mutex{},
 
-		id:       id,
 
-		flow:     f,
 
-		name:     fmt.Sprintf("(error)<%v>", err),
 
-		kind:     "error",
 
-		inputs:   nil,
 
-		setter:   nil,
 
-		executor: f.asTrigger(id, func(OpCtx, ...Data) (Data, error) { return nil, err }),
 
+		Mutex: sync.Mutex{},
 
+		//id:     id,
 
+		flow:   f,
 
+		name:   fmt.Sprintf("(error)<%v>", err),
 
+		kind:   "error",
 
+		inputs: nil,
 
 	}
 
 	}
 
-	f.operations.Store(id, op)
 
+	op.executor = f.asTrigger(op, func(OpCtx, ...Data) (Data, error) { return nil, err })
 
+	//f.operations = append(f.operations, op)
 
 	return op
 
 	return op
 
 }
 
 }
 
 
 
-// DefConst define a const by defined ID
 
-func (f *Flow) DefConst(id string, value Data) Operation {
 
+// Const define a const by defined ID
 
+func (f *Flow) Const(value Data) Operation {
 
 	// Optimize this definition
 
 	// Optimize this definition
 
-	f.consts[id] = value
 
+	constID := -1
 
+	for k, v := range f.consts {
 
+		if v == value {
 
+			constID = k
 
+			break
 
+		}
 
+	}
 
+	if constID == -1 {
 
+		constID = len(f.consts)
 
+		f.consts = append(f.consts, value)
 
+	}
 
 
 
 	op := &operation{
 
 	op := &operation{
 
-		id:       id,
 
-		Mutex:    sync.Mutex{},
 
-		flow:     f,
 
-		name:     fmt.Sprintf("(const)<%s>", id),
 
-		kind:     "const",
 
-		inputs:   nil,
 
-		setter:   nil,
 
-		executor: f.asTrigger(id, func(OpCtx, ...Data) (Data, error) { return f.consts[id], nil }),
 
+		Mutex:  sync.Mutex{},
 
+		flow:   f,
 
+		name:   fmt.Sprintf("(const)<%v:%v>", constID, value),
 
+		kind:   "const",
 
+		inputs: nil,
 
 	}
 
 	}
 
-	f.operations.Store(id, op)
 
+	op.executor = f.asTrigger(op, func(OpCtx, ...Data) (Data, error) { return f.consts[constID], nil })
 
+	//f.operations = append(f.operations, op)
 
 
 
 	return op
 
 	return op
 
 }
 
 }
 
 
 
-// DefIn define input operation
 
-func (f *Flow) DefIn(id string, paramID int) Operation {
 
+// In define input operation
 
+func (f *Flow) In(paramID int) Operation {
 
 	op := &operation{
 
 	op := &operation{
 
-		id:     id,
 
 		Mutex:  sync.Mutex{},
 
 		Mutex:  sync.Mutex{},
 
 		flow:   f,
 
 		flow:   f,
 
 		name:   fmt.Sprintf("(in)<%d>", paramID),
 
 		name:   fmt.Sprintf("(in)<%d>", paramID),
 
 		kind:   "in",
 
 		kind:   "in",
 
 		inputs: nil,
 
 		inputs: nil,
 
-		setter: nil,
 
-		executor: f.asTrigger(id, func(ctx OpCtx, params ...Data) (Data, error) {
 
-			if paramID < 0 || paramID >= len(params) {
 
-				return nil, ErrInput
 
-			}
 
-			return params[paramID], nil
 
-		}),
 
 	}
 
 	}
 
-	f.operations.Store(id, op)
 
+	op.executor = f.asTrigger(op, func(ctx OpCtx, params ...Data) (Data, error) {
 
+		if paramID < 0 || paramID >= len(params) {
 
+			return nil, ErrInput
 
+		}
 
+		return params[paramID], nil
 
+	})
 
+	f.operations = append(f.operations, op)
 
 	return op
 
 	return op
 
 }
 
 }

+ 14 - 7
go/src/flow/registry/registry.go
View File 

@@ -19,7 +19,7 @@ type M = map[string]interface{}
 
 
 
 // R the function registry
 
 // R the function registry
 
 type R struct {
 
 type R struct {
 
-	data map[string]*Entry
 
+	entries map[string]*Entry
 
 }
 
 }
 
 
 
 // New creates a new registry
 
 // New creates a new registry
 
@@ -32,12 +32,19 @@ func New() *R {
 
 // Clone an existing registry
 
 // Clone an existing registry
 
 func (r *R) Clone() *R {
 
 func (r *R) Clone() *R {
 
 	newR := &R{map[string]*Entry{}}
 
 	newR := &R{map[string]*Entry{}}
 
-	for k, v := range r.data {
 
-		newR.data[k] = v
 
+	for k, v := range r.entries {
 
+		newR.entries[k] = v
 
 	}
 
 	}
 
 	return newR
 
 	return newR
 
 }
 
 }
 
 
 
+// Merge other registry
 
+func (r *R) Merge(or *R) {
 
+	for k, v := range or.entries {
 
+		r.entries[k] = v
 
+	}
 
+}
 
+
 
 // Add function to registry
 
 // Add function to registry
 
 func (r *R) Add(args ...interface{}) *EDescriber {
 
 func (r *R) Add(args ...interface{}) *EDescriber {
 
 
 
@@ -92,13 +99,13 @@ func (r *R) register(name string, v interface{}) (*Entry, error) {
 
 	if err != nil {
 
 	if err != nil {
 
 		return nil, err
 
 		return nil, err
 
 	}
 
 	}
 
-	r.data[name] = e
 
+	r.entries[name] = e
 
 	return e, nil
 
 	return e, nil
 
 }
 
 }
 
 
 
 // Get an entry
 
 // Get an entry
 
 func (r *R) Get(name string, params ...interface{}) (interface{}, error) {
 
 func (r *R) Get(name string, params ...interface{}) (interface{}, error) {
 
-	e, ok := r.data[name]
 
+	e, ok := r.entries[name]
 
 	if !ok {
 
 	if !ok {
 
 		return nil, ErrNotFound
 
 		return nil, ErrNotFound
 
 	}
 
 	}
 
@@ -123,7 +130,7 @@ func (r *R) Get(name string, params ...interface{}) (interface{}, error) {
 
 
 
 // Entry fetches entries from the register
 
 // Entry fetches entries from the register
 
 func (r *R) Entry(name string) (*Entry, error) {
 
 func (r *R) Entry(name string) (*Entry, error) {
 
-	e, ok := r.data[name]
 
+	e, ok := r.entries[name]
 
 	if !ok {
 
 	if !ok {
 
 		return nil, ErrNotFound
 
 		return nil, ErrNotFound
 
 	}
 
 	}
 
@@ -135,7 +142,7 @@ func (r *R) Entry(name string) (*Entry, error) {
 
 // Descriptions Description list
 
 // Descriptions Description list
 
 func (r *R) Descriptions() (map[string]Description, error) {
 
 func (r *R) Descriptions() (map[string]Description, error) {
 
 	ret := map[string]Description{}
 
 	ret := map[string]Description{}
 
-	for k, e := range r.data {
 
+	for k, e := range r.entries {
 
 		ret[k] = e.Description
 
 		ret[k] = e.Description
 
 	}
 
 	}
 
 	return ret, nil
 
 	return ret, nil

+ 17 - 0
go/src/flow/registry/registry_test.go
View File 

@@ -3,6 +3,7 @@ package registry_test
 
 import (
 
 import (
 
 	"flow/internal/assert"
 
 	"flow/internal/assert"
 
 	"flow/registry"
 
 	"flow/registry"
 
+	"strings"
 
 	"testing"
 
 	"testing"
 
 
 
 	"github.com/gohxs/prettylog"
 
 	"github.com/gohxs/prettylog"
 
@@ -177,6 +178,22 @@ func TestRegisterErr(t *testing.T) {
 
 	assert.Eq(t, d.Err, registry.ErrNotAFunc, "should give a func error")
 
 	assert.Eq(t, d.Err, registry.ErrNotAFunc, "should give a func error")
 
 }
 
 }
 
 
 
+func TestMerge(t *testing.T) {
 
+	a := assert.A(t)
 
+	r1 := registry.New()
 
+	r1.Add(strings.Join)
 
+
 
+	r2 := registry.New()
 
+	r2.Add(strings.Split)
 
+
 
+	r2.Merge(r1)
 
+
 
+	m, err := r2.Descriptions()
 
+	a.Eq(err, nil, "should not error fetching descriptions")
 
+	a.Eq(len(m), 2, "Number of descriptions should be 2")
 
+
 
+}
 
+
 
 /*func TestAddEntry(t *testing.T) {
 
 /*func TestAddEntry(t *testing.T) {
 
 	a := assert.A(t)
 
 	a := assert.A(t)
 
 	r := registry.New()
 
 	r := registry.New()

+ 55 - 0
go/src/flow/session.go
View File 

@@ -0,0 +1,55 @@
 
+// TODO operations shoudl be run in a session to store variables structures etc
 
+
 
+package flow
 
+
 
+import "sync"
 
+
 
+// RunCtx operation Context
 
+type RunCtx = *sync.Map
 
+
 
+// NewOpCtx creates a running context
 
+func newRunCtx() RunCtx {
 
+	return &sync.Map{}
 
+}
 
+
 
+// Session running session
 
+// should hold variables and operation clones?
 
+type Session struct {
 
+	flow *Flow
 
+	data map[string]Data
 
+}
 
+
 
+// Run operation runner
 
+func (s *Session) Run(op *operation) (Data, error) {
 
+	runCtx := newRunCtx()
 
+	return s.run(runCtx, op)
 
+}
 
+
 
+// internal run
 
+func (s *Session) run(runCtx RunCtx, op *operation) (Data, error) {
 
+	// If cache return cache
 
+	// Cache check
 
+	if val, ok := runCtx.Load(op); ok {
 
+		return val, nil
 
+	}
 
+
 
+	// Process inputs?
 
+	//
 
+
 
+	return nil, nil
 
+}
 
+
 
+// Process inputs for operation
 
+func (s *Session) processInputs(RunCtx, op *operation, params ...Data) ([]Data, error) {
 
+	ret := make([]Data, len(op.inputs))
 
+
 
+	wg := sync.WaitGroup{}
 
+	wg.Add(len(op.inputs))
 
+	for i, in := range op.inputs {
 
+		go func(i, in *operation) {
 
+			r, err := in.executor(params...)
 
+
 
+		}(i, in)
 
+
 
+	}
 
+}