final commit for persistent flu

soniabhishek · Oct 6, 2016 · 1d7d987 · 1d7d987
1 parent a220db5
commit 1d7d987
Show file tree

Hide file tree

Showing 6 changed files with 18 additions and 12 deletions.
diff --git a/app/models/uuid/uuid.go b/app/models/uuid/uuid.go
@@ -288,7 +288,7 @@ func (u *UUID) UnmarshalBinary(data []byte) (err error) {
 func (u UUID) Value() (driver.Value, error) {
 	if u == Nil {
 		defer func() {
-			plog.Info("Nil uuid Value() called")
+			//plog.Info("Nil uuid Value() called")
 		}()
 		return nil, nil
 	}

diff --git a/app/services/flu_svc/export.go b/app/services/flu_svc/export.go
@@ -7,7 +7,7 @@ import (
 	"github.com/crowdflux/angel/app/DAL/repositories/projects_repo"
 	"github.com/crowdflux/angel/app/services/flu_svc/flu_validator"
 	"github.com/crowdflux/angel/app/services/work_flow_svc"
-	"github.com/robfig/cron"
+	"time"
 )
 
 func New() IFluService {
@@ -37,16 +37,18 @@ func NewWithExposedValidators() IFluServiceExtended {
 
 func StartFeedLineSync() {
 
-	fSvc := New()
-	c := cron.New()
+	go func() {
 
-	syncFeedLine := func() {
-		err := fSvc.SyncInputFeedLine()
-		if err != nil {
-			fmt.Println(err)
+		fSvc := New()
+
+		ticker := time.Tick(time.Duration(2) * time.Minute)
+
+		for _ = range ticker {
+			err := fSvc.SyncInputFeedLine()
+			if err != nil {
+				fmt.Println(err)
+			}
 		}
-	}
+	}()
 
-	c.AddFunc("0 */2 * * * *", syncFeedLine)
-	c.Start()
 }
diff --git a/app/services/flu_svc/flu_output/flu_monitor.go b/app/services/flu_svc/flu_output/flu_monitor.go
@@ -60,7 +60,7 @@ func (fm *FluMonitor) AddToOutputQueue(flu models.FeedLineUnit) error {
 
 func (fm *FluMonitor) AddManyToOutputQueue(fluBundle []models.FeedLineUnit) error {
 
-	plog.Info("FLu Monitor", fluBundle)
+	plog.Info("FLu Monitor, flubundle count:", len(fluBundle))
 
 	mutex.Lock()
 	for _, flu := range fluBundle {

diff --git a/app/services/work_flow_svc/step/manual_step_svc/manual_step.go b/app/services/work_flow_svc/step/manual_step_svc/manual_step.go
@@ -14,6 +14,7 @@ type manualStep struct {
 
 func (m *manualStep) processFlu(flu feed_line.FLU) {
 	m.AddToBuffer(flu)
+	flu.ConfirmReceive()
 	plog.Info("Manual Step flu reached", flu.ID)
 }
 

diff --git a/app/services/work_flow_svc/work_flow/workflow.go b/app/services/work_flow_svc/work_flow/workflow.go
@@ -60,6 +60,7 @@ func NewShortCircuit() WorkFlow {
 	// and send back the input as output (short circuit)
 	go func() {
 		for flu := range w.InQ.Receiver() {
+			counter.Print(flu, "shortcircuit workflow out")
 			w.OutQ.Push(flu)
 		}
 	}()

diff --git a/app/services/work_flow_svc/work_flow_svc.go b/app/services/work_flow_svc/work_flow_svc.go
@@ -46,6 +46,7 @@ func startWorkflowSvc(w *workFlowSvc) {
 	go func() {
 		for flu := range w.OutQ.Receiver() {
 			w.complete(flu.FeedLineUnit)
+			flu.ConfirmReceive()
 		}
 	}()
 }
@@ -54,6 +55,7 @@ func startWorkflowSvcNLog(w *workFlowSvc) {
 	go func() {
 		for flu := range w.OutQ.Receiver() {
 			fmt.Println(flu.ID)
+			flu.ConfirmReceive()
 		}
 	}()
 }