possible fix for caching

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

@@ -66,21 +66,6 @@ export default {
       return utils.padStart(min.toFixed(0), 2, '0') + ':' + utils.padStart(sec.toFixed(0), 2, '0')
     }
   },
-  watch: {
-    activity: {
-      handler (val) {
-        console.log('Activity changed', val)
-        const e = new Date(Date.parse(val.endTime))
-        if (dateIsValid(e)) {
-          console.log('End time valid', e.getTime())
-        } else {
-          console.log('End time invalid')
-        }
-      },
-      deep: true
-    }
-  },
-
   mounted () {
     console.log('Activity', this.activity)
     this._timeOut = setTimeout(this.updateTime, 999)
@@ -92,7 +77,6 @@ export default {
     updateTime () {
       const finish = new Date(Date.parse(this.activity.endTime))
       if (dateIsValid(finish)) {
-        console.log('Date is valid')
         this.finishTime = finish
         return
       }

go/src/flow/flow.go

@@ -22,6 +22,7 @@ var (
 type Data = interface{}
 
 type opEntry struct {
+	sync.Mutex
 	name     string
 	inputs   []*operation // still figuring, might be operation
 	executor interface{}
@@ -86,7 +87,7 @@ func (f *Flow) DefOp(id string, name string, params ...interface{}) Operation {
 		f.err = err
 		return opNil(f)
 	}
-	f.operations.Store(id, &opEntry{name, inputs, executor})
+	f.operations.Store(id, &opEntry{sync.Mutex{}, name, inputs, executor})
 	return opFunc(f, id)
 }
 
@@ -126,7 +127,7 @@ func (f *Flow) Op(name string, params ...interface{}) Operation {
 		if _, ok := f.operations.Load(id); ok {
 			continue
 		}
-		f.operations.Store(id, &opEntry{name, inputs, executor})
+		f.operations.Store(id, &opEntry{sync.Mutex{}, name, inputs, executor})
 		return opFunc(f, id)
 	}
 	// Initialize opfunc maybe

go/src/flow/flow_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"testing"
+	"time"
 
 	"flow"
 
@@ -133,10 +134,10 @@ func TestDefOp(t *testing.T) {
 func TestHandler(t *testing.T) {
 	f, op := prepareComplex()
 	f.Hook(flow.Hook{
-		Wait:   func(ID string) { t.Logf("[%s] Wait", ID) },
-		Start:  func(ID string) { t.Logf("[%s]Start", ID) },
-		Finish: func(ID string, res flow.Data) { t.Logf("[%s] Finish %v", ID, res) },
-		Error:  func(ID string, err error) { t.Logf("[%s] Error %v", ID, err) },
+		Wait:   func(ID string, triggerTime time.Time) { t.Logf("[%s] Wait", ID) },
+		Start:  func(ID string, triggerTime time.Time) { t.Logf("[%s]Start", ID) },
+		Finish: func(ID string, triggerTime time.Time, res flow.Data) { t.Logf("[%s] Finish %v", ID, res) },
+		Error:  func(ID string, triggerTime time.Time, err error) { t.Logf("[%s] Error %v", ID, err) },
 	})
 	op.Process()
 }

go/src/flow/operation.go

@@ -11,8 +11,6 @@ import (
 	"log"
 	"reflect"
 	"sync"
-
-	"github.com/gohxs/prettylog"
 )
 
 // OpCtx operation Context
@@ -37,11 +35,9 @@ type Operation interface { // Id perhaps?
 
 //local operation information
 type operation struct {
-	sync.Mutex
 	flow    *Flow
 	id      interface{} // Interface key
 	kind    string
-	src     string
 	set     func(params ...Data)
 	process func(ctx OpCtx, params ...Data) Data
 }
@@ -59,32 +55,20 @@ 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))
-	log := prettylog.New(o.ID() + ":" + entry.name)
-
-	log.Printf("Operation waiting")
-	o.Lock()
-	defer o.Unlock()
-	log.Printf("Executing")
-
-	log.Println("Context", ctx)
+	entry.Lock()
+	defer entry.Unlock()
 
 	if o.flow.err != nil {
 		return nil
 	}
 	if ctx == nil { // No cache/Context
-		log.Printf("Processing")
 		return o.process(ctx, params...)
 	}
-	if v, ok := ctx.Load(o); ok {
-		log.Printf("Cached")
+	if v, ok := ctx.Load(o.id); ok {
 		return v
 	}
-
-	log.Printf("Processing")
 	res := o.process(ctx, params...)
-	log.Println("Storing", res)
-	ctx.Store(o, res)
-
+	ctx.Store(o.id, res)
 	return res
 }