// Run the state machine enabled handler. Loads the initial state and passes
// control to the internal stateful handler passing commands from the command
// listener into the handler's commands chan.
Run the state machine enabled handler. Loads the initial state and passes
control to the internal stateful handler passing commands from the command
listener into the handler's commands chan.
// Run the state machine enabled handler . Loads the initial state and passes // control to the internal stateful handler passing commands from the command // listener into the handler ' s commands chan .
func (s *stateMachine) Run() (done bool) {
	s.cmds = make(chan *Message)
	go func() {
		for {
			select {
			case m := <-s.cl.Receive():
				if !m.Valid() {
					metafora.Warnf("Ignoring invalid command: %q", m)
					continue
				}
				select {
				case s.cmds <- m:
				case <-s.stopped:
					return
				}
			case <-s.stopped:
				return
			}
		}
	}()
	defer func() {
		s.cl.Stop()
		s.stop()
	}()
	tid := s.task.ID()
	state, err := s.ss.Load(s.task)
	if err == ReleasableError {
		metafora.Errorf("task=%q could not load initial state but the task is retriable!", tid)
		time.Sleep(time.Second) 
		return false
	} else if err != nil {
		metafora.Errorf("task=%q could not load initial state. Marking done! Error: %v", tid, err)
		return true
	}
	if state == nil {
		metafora.Errorf("statestore %T returned nil state and err for task=%q - unscheduling", s.ss, tid)
		return true
	}
	if state.Code.Terminal() {
		metafora.Warnf("task=%q in terminal state %s - exiting.", tid, state.Code)
		return true
	}
	s.setState(state) 
	done = false
	for {
		metafora.Debugf("task=%q in state %s", tid, state.Code)
		msg := s.exec(state)
		newstate, ok := apply(state, msg)
		if !ok {
			metafora.Warnf("task=%q Invalid state transition=%q returned by task. Old state=%q", tid, msg.Code, state.Code)
			msg = ErrorMessage(err)
			if newstate, ok = apply(state, msg); !ok {
				metafora.Errorf("task=%q Unable to transition to error state! Exiting with state=%q", tid, state.Code)
				return state.Code.Terminal()
			}
		}
		metafora.Infof("task=%q transitioning %s --> %s --> %s", tid, state, msg, newstate)
		if err := s.ss.Store(s.task, newstate); err != nil {
			metafora.Errorf("task=%q Unable to persist state=%q. Unscheduling.", tid, newstate.Code)
			return true
		}
		state = newstate
		s.setState(state)
		if state.Code.Terminal() {
			return true
		}
		if msg.Code == Release {
			return false
		}
		select {
		case <-s.stopped:
			return false
		default:
		}
	}
}
func ( s * stateMachine ) Run ( ) ( done bool ) { s . cmds = make ( chan * Message ) go func ( ) { for { select { case m := <- s . cl . Receive ( ) : if ! m . Valid ( ) { metafora . Warnf ( " " , m ) continue } select { case s . cmds <- m : case <- s . stopped : return } case <- s . stopped : return } } } ( ) defer func ( ) { s . cl . Stop ( ) s . stop ( ) } ( ) tid := s . task . ID ( ) state , err := s . ss . Load ( s . task ) if err == ReleasableError { metafora . Errorf ( " " , tid ) time . Sleep ( time . Second ) return false } else if err != nil { metafora . Errorf ( " " , tid , err ) return true } if state == nil { metafora . Errorf ( " " , s . ss , tid ) return true } if state . Code . Terminal ( ) { metafora . Warnf ( " " , tid , state . Code ) return true } s . setState ( state ) done = false for { metafora . Debugf ( " " , tid , state . Code ) msg := s . exec ( state ) newstate , ok := apply ( state , msg ) if ! ok { metafora . Warnf ( " " , tid , msg . Code , state . Code ) msg = ErrorMessage ( err ) if newstate , ok = apply ( state , msg ) ; ! ok { metafora . Errorf ( " " , tid , state . Code ) return state . Code . Terminal ( ) } } metafora . Infof ( " " , tid , state , msg , newstate ) if err := s . ss . Store ( s . task , newstate ) ; err != nil { metafora . Errorf ( " " , tid , newstate . Code ) return true } state = newstate s . setState ( state ) if state . Code . Terminal ( ) { return true } if msg . Code == Release { return false } select { case <- s . stopped : return false default : } } }
Run the state machine enabled handler.
Run the state machine enabled handler .
// Multiplex external (Stop) messages and internal ones // Stop the command listener and internal message multiplexer when Run exits // Load the initial state // A failure to load was reported by our provided loader, but the loader believed the failure // was retriable. In most cases this will be some type of network partition or communication error, // too many file handles, etc. //defer releasing the task so other nodes don't thunder herd retrying it. // A failure to load the state for a task is *fatal* - the task will be // unscheduled and requires operator intervention to reschedule. // Note to StateStore implementors: This should not happen! Either state or // err must be non-nil. This code is simply to prevent a nil pointer panic. // for introspection/debugging // Main Statemachine Loop // Enter State // Apply Message // Save state // Set next state and loop if non-terminal // Expose the state for introspection // Exit and unschedule task on terminal state. // Release messages indicate the task should exit but not unschedule. // Alternatively Stop() may have been called but the handler may not have // returned the Release message. Always exit if we've been told to Stop() // even if the handler has returned a different Message.