func (e *Engine) waitCloseOutputCh(stage stages.Stage) {
	for {
		receive, ok := <-*stage.GetOutputCh()
		if !ok {
			log.Debugf("outputCh closed")
			break
		}
		log.Debugf("outputCh received  %+v\n", receive)
	}
}

func (e *Engine) waitMonitorChFinalized(name string) stages.Mediator {
	var receives = make([]stages.Mediator, 0)
	for {
		receive := <-*e.MonitorCh
		receives = append(receives, receive)
		if receive.Type == "end" {
			log.Debugf("monitorCh closed")
			log.Debugf("monitorCh last received:  %+v\n", receive)
			log.Debugf("----- Execute %v done ------\n\n", name)
			return e.bindReceives(&receives)
		}
		log.Debugf("monitorCh received %+v\n", receive)
	}
}

func (e *Engine) executeAllChildStages(childStages *list.List, mediator stages.Mediator) {
	for childStage := childStages.Front(); childStage != nil; childStage = childStage.Next() {
		log.Debugf("Child name %+v\n", childStage.Value.(stages.Stage).GetStageName())
		childInputCh := *childStage.Value.(stages.Stage).GetInputCh()

		go func() {
			e.ExecuteStage(childStage.Value.(stages.Stage))
		}()

		log.Debugf("Input child: %+v", mediator)
		childInputCh <- mediator
		log.Debugf("Closing input: %+v", childStage.Value.(stages.Stage).GetStageName())
		close(childInputCh)
	}
}

func (e *Engine) executeChildStages(stage *stages.Stage, mediator *stages.Mediator) {
	if childStages := (*stage).GetChildStages(); childStages.Len() > 0 {
		log.Debugf("Execute childstage: %v", childStages)
		e.executeAllChildStages(&childStages, *mediator)
		e.waitAllChildStages(&childStages, stage)
	}
}

func (e *Engine) waitAllChildStages(childStages *list.List, stage *stages.Stage) {
	for childStage := childStages.Front(); childStage != nil; childStage = childStage.Next() {
		s := childStage.Value.(stages.Stage)
		for {
			log.Debugf("Receiving child: %v", s.GetStageName())
			childReceived, ok := <-*s.GetOutputCh()
			if !ok {
				log.Debug("Closing child output")
				break
			}
			log.Debugf("Sending child: %v", childReceived)
			*(*stage).GetOutputCh() <- childReceived
			log.Debugf("Send child: %v", childReceived)
		}
		log.Debugf("Finished executing child: %v", s.GetStageName())
	}
}

// Run exectues specified shell script.
func (shellScriptStage *ShellScriptStage) Run() bool {
	log.Infof("[shell] exec: %s", shellScriptStage.BaseStage.StageName)
	log.Debugf("[shell] specified file: %s\n", shellScriptStage.File)
	if shellScriptStage.preCheck() == false {
		log.Infof("failed preCheck before running script...")
		return false
	}
	shellScriptStage.AddCommand("sh " + shellScriptStage.File)
	return shellScriptStage.CommandStage.Run()
}

func (e *Engine) executePipeline(pipeline *pipelines.Pipeline, name string) stages.Mediator {
	log.Infof("Preparing to run %s process...", name)
	var mediator stages.Mediator
	for stageItem := pipeline.Stages.Front(); stageItem != nil; stageItem = stageItem.Next() {
		log.Debugf("Executing planned stage: %s\n", stageItem.Value)
		mediator = e.Execute(stageItem.Value.(stages.Stage), mediator)
	}
	log.Infof("Finished running %s process...", name)
	return mediator
}

func (commandStage *CommandStage) runCommand() bool {
	cmd := exec.Command("sh", "-c", commandStage.Command)
	log.Infof("[command] exec: %s", commandStage.BaseStage.StageName)
	log.Debugf("[command] exec command literal: %s", commandStage.Command)
	cmd.Dir = commandStage.Directory
	result, outResult, errResult := execCommand(cmd, "exec", commandStage.BaseStage.StageName)
	commandStage.SetOutResult(*outResult)
	commandStage.SetErrResult(*errResult)
	return result
}

func (commandStage *CommandStage) runOnlyIf() bool {
	if commandStage.OnlyIf == "" {
		return true
	}
	log.Infof("[command] only_if: found \"only_if\" attribute in stage \"%s\"", commandStage.BaseStage.StageName)
	cmd := exec.Command("sh", "-c", commandStage.OnlyIf)
	log.Infof("[command] only_if: %s", commandStage.BaseStage.StageName)
	log.Debugf("[command] only_if literal: %s", commandStage.OnlyIf)
	cmd.Dir = commandStage.Directory
	result, _, _ := execCommand(cmd, "only_if", commandStage.BaseStage.StageName)
	return result
}

func (e *Engine) receiveInputs(inputCh *chan stages.Mediator) []stages.Mediator {
	var mediatorsReceived = make([]stages.Mediator, 0)
	for {
		m, ok := <-*inputCh
		if !ok {
			break
		}
		log.Debugf("received input: %+v", m)
		mediatorsReceived = append(mediatorsReceived, m)
	}
	return mediatorsReceived
}

func (commandStage *CommandStage) runCommand() bool {
	cmd := exec.Command("sh", "-c", commandStage.Command)
	log.Infof("[command] exec: %s", commandStage.BaseStage.StageName)
	log.Debugf("[command] exec command literal: %s", commandStage.Command)
	cmd.Dir = commandStage.Directory
	commandStage.SetStart(time.Now().Unix())
	result, outResult, errResult, combinedResult := execCommand(cmd, "exec", commandStage.BaseStage.StageName)
	commandStage.SetEnd(time.Now().Unix())
	commandStage.SetOutResult(*outResult)
	commandStage.SetErrResult(*errResult)
	commandStage.SetCombinedResult(*combinedResult)
	commandStage.SetReturnValue(result)
	return result
}

//Execute executes a stage using the supplied mediator
func (e *Engine) Execute(stage stages.Stage, mediator stages.Mediator) stages.Mediator {
	mediator.Type = "start"
	name := stage.GetStageName()
	log.Debugf("----- Execute %v start ------\n", name)

	go func() {
		*stage.GetInputCh() <- mediator
		close(*stage.GetInputCh())
	}()

	go e.ExecuteStage(stage)
	e.waitCloseOutputCh(stage)
	return e.waitMonitorChFinalized(name)
}

func (e *Engine) executeStage(stage stages.Stage, received []stages.Mediator) string {
	var result string
	if !e.isUpstreamAnyFailure(received) || e.Opts.StopOnAnyFailure {
		result = strconv.FormatBool(stage.(stages.Runner).Run())
		e.EnvVariables.ExportSpecialVariable("__OUT[\""+stage.GetStageName()+"\"]", stage.GetOutResult())
		e.EnvVariables.ExportSpecialVariable("__ERR[\""+stage.GetStageName()+"\"]", stage.GetErrResult())
		e.EnvVariables.ExportSpecialVariable("__RESULT[\""+stage.GetStageName()+"\"]", result)
	} else {
		log.Warnf("Execution is skipped: %v", stage.GetStageName())
		result = "skipped"
	}
	e.Resources.ReportStageResult(stage, result)
	log.Debugf("Stage execution results: %+v, %+v", stage.GetStageName(), result)
	return result
}

//Validate validates if the command can be executed
func (resourceValidator *ResourceValidator) Validate() bool {
	// check if files exists
	for file := resourceValidator.files.Front(); file != nil; file = file.Next() {
		filePath := file.Value.(string)
		log.Debugf("checking file: %v", filePath)
		if _, err := os.Stat(filePath); err == nil {
			log.Debugf("file exists")
		} else {
			log.Errorf("file: %v does not exists", filePath)
			return false
		}
	}
	// check if command exists
	if len(resourceValidator.command) == 0 { // return true when no command is registrated
		return true
	}
	cmd := exec.Command("which", resourceValidator.command)
	err := cmd.Run()
	if err != nil {
		log.Errorf("command: %v does not exists", resourceValidator.command)
		return false
	}
	return true
}

func (e *Engine) finalizeMonitorChAfterExecute(mediators []stages.Mediator, mediator stages.Mediator) {
	// Append to monitorCh
	*e.MonitorCh <- mediator
	for _, m := range mediators {
		*e.MonitorCh <- m
	}

	// Set type
	if mediators[0].Type == "start" {
		log.Debug("Finalize monitor channel..")
		mediatorEnd := stages.Mediator{States: make(map[string]string), Type: "end"}
		*e.MonitorCh <- mediatorEnd
	} else {
		log.Debugf("Skipped finalizing")
	}
}

func (e *Engine) executeStage(stage stages.Stage, received []stages.Mediator, mediator stages.Mediator) string {
	var result string
	if !e.isUpstreamAnyFailure(received) || e.Opts.StopOnAnyFailure {
		result = strconv.FormatBool(stage.(stages.Runner).Run())
		e.EnvVariables.ExportSpecialVariable("__OUT[\""+stage.GetStageName()+"\"]", stage.GetOutResult())
		e.EnvVariables.ExportSpecialVariable("__ERR[\""+stage.GetStageName()+"\"]", stage.GetErrResult())
		e.EnvVariables.ExportSpecialVariable("__COMBINED[\""+stage.GetStageName()+"\"]", stage.GetCombinedResult())
		e.EnvVariables.ExportSpecialVariable("__RESULT[\""+stage.GetStageName()+"\"]", result)
		e.executeChildStages(&stage, &mediator)
	} else {
		log.Warnf("Execution is skipped: %v", stage.GetStageName())
		if childStages := stage.GetChildStages(); childStages.Len() > 0 {
			for childStage := childStages.Front(); childStage != nil; childStage = childStage.Next() {
				log.Warnf("Execution of child stage is skipped: %v", childStage.Value.(stages.Stage).GetStageName())
			}
		}
		result = "skipped"
	}
	if !stage.GetSuppressAll() {
		e.Resources.ReportStageResult(stage, result)
	}
	log.Debugf("Stage execution results: %+v, %+v", stage.GetStageName(), result)
	return result
}

// ExecuteStage executes the supplied stage
func (e *Engine) ExecuteStage(stage stages.Stage) {
	log.Debug("Receiving input")
	mediatorsReceived := e.receiveInputs(stage.GetInputCh())

	log.Debugf("Received input size: %v", len(mediatorsReceived))
	log.Debugf("Mediator received: %+v", mediatorsReceived)
	log.Debugf("Execute as parent: %+v", stage)
	log.Debugf("Execute as parent name %+v", stage.GetStageName())

	mediator := stages.Mediator{States: make(map[string]string)}
	mediator.States[stage.GetStageName()] = e.executeStage(stage, mediatorsReceived, mediator)

	log.Debugf("Sending output of stage: %+v %v", stage.GetStageName(), mediator)
	*stage.GetOutputCh() <- mediator
	close(*stage.GetOutputCh())
	log.Debugf("Closed output of stage: %+v", stage.GetStageName())

	e.finalizeMonitorChAfterExecute(mediatorsReceived, mediator)
}

// AddChildStage appends one child stage.
func (b *BaseStage) AddChildStage(stage Stage) {
	log.Debugf("added childstage: %v", stage)
	b.ChildStages.PushBack(stage)
}