[core] fix subscribe to workflow state change

Author: miltalex

core/environment/manager.go

@@ -115,41 +115,6 @@ func (envs *Manager) CreateEnvironment(workflowPath string, userVars map[string]
 		return env.id, err
 	}
 
-	// This will subscribe to workflow state change. In case of workflow state
-	// ERROR will transition environment to ERROR state. The goroutine starts 
-	// after a successful transition to CONFIGURE in order to handle only ERROR
-	// states triggered by mesos.(TASK_LOST,TASK_KILLED,TASK_FAILED,TASK_ERROR)
-	go func() {
-		wf := env.Workflow()
-		notify := make(chan task.State)
-		subscriptionId := uuid.NewUUID().String()
-		env.wfAdapter.SubscribeToStateChange(subscriptionId, notify)
-		defer env.wfAdapter.UnsubscribeFromStateChange(subscriptionId)
-
-		wfState := wf.GetState()
-		if wfState != task.ERROR {
-			WORKFLOW_STATE_LOOP:
-			for {
-				select {
-				case wfState = <-notify:
-					if wfState == task.DONE {
-						break WORKFLOW_STATE_LOOP
-					}
-					if wfState == task.ERROR {
-						env.setState(wfState.String())
-						break WORKFLOW_STATE_LOOP
-					}
-					continue
-				default:
-					if _, ok := envs.m[env.id.Array()]; !ok {
-						break WORKFLOW_STATE_LOOP
-					}
-					continue
-				}
-			}
-		}
-	}()
-
 	return env.id, err
 }
 

core/environment/transition_configure.go

@@ -178,7 +178,44 @@ func (t ConfigureTransition) do(env *Environment) (err error) {
 
 	if len(tasks) != 0 {
 		err = t.taskman.ConfigureTasks(env.Id().Array(), tasks)
+		if err != nil {
+			return
+		}
 	}
 
+	// This will subscribe to workflow state change. In case of workflow state
+	// ERROR will transition environment to ERROR state. The goroutine starts 
+	// after a successful transition to CONFIGURE in order to handle only ERROR
+	// states triggered by mesos.(TASK_LOST,TASK_KILLED,TASK_FAILED,TASK_ERROR)
+	go func() {
+		wf := env.Workflow()
+		notify := make(chan task.State)
+		subscriptionId := uuid.NewUUID().String()
+		env.wfAdapter.SubscribeToStateChange(subscriptionId, notify)
+		defer env.wfAdapter.UnsubscribeFromStateChange(subscriptionId)
+
+		wfState := wf.GetState()
+		if wfState != task.ERROR {
+			WORKFLOW_STATE_LOOP:
+			for {
+				select {
+				case wfState = <-notify:
+					if wfState == task.DONE {
+						break WORKFLOW_STATE_LOOP
+					}
+					// We kill the goroutine on a reset or a teardown
+					// of the environment
+					if wfState == task.STANDBY {
+						break WORKFLOW_STATE_LOOP
+					}
+					if wfState == task.ERROR {
+						env.setState(wfState.String())
+						break WORKFLOW_STATE_LOOP
+					}
+				}
+			}
+		}
+	}()
+
 	return
 }