diff --git a/statemachine/action/action.go b/statemachine/action/action.go new file mode 100644 index 0000000..fa80261 --- /dev/null +++ b/statemachine/action/action.go @@ -0,0 +1,17 @@ +/* +Actions for a state transitions are modelled after [action/0]. + +These transition actions can be invoked by returning them from the handlers: + - [ergo.services/actor/statemachine/cb.StateMessageHandler] + - [ergo.services/actor/statemachine/cb.StateCallHandler] + - [ergo.services/actor/statemachine/cb.EventHandler] + - [ergo.services/actor/statemachine/cb.StateEnterCallback] + +[action/0]: https://www.erlang.org/doc/apps/stdlib/gen_statem.html#t:action/0 +*/ +package action + +// Represents Erlang's action returned from state handler function. +type Action interface { + isAction() +} diff --git a/statemachine/action/timeout.go b/statemachine/action/timeout.go new file mode 100644 index 0000000..70ac038 --- /dev/null +++ b/statemachine/action/timeout.go @@ -0,0 +1,49 @@ +package action + +import ( + "time" + + "ergo.services/ergo/gen" +) + +// [StateTimeout] defines how long to wait in the current state. +// It is cancelled if statemachine change it's state before Duraion, otherwise Message +// will be sent to itself. +// +// This action is modelled after [state_timeout/0]. +// +// [state_timeout/0]: https://www.erlang.org/doc/apps/stdlib/gen_statem.html#t:state_timeout/0 +type StateTimeout struct { + Duration time.Duration + Message any +} + +func (StateTimeout) isAction() {} + +// [MessageTimeout] defines how long to wait for event. +// It is cancelled if any event arrives before Duraion, otherwise Message +// will be sent to the statemachine. +// +// This action is modelled after [event_timeout/0]. +// +// [event_timeout/0]: https://www.erlang.org/doc/apps/stdlib/gen_statem.html#t:event_timeout/0 +type MessageTimeout struct { + Duration time.Duration + Message any +} + +func (MessageTimeout) isAction() {} + +// [GenericTimeout] defines how long to wait for a named time-out event. +// When timer expires, Message will be sent to the statemachine. +// +// This action is modelled after [generic_timeout/0]. +// +// [generic_timeout/0]: https://www.erlang.org/doc/apps/stdlib/gen_statem.html#t:generic_timeout/0 +type GenericTimeout struct { + Name gen.Atom + Duration time.Duration + Message any +} + +func (GenericTimeout) isAction() {} diff --git a/statemachine/action_processor.go b/statemachine/action_processor.go new file mode 100644 index 0000000..1b4fc7a --- /dev/null +++ b/statemachine/action_processor.go @@ -0,0 +1,60 @@ +package statemachine + +import ( + "ergo.services/actor/statemachine/action" + "ergo.services/ergo/gen" +) + +func (s *StateMachine[D]) ProcessActions(actions []action.Action, state gen.Atom) { + for _, act := range actions { + switch act := act.(type) { + case action.StateTimeout: + if s.hasActiveStateTimeout() { + s.stateTimeout.cancel() + s.stateTimeout.cancelled = true + } + // Use SendAfter instead of manual goroutine + Send. + // In Ergo v3.2.0+, proc.Send() returns "not allowed" when called from + // a goroutine while the process is in Sleep state. SendAfter properly + // handles this by using the node's routing methods directly. + cancelFunc, err := s.SendAfter(s.PID(), act.Message, act.Duration) + if err != nil { + s.Log().Error("StateMachine: failed to schedule state timeout: %v", err) + continue + } + s.stateTimeout = &activeStateTimeout{ + state: state, + timeout: act, + cancel: func() { cancelFunc() }, + } + case action.GenericTimeout: + if s.hasActiveGenericTimeout(act.Name) { + s.genericTimeouts[act.Name].cancel() + s.genericTimeouts[act.Name].cancelled = true + } + // Use SendAfter instead of manual goroutine + Send. + cancelFunc, err := s.SendAfter(s.PID(), act.Message, act.Duration) + if err != nil { + s.Log().Error("StateMachine: failed to schedule generic timeout %s: %v", act.Name, err) + continue + } + s.genericTimeouts[act.Name] = &activeGenericTimeout{ + timeout: act, + cancel: func() { cancelFunc() }, + } + case action.MessageTimeout: + // Use SendAfter instead of manual goroutine + Send. + cancelFunc, err := s.SendAfter(s.PID(), act.Message, act.Duration) + if err != nil { + s.Log().Error("StateMachine: failed to schedule message timeout: %v", err) + continue + } + s.messageTimeout = &activeMessageTimeout{ + timeout: act, + cancel: func() { cancelFunc() }, + } + default: + panic("unsupported action") + } + } +} diff --git a/statemachine/behavior.go b/statemachine/behavior.go new file mode 100644 index 0000000..ef7eb10 --- /dev/null +++ b/statemachine/behavior.go @@ -0,0 +1,87 @@ +package statemachine + +import "ergo.services/ergo/gen" + +type StateMachineBehavior[D any] interface { + gen.ProcessBehavior + + Init(args ...any) (StateMachineSpec[D], error) + + HandleMessage(from gen.PID, message any) error + + HandleCall(from gen.PID, ref gen.Ref, message any) (any, error) + + HandleEvent(event gen.MessageEvent) error + + HandleInspect(from gen.PID, item ...string) map[string]string + + Terminate(reason error) + + CurrentState() gen.Atom + + SetCurrentState(gen.Atom) error + + Data() D + + SetData(data D) +} + +func (s *StateMachine[D]) HandleMessage(from gen.PID, message any) error { + s.Log().Warning("StateMachine.HandleMessage: unhandled message from %s", from) + return nil +} + +func (s *StateMachine[D]) HandleCall(from gen.PID, ref gen.Ref, request any) (any, error) { + s.Log().Warning("StateMachine.HandleCall: unhandled request from %s", from) + return nil, nil +} +func (s *StateMachine[D]) HandleEvent(message gen.MessageEvent) error { + s.Log().Warning("StateMachine.HandleEvent: unhandled event message %#v", message) + return nil +} + +func (s *StateMachine[D]) HandleInspect(from gen.PID, item ...string) map[string]string { + return nil +} + +func (s *StateMachine[D]) Terminate(reason error) {} + +func (s *StateMachine[D]) CurrentState() gen.Atom { + return s.currentState +} + +func (s *StateMachine[D]) SetCurrentState(state gen.Atom) error { + if state != s.currentState { + s.Log().Debug("StateMachine: switching to state %s", state) + oldState := s.currentState + s.currentState = state + + // If there is a state timeout set up for the new state then we have + // just registered this timeout in `ProcessActions` and we should not + // touch it. Otherwise we should cancel the active state timeout if there + // is one. + if s.hasActiveStateTimeout() && s.stateTimeout.state != state { + s.Log().Debug("StateMachine: canceling state timeout for state %s", state) + s.stateTimeout.cancel() + s.stateTimeout.cancelled = true + } + // Execute state enter callback until no new transition is triggered. + if s.stateEnterCallback != nil { + newState, newData, err := s.stateEnterCallback(oldState, state, s.data, s) + if err != nil { + return err + } + s.SetData(newData) + s.SetCurrentState(newState) + } + } + return nil +} + +func (s *StateMachine[D]) Data() D { + return s.data +} + +func (s *StateMachine[D]) SetData(data D) { + s.data = data +} diff --git a/statemachine/cb/callbacks.go b/statemachine/cb/callbacks.go new file mode 100644 index 0000000..f494987 --- /dev/null +++ b/statemachine/cb/callbacks.go @@ -0,0 +1,134 @@ +package cb + +import ( + "ergo.services/actor/statemachine/action" + "ergo.services/ergo/gen" +) + +/* +Type alias for handlers called upon messages generated by [gen.Process.Send] (MessageHandlers). + +Each message handler receives as its inputs: + - sender's [gen.PID] + - current statemachine state + - current data associated to it + - message of type M that was received + - and a pointer to statemachine process + +Following Erlang's phylosopy, callback handler returns a number of values: + - a state to which statemachine should be transitioned (newState). It may be the same as current state. + - an updated Data object + - a list of [Action]'s to perform before switching to newState + - an error object + +If err is a non-nil object, the actor terminates. You can use [gen.TerminateReasonNormal] +as usual to terminate the actor gracefully. +*/ +type StateMessageHandler[D any, M any] func( + sender gen.PID, + state gen.Atom, + data D, + message M, + p gen.Process, +) ( + newState gen.Atom, + newData D, + actions []action.Action, + err error, +) + +/* +Type alias for handlers called upon messages generated by [gen.Process.Call] (CallHandlers). + +Each message handler receives as its inputs: + - sender's [gen.PID] + - current statemachine state + - current data associated to it + - message of type M that was received + - and a pointer to statemachine process + +Following Erlang's phylosopy, callback handler returns a number of values: + - a state to which statemachine should be transitioned (newState). It may be the same as current state. + - an updated Data object + - a list of [Action]'s to perform before switching to newState + - result, that will be send back to the caller in the synchronous manner + - an error object + +If err is a non-nil object, the actor terminates. You can use [gen.TerminateReasonNormal] +as usual to terminate the actor gracefully. +*/ +type StateCallHandler[D any, M any, R any] func( + sender gen.PID, + state gen.Atom, + data D, + message M, + p gen.Process, +) ( + newState gen.Atom, + newData D, + result R, + actions []action.Action, + err error, +) + +// Type alias for event handler callbacks. +// D is the type of the data associated with the StateMachine. +// E is the type of the event. + +/* +Type alias for handlers called upon events generated by [gen.Process.SendEvent] (EventHandlers). + +Each message handler receives as its inputs: + - current statemachine state + - current data associated to it + - message of type M that was received + - and a pointer to statemachine process + +Following Erlang's phylosopy, callback handler returns a number of values: + - a state to which statemachine should be transitioned (newState). It may be the same as current state. + - an updated Data object + - a list of [Action]'s to perform before switching to newState + - an error object + +If err is a non-nil object, the actor terminates. You can use [gen.TerminateReasonNormal] +as usual to terminate the actor gracefully. +*/ +type EventHandler[D any, E any] func( + state gen.Atom, + data D, + event E, + p gen.Process, +) ( + newState gen.Atom, + newData D, + actions []action.Action, + err error, +) + +// Type alias for StateEnter callback. +// D is the type of the data associated with the StateMachine. + +/* +Following the [gen_statem]'s phylosophy, there is a handler function, that is called upon +every state change. The [state_enter/0]. + +StateEnter handler receives as its inputs: + - current statemachine state + - target statemachine state + - current data associated to it + - and a pointer to statemachine process + +Transition may be overriden by returning newState different from toState. + +[state_enter/0]: https://www.erlang.org/doc/apps/stdlib/gen_statem.html#t:state_enter/0 +*/ +type StateEnterCallback[D any] func( + fromState gen.Atom, + toState gen.Atom, + data D, + p gen.Process, +) ( + newState gen.Atom, + newData D, + err error, +) diff --git a/statemachine/compat.go b/statemachine/compat.go new file mode 100644 index 0000000..b39f0e0 --- /dev/null +++ b/statemachine/compat.go @@ -0,0 +1,33 @@ +package statemachine + +import ( + "ergo.services/actor/statemachine/action" + "ergo.services/actor/statemachine/cb" +) + +// Compatibility layer to aid migration after code reorg. +// TODO: remove after consumers complete migration + +// Deprecated: backwards compatibility alias. Use action.Action instead +type Action = action.Action + +// Deprecated: backwards compatibility alias. Use action.StateTimeout instead +type StateTimeout = action.StateTimeout + +// Deprecated: backwards compatibility alias. Use action.MessageTimeout instead +type MessageTimeout = action.MessageTimeout + +// Deprecated: backwards compatibility alias. Use action.GenericTimeout instead +type GenericTimeout = action.GenericTimeout + +// Deprecated: backwards compatibility alias. Use cb.StateMessageHandler instead +type StateMessageHandler[D any, M any] = cb.StateMessageHandler[D, M] + +// Deprecated: backwards compatibility alias. Use cb.StateCallHandler instead +type StateCallHandler[D any, M any, R any] = cb.StateCallHandler[D, M, R] + +// Deprecated: backwards compatibility alias. Use cb.EventHandler instead +type EventHandler[D any, E any] = cb.EventHandler[D, E] + +// Deprecated: backwards compatibility alias. Use cb.StateEnterCallback instead +type StateEnterCallback[D any] = cb.StateEnterCallback[D] diff --git a/statemachine/doc.go b/statemachine/doc.go new file mode 100644 index 0000000..cd5795e --- /dev/null +++ b/statemachine/doc.go @@ -0,0 +1,48 @@ +/* +Statemachine package and corresponding actor is modelled after Erlang's [gen_statem]. + +The [StateMachine] is a [gen.ProcessBehavior] implementation. By Erlang ideology, +[gen_statem] has a Data associated with it. Here it is achieved by using a generic type +parameter D. A difference with the Erlang implementation is that the data and the new +state are not returned from the state callback functions, instead they are part of the +mutable state of the actor. + +States are of type [gen.Atom] (to make it feel even more Erlang-ish) and the +[StateMachineSpec] is defined by means of [NewStateMachineSpec] function. + +The Init function implementation is mandatory and returns the +[statemachine.StateMachineSpec] of the [StateMachine]. There are several helper functions +and types to aid capturing the type of the message being processed by the callbacks. Using +functional option pattern also avoids unnecessary copying of the spec. + +When attempting an invalid state transition, following Erlang/OTP's "let it crash" +philosophy, the process terminates. + +Quick glance at a statemachine creation: + + type yourFsmData struct {} + type YourStatemachine struct { + statemachine.StateMachine[yourFsmData] + } + + const ( + initialState = gen.Atom("init") + otherState = gen.Atom("other") + ) + + func (s *YourStatemachine) Init(args ...any) (spec statemachine.StateMachineSpec[yourFsmData], err error) { + spec:= statemachine.NewStateMachineSpec( + initialState, + statemachine.WithStateEnterCallback(...), + statemachine.WithStateMessageHandler(initialState, ...), + statemachine.WithStateMessageHandler(otherState, ...), + ) + + return spec, nil + } + +For specific details see below. + +[gen_statem]: https://www.erlang.org/doc/apps/stdlib/gen_statem.html +*/ +package statemachine diff --git a/statemachine/internals.go b/statemachine/internals.go new file mode 100644 index 0000000..cfff0bf --- /dev/null +++ b/statemachine/internals.go @@ -0,0 +1,167 @@ +package statemachine + +import ( + "fmt" + "reflect" + + "ergo.services/ergo/gen" +) + +func (s *StateMachine[D]) hasActiveStateTimeout() bool { + return s.stateTimeout != nil && !s.stateTimeout.cancelled +} + +func (s *StateMachine[D]) hasActiveMessageTimeout() bool { + return s.messageTimeout != nil && !s.messageTimeout.cancelled +} + +func (s *StateMachine[D]) hasActiveGenericTimeout(name gen.Atom) bool { + if timeout, exists := s.genericTimeouts[name]; exists { + return !timeout.cancelled + } + return false +} + +func (s *StateMachine[D]) lookupMessageHandler(messageType string) (any, bool) { + if stateMessageHandlers, exists := s.stateMessageHandlers[s.currentState]; exists == true { + if callback, exists := stateMessageHandlers[messageType]; exists == true { + return callback, true + } + } + return nil, false +} + +func (s *StateMachine[D]) invokeMessageHandler(handler any, message *gen.MailboxMessage) error { + stateMachineValue := reflect.ValueOf(s) + callbackValue := reflect.ValueOf(handler) + fromValue := reflect.ValueOf(message.From) + stateValue := reflect.ValueOf(s.currentState) + dataValue := reflect.ValueOf(s.Data()) + msgValue := reflect.ValueOf(message.Message) + messageType := reflect.TypeOf(message).String() + + results := callbackValue.Call([]reflect.Value{fromValue, stateValue, dataValue, msgValue, stateMachineValue}) + + validateResultSize(results, 4, messageType) + if isError, err := resultIsError(results); isError == true { + return err + } + + return updateStateMachineWithResults(stateMachineValue, results) +} + +func (s *StateMachine[D]) lookupCallHandler(messageType string) (any, bool) { + if stateCallHandlers, exists := s.stateCallHandlers[s.currentState]; exists == true { + if callback, exists := stateCallHandlers[messageType]; exists == true { + return callback, true + } + } + return nil, false +} + +func (s *StateMachine[D]) invokeCallHandler(handler any, message *gen.MailboxMessage) (any, error) { + stateMachineValue := reflect.ValueOf(s) + callbackValue := reflect.ValueOf(handler) + fromValue := reflect.ValueOf(message.From) + stateValue := reflect.ValueOf(s.currentState) + dataValue := reflect.ValueOf(s.Data()) + msgValue := reflect.ValueOf(message.Message) + messageType := reflect.TypeOf(message).String() + + results := callbackValue.Call([]reflect.Value{fromValue, stateValue, dataValue, msgValue, stateMachineValue}) + + validateResultSize(results, 5, messageType) + if isError, err := resultIsError(results); isError == true { + return nil, err + } + err := updateStateMachineWithResults(stateMachineValue, results) + if err != nil { + return nil, err + } + result := results[2].Interface() + return result, nil +} + +func (s *StateMachine[D]) invokeEventHandler(handler any, message *gen.MessageEvent) error { + stateMachineValue := reflect.ValueOf(s) + callbackValue := reflect.ValueOf(handler) + stateValue := reflect.ValueOf(s.currentState) + dataValue := reflect.ValueOf(s.Data()) + msgValue := reflect.ValueOf(message.Message) + messageType := reflect.TypeOf(message).String() + + results := callbackValue.Call([]reflect.Value{stateValue, dataValue, msgValue, stateMachineValue}) + + validateResultSize(results, 4, messageType) + if isError, err := resultIsError(results); isError == true { + return err + } + updateStateMachineWithResults(stateMachineValue, results) + + return nil +} + +func validateResultSize(results []reflect.Value, expectedSize int, messageType string) { + if len(results) != expectedSize { + panic(fmt.Sprintf("StateMachine terminated. Panic reason: unexpected "+ + "error when invoking call handler for %s", messageType)) + } +} + +func resultIsError(results []reflect.Value) (bool, error) { + errIndex := len(results) - 1 + if !results[errIndex].IsNil() { + err := results[errIndex].Interface().(error) + return true, err + } + return false, nil +} + +func updateStateMachineWithResults(s reflect.Value, results []reflect.Value) error { + // Check if any actions were returned. MessageHandler and EventHandler have + // the result tuple (gen.Atom, D, []Action, error) with the actions at index + // 2. CallHandler has the result typle (gen.Atom, D, R, []Action, error) + // with the actions at index 3. + var actionsIndex int + hasResult := len(results) == 5 + if hasResult { + actionsIndex = 3 + } else { + actionsIndex = 2 + } + if !isSliceNilOrEmpty(results[actionsIndex]) { + processActionsMethod := s.MethodByName("ProcessActions") + if processActionsMethod.IsNil() { + } + processActionsMethod.Call([]reflect.Value{results[actionsIndex], results[0]}) + } + + // Update the data + setDataMethod := s.MethodByName("SetData") + setDataMethod.Call([]reflect.Value{results[1]}) + + // It is important that we set the state last as this can potentially trigger + // a state enter callback. By design state enter callbacks are triggered + // after setting up state timeouts as state timeouts are tied to te state + // they are defined for. A state enter callback could transition to another + // state which then will cancel the state timeout. + setCurrentStateMethod := s.MethodByName("SetCurrentState") + err := setCurrentStateMethod.Call([]reflect.Value{results[0]})[0] + if !err.IsNil() { + return err.Interface().(error) + } + + return nil +} + +func isSliceNilOrEmpty(resultValue reflect.Value) bool { + if resultValue.IsNil() { + return true + } + + if resultValue.Len() == 0 { + return true + } + + return false +} diff --git a/statemachine/process_behavior.go b/statemachine/process_behavior.go new file mode 100644 index 0000000..e88933a --- /dev/null +++ b/statemachine/process_behavior.go @@ -0,0 +1,259 @@ +package statemachine + +import ( + "fmt" + "reflect" + "runtime" + "strings" + + "ergo.services/actor/statemachine/cb" + "ergo.services/ergo/gen" + "ergo.services/ergo/lib" +) + +type StateMachine[D any] struct { + gen.Process + + behavior StateMachineBehavior[D] + mailbox gen.ProcessMailbox + + // The specification for the StateMachine + spec StateMachineSpec[D] + + // The state the StateMachine is currently in + currentState gen.Atom + + // The data associated with the StateMachine + data D + + // stateMessageHandlers maps states to the (asynchronous) handlers for the state. + // Key: State (gen.Atom) - The state for which the handler is registered. + // Value: Map of message type to the handler for that message. + // Key: The type of the message received (String). + // Value: The message handler (any). There is a compile-time guarantee + // that the handler is of type StateMessageHandler[D, M]. + stateMessageHandlers map[gen.Atom]map[string]any + + // stateCallHandlers maps states to the (synchronous) handlers for the state. + // Key: State (gen.Atom) - The state for which the handler is registered. + // Value: Map of message type to the handler for that message. + // Key: The type of the message received (String). + // Value: The message handler (any). There is a compile-time guarantee + // that the handler is of type StateCallHandler[D, M, R]. + stateCallHandlers map[gen.Atom]map[string]any + + // eventHandlers maps events to the handler for the event. + // Key: Event name (gen.Atom) - The name of the event + // Value: The event handler (any). There is a compile-time guarantee that + // the handler is of type EventHandler[D, E] + eventHandlers map[gen.Event]any + + // Callback that is invoked immediately after every state change. If no + // callback is registered stateEnterCallback is nil. + stateEnterCallback cb.StateEnterCallback[D] + + // Pointer to the most recently configured state timeout. + stateTimeout *activeStateTimeout + + // Pointer to the most recently configured message timeout. + messageTimeout *activeMessageTimeout + + // genericTimeouts maps the name of a generic timeout to the timeout + genericTimeouts map[gen.Atom]*activeGenericTimeout +} + +// +// ProcessBehavior implementation +// + +func (s *StateMachine[D]) ProcessInit(process gen.Process, args ...any) (rr error) { + var ok bool + + if s.behavior, ok = process.Behavior().(StateMachineBehavior[D]); ok == false { + unknown := strings.TrimPrefix(reflect.TypeOf(process.Behavior()).String(), "*") + return fmt.Errorf("ProcessInit: not a StateMachineBehavior %s", unknown) + } + + s.Process = process + s.mailbox = process.Mailbox() + + if lib.Recover() { + defer func() { + if r := recover(); r != nil { + pc, fn, line, _ := runtime.Caller(2) + s.Log().Panic("StateMachine initialization failed. Panic reason: %#v at %s[%s:%d]", + r, runtime.FuncForPC(pc).Name(), fn, line) + rr = gen.TerminateReasonPanic + } + }() + } + + spec, err := s.behavior.Init(args...) + if err != nil { + return err + } + + s.currentState = spec.initialState + s.data = spec.data + s.stateMessageHandlers = spec.stateMessageHandlers + s.stateCallHandlers = spec.stateCallHandlers + s.eventHandlers = spec.eventHandlers + s.stateEnterCallback = spec.stateEnterCallback + s.genericTimeouts = make(map[gen.Atom]*activeGenericTimeout) + + // Send a message to ourselves to start monitoring events if there are + // event handlers registerd. + if len(s.eventHandlers) > 0 { + s.Send(s.PID(), startMonitoringEvents{}) + } + s.Log().Debug("StateMachine: started in state %s", s.currentState) + + return nil +} + +func (s *StateMachine[D]) ProcessRun() (rr error) { + var message *gen.MailboxMessage + + if lib.Recover() { + defer func() { + if r := recover(); r != nil { + pc, fn, line, _ := runtime.Caller(2) + s.Log().Panic("StateMachine terminated. Panic reason: %#v at %s[%s:%d]", + r, runtime.FuncForPC(pc).Name(), fn, line) + rr = gen.TerminateReasonPanic + } + }() + } + + for { + if s.State() != gen.ProcessStateRunning { + // process was killed by the node. + return gen.TerminateReasonKill + } + + if message != nil { + gen.ReleaseMailboxMessage(message) + message = nil + } + + for { + // check queues + msg, ok := s.mailbox.Urgent.Pop() + if ok { + // got new urgent message. handle it + message = msg.(*gen.MailboxMessage) + break + } + + msg, ok = s.mailbox.System.Pop() + if ok { + // got new system message. handle it + message = msg.(*gen.MailboxMessage) + break + } + + msg, ok = s.mailbox.Main.Pop() + if ok { + // got new regular message. handle it + message = msg.(*gen.MailboxMessage) + break + } + + if _, ok := s.mailbox.Log.Pop(); ok { + panic("statemachne process can not be a logger") + } + + // no messages in the mailbox + return nil + } + + // Any message should cancel the active message timeout + if s.hasActiveMessageTimeout() { + s.messageTimeout.cancel() + s.messageTimeout.cancelled = true + } + + switch message.Type { + case gen.MailboxMessageTypeRegular: + switch message.Message.(type) { + case startMonitoringEvents: + // start monitoring + for event := range s.eventHandlers { + if _, err := s.MonitorEvent(event); err != nil { + panic(fmt.Sprintf("Error monitoring event: %v.", err)) + } + } + s.Log().Debug("StateMachine: monitoring events") + return nil + + default: + // check if there is a handler for the message in the current state + messageType := reflect.TypeOf(message.Message).String() + handler, ok := s.lookupMessageHandler(messageType) + if ok == false { + return fmt.Errorf("No handler for message %s in state %s", messageType, s.currentState) + } + return s.invokeMessageHandler(handler, message) + } + + case gen.MailboxMessageTypeRequest: + var reason error + var result any + + // check if there is a handler for the call in the current state + messageType := reflect.TypeOf(message.Message).String() + handler, ok := s.lookupCallHandler(messageType) + if ok == false { + return fmt.Errorf("No handler for message %s in state %s", messageType, s.currentState) + } + result, reason = s.invokeCallHandler(handler, message) + + if reason != nil { + // if reason is "normal" and we got response - send it before termination + if reason == gen.TerminateReasonNormal { + s.SendResponse(message.From, message.Ref, result) + } + return reason + } + // Note: we do not support async handling of sync request at the moment + s.SendResponse(message.From, message.Ref, result) + + case gen.MailboxMessageTypeEvent: + event := message.Message.(gen.MessageEvent) + handler, exists := s.eventHandlers[event.Event] + if exists == false { + return fmt.Errorf("No handler for event %v", event) + } + return s.invokeEventHandler(handler, &event) + + case gen.MailboxMessageTypeExit: + switch exit := message.Message.(type) { + case gen.MessageExitPID: + return fmt.Errorf("%s: %w", exit.PID, exit.Reason) + + case gen.MessageExitProcessID: + return fmt.Errorf("%s: %w", exit.ProcessID, exit.Reason) + + case gen.MessageExitAlias: + return fmt.Errorf("%s: %w", exit.Alias, exit.Reason) + + case gen.MessageExitEvent: + return fmt.Errorf("%s: %w", exit.Event, exit.Reason) + + case gen.MessageExitNode: + return fmt.Errorf("%s: %w", exit.Name, gen.ErrNoConnection) + + default: + panic(fmt.Sprintf("unknown exit message: %#v", exit)) + } + + case gen.MailboxMessageTypeInspect: + result := s.behavior.HandleInspect(message.From, message.Message.([]string)...) + s.SendResponse(message.From, message.Ref, result) + } + } +} + +func (s *StateMachine[D]) ProcessTerminate(reason error) { + s.behavior.Terminate(reason) +} diff --git a/statemachine/spec.go b/statemachine/spec.go new file mode 100644 index 0000000..a6de419 --- /dev/null +++ b/statemachine/spec.go @@ -0,0 +1,137 @@ +package statemachine + +import ( + "reflect" + + "ergo.services/actor/statemachine/cb" + "ergo.services/ergo/gen" +) + +type StateMachineSpec[D any] struct { + initialState gen.Atom + data D + stateMessageHandlers map[gen.Atom]map[string]any + stateCallHandlers map[gen.Atom]map[string]any + eventHandlers map[gen.Event]any + stateEnterCallback cb.StateEnterCallback[D] +} + +// [StateMachineSpec] fragments defined in a functional manner. +type Option[D any] func(*StateMachineSpec[D]) + +/* +Use this function inside your stattemachine Init() to create [StateMachineSpec]. + +States of the machine are denoted by [gen.Atom] values. + +First argument is the initial state, later arguments placed in functional manner define +different aspects of the statemachine: + - [WithData]() specifies data object for the statemachine instance + - [WithStateEnterCallback]() defines the callback invoked whenever statemachine changes its state + +Special methods to bind current state, incoming message and callback handler together: + - [WithStateMessageHandler] + - [WithStateCallHandler] + - [WithEventHandler] + +Example: + + var ( + initialState = gen.Atom("init") + otherState = gen.Atom("other") + ) + + spec := statemachine.NewStateMachineSpec( + initialState, + statemachine.WithStateEnterCallback(onEnterNewState), + statemachine.WithStateMessageHandler(initialState, onSomeMessage), + statemachine.WithStateMessageHandler(initialState, onAnotherMessage), + ... + statemachine.WithStateMessageHandler(anotherState, ...), + ) +*/ +func NewStateMachineSpec[D any](initialState gen.Atom, options ...Option[D]) StateMachineSpec[D] { + spec := StateMachineSpec[D]{ + initialState: initialState, + stateMessageHandlers: make(map[gen.Atom]map[string]any), + stateCallHandlers: make(map[gen.Atom]map[string]any), + eventHandlers: make(map[gen.Event]any), + } + for _, opt := range options { + opt(&spec) + } + return spec +} + +// Assigns data object of type D to the statemachine +func WithData[D any](data D) Option[D] { + return func(s *StateMachineSpec[D]) { + s.data = data + } +} + +// Binds a callback to the event of changing state +func WithStateEnterCallback[D any](callback cb.StateEnterCallback[D]) Option[D] { + return func(s *StateMachineSpec[D]) { + s.stateEnterCallback = callback + } +} + +/* +Binds a callback function `handler` to statemachine's state `state` + +The specified handler will be executed if: + - statemachine is in the state `state` and + - statemachine actor receives a message of type M sent asynchronously via [gen.Process.Send] + +To handle multiple message types while statemachine is in state `state`, use multiple [WithStateMessageHandler] calls with same state and different [StateMessageHandler]'s. + +For handler implementation specifics see [StateMessageHandler]. +*/ +func WithStateMessageHandler[D any, M any](state gen.Atom, handler cb.StateMessageHandler[D, M]) Option[D] { + messageType := reflect.TypeOf((*M)(nil)).Elem().String() + return func(s *StateMachineSpec[D]) { + if _, exists := s.stateMessageHandlers[state]; exists == false { + s.stateMessageHandlers[state] = make(map[string]any) + } + s.stateMessageHandlers[state][messageType] = handler + } +} + +/* +Binds a callback function `handler` to statemachine's state `state` + +The specified handler will be executed if: + - statemachine is in the state `state` and + - statemachine actor receives a message of type M sent synchronously via [gen.Process.Call] + +To handle multiple message types while statemachine is in state `state`, use multiple [WithStateCallHandler] calls with same state and different [StateCallHandler]'s. + +For handler implementation specifics see [StateCallHandler]. +*/ +func WithStateCallHandler[D any, M any, R any](state gen.Atom, handler cb.StateCallHandler[D, M, R]) Option[D] { + messageType := reflect.TypeOf((*M)(nil)).Elem().String() + return func(s *StateMachineSpec[D]) { + if _, exists := s.stateCallHandlers[state]; exists == false { + s.stateCallHandlers[state] = make(map[string]any) + } + s.stateCallHandlers[state][messageType] = handler + } +} + +/* +Binds a callback function `handler` to statemachine's state `state` + +The specified handler will be executed if: + - statemachine is in the state `state` and + - statemachine actor receives an event of type E sent synchronously via [gen.Process.SendEvent] + +To handle multiple event types while statemachine is in state `state`, use multiple [WithEventHandler] calls with same state and different [EventHandler]'s. + +For handler implementation specifics see [EventHandler]. +*/ +func WithEventHandler[D any, E any](event gen.Event, handler cb.EventHandler[D, E]) Option[D] { + return func(s *StateMachineSpec[D]) { + s.eventHandlers[event] = handler + } +} diff --git a/statemachine/statemachine.go b/statemachine/statemachine.go deleted file mode 100644 index a7ea6cd..0000000 --- a/statemachine/statemachine.go +++ /dev/null @@ -1,693 +0,0 @@ -package statemachine - -import ( - "fmt" - "reflect" - "runtime" - "strings" - "time" - - "ergo.services/ergo/gen" - "ergo.services/ergo/lib" -) - -type StateMachineBehavior[D any] interface { - gen.ProcessBehavior - - Init(args ...any) (StateMachineSpec[D], error) - - HandleMessage(from gen.PID, message any) error - - HandleCall(from gen.PID, ref gen.Ref, message any) (any, error) - - HandleEvent(event gen.MessageEvent) error - - HandleInspect(from gen.PID, item ...string) map[string]string - - Terminate(reason error) - - CurrentState() gen.Atom - - SetCurrentState(gen.Atom) error - - Data() D - - SetData(data D) -} - -type StateMachine[D any] struct { - gen.Process - - behavior StateMachineBehavior[D] - mailbox gen.ProcessMailbox - - // The specification for the StateMachine - spec StateMachineSpec[D] - - // The state the StateMachine is currently in - currentState gen.Atom - - // The data associated with the StateMachine - data D - - // stateMessageHandlers maps states to the (asynchronous) handlers for the state. - // Key: State (gen.Atom) - The state for which the handler is registered. - // Value: Map of message type to the handler for that message. - // Key: The type of the message received (String). - // Value: The message handler (any). There is a compile-time guarantee - // that the handler is of type StateMessageHandler[D, M]. - stateMessageHandlers map[gen.Atom]map[string]any - - // stateCallHandlers maps states to the (synchronous) handlers for the state. - // Key: State (gen.Atom) - The state for which the handler is registered. - // Value: Map of message type to the handler for that message. - // Key: The type of the message received (String). - // Value: The message handler (any). There is a compile-time guarantee - // that the handler is of type StateCallHandler[D, M, R]. - stateCallHandlers map[gen.Atom]map[string]any - - // eventHandlers maps events to the handler for the event. - // Key: Event name (gen.Atom) - The name of the event - // Value: The event handler (any). There is a compile-time guarantee that - // the handler is of type EventHandler[D, E] - eventHandlers map[gen.Event]any - - // Callback that is invoked immediately after every state change. If no - // callback is registered stateEnterCallback is nil. - stateEnterCallback StateEnterCallback[D] - - // Pointer to the most recently configured state timeout. - stateTimeout *ActiveStateTimeout - - // Pointer to the most recently configured message timeout. - messageTimeout *ActiveMessageTimeout - - // genericTimeouts maps the name of a generic timeout to the timeout - genericTimeouts map[gen.Atom]*ActiveGenericTimeout -} - -type Action interface { - isAction() -} - -type StateTimeout struct { - Duration time.Duration - Message any -} - -func (StateTimeout) isAction() {} - -type ActiveStateTimeout struct { - state gen.Atom - timeout StateTimeout - cancel func() - cancelled bool -} - -type GenericTimeout struct { - Name gen.Atom - Duration time.Duration - Message any -} - -func (GenericTimeout) isAction() {} - -type ActiveGenericTimeout struct { - timeout GenericTimeout - cancel func() - cancelled bool -} - -type MessageTimeout struct { - Duration time.Duration - Message any -} - -func (MessageTimeout) isAction() {} - -type ActiveMessageTimeout struct { - timeout MessageTimeout - cancel func() - cancelled bool -} - -// Type alias for MessageHandler callbacks. -// D is the type of the data associated with the StateMachine. -// M is the type of the message this handler accepts. -type StateMessageHandler[D any, M any] func(gen.PID, gen.Atom, D, M, gen.Process) (gen.Atom, D, []Action, error) - -// Type alias for CallHandler callbacks. -// D is the type of the data associated with the StateMachine. -// M is the type of the message this handler accepts. -// R is the type of the result value. -type StateCallHandler[D any, M any, R any] func(gen.PID, gen.Atom, D, M, gen.Process) (gen.Atom, D, R, []Action, error) - -// Type alias for event handler callbacks. -// D is the type of the data associated with the StateMachine. -// E is the type of the event. -type EventHandler[D any, E any] func(gen.Atom, D, E, gen.Process) (gen.Atom, D, []Action, error) - -// Type alias for StateEnter callback. -// D is the type of the data associated with the StateMachine. -type StateEnterCallback[D any] func(gen.Atom, gen.Atom, D, gen.Process) (gen.Atom, D, error) - -type StateMachineSpec[D any] struct { - initialState gen.Atom - data D - stateMessageHandlers map[gen.Atom]map[string]any - stateCallHandlers map[gen.Atom]map[string]any - eventHandlers map[gen.Event]any - stateEnterCallback StateEnterCallback[D] -} - -type Option[D any] func(*StateMachineSpec[D]) - -func NewStateMachineSpec[D any](initialState gen.Atom, options ...Option[D]) StateMachineSpec[D] { - spec := StateMachineSpec[D]{ - initialState: initialState, - stateMessageHandlers: make(map[gen.Atom]map[string]any), - stateCallHandlers: make(map[gen.Atom]map[string]any), - eventHandlers: make(map[gen.Event]any), - } - for _, opt := range options { - opt(&spec) - } - return spec -} - -func WithData[D any](data D) Option[D] { - return func(s *StateMachineSpec[D]) { - s.data = data - } -} - -func WithStateMessageHandler[D any, M any](state gen.Atom, handler StateMessageHandler[D, M]) Option[D] { - messageType := reflect.TypeOf((*M)(nil)).Elem().String() - return func(s *StateMachineSpec[D]) { - if _, exists := s.stateMessageHandlers[state]; exists == false { - s.stateMessageHandlers[state] = make(map[string]any) - } - s.stateMessageHandlers[state][messageType] = handler - } -} - -func WithStateCallHandler[D any, M any, R any](state gen.Atom, handler StateCallHandler[D, M, R]) Option[D] { - messageType := reflect.TypeOf((*M)(nil)).Elem().String() - return func(s *StateMachineSpec[D]) { - if _, exists := s.stateCallHandlers[state]; exists == false { - s.stateCallHandlers[state] = make(map[string]any) - } - s.stateCallHandlers[state][messageType] = handler - } -} - -func WithStateEnterCallback[D any](callback StateEnterCallback[D]) Option[D] { - return func(s *StateMachineSpec[D]) { - s.stateEnterCallback = callback - } -} - -func WithEventHandler[D any, E any](event gen.Event, handler EventHandler[D, E]) Option[D] { - return func(s *StateMachineSpec[D]) { - s.eventHandlers[event] = handler - } -} - -func (s *StateMachine[D]) CurrentState() gen.Atom { - return s.currentState -} - -func (s *StateMachine[D]) SetCurrentState(state gen.Atom) error { - if state != s.currentState { - s.Log().Debug("StateMachine: switching to state %s", state) - oldState := s.currentState - s.currentState = state - - // If there is a state timeout set up for the new state then we have - // just registered this timeout in `ProcessActions` and we should not - // touch it. Otherwise we should cancel the active state timeout if there - // is one. - if s.hasActiveStateTimeout() && s.stateTimeout.state != state { - s.Log().Debug("StateMachine: canceling state timeout for state %s", state) - s.stateTimeout.cancel() - s.stateTimeout.cancelled = true - } - // Execute state enter callback until no new transition is triggered. - if s.stateEnterCallback != nil { - newState, newData, err := s.stateEnterCallback(oldState, state, s.data, s) - if err != nil { - return err - } - s.SetData(newData) - s.SetCurrentState(newState) - } - } - return nil -} - -func (s *StateMachine[D]) Data() D { - return s.data -} - -func (s *StateMachine[D]) SetData(data D) { - s.data = data -} - -func (s *StateMachine[D]) hasActiveStateTimeout() bool { - return s.stateTimeout != nil && !s.stateTimeout.cancelled -} - -func (s *StateMachine[D]) hasActiveMessageTimeout() bool { - return s.messageTimeout != nil && !s.messageTimeout.cancelled -} - -func (s *StateMachine[D]) hasActiveGenericTimeout(name gen.Atom) bool { - if timeout, exists := s.genericTimeouts[name]; exists { - return !timeout.cancelled - } - return false -} - -type startMonitoringEvents struct{} - -// -// ProcessBehavior implementation -// - -func (s *StateMachine[D]) ProcessInit(process gen.Process, args ...any) (rr error) { - var ok bool - - if s.behavior, ok = process.Behavior().(StateMachineBehavior[D]); ok == false { - unknown := strings.TrimPrefix(reflect.TypeOf(process.Behavior()).String(), "*") - return fmt.Errorf("ProcessInit: not a StateMachineBehavior %s", unknown) - } - - s.Process = process - s.mailbox = process.Mailbox() - - if lib.Recover() { - defer func() { - if r := recover(); r != nil { - pc, fn, line, _ := runtime.Caller(2) - s.Log().Panic("StateMachine initialization failed. Panic reason: %#v at %s[%s:%d]", - r, runtime.FuncForPC(pc).Name(), fn, line) - rr = gen.TerminateReasonPanic - } - }() - } - - spec, err := s.behavior.Init(args...) - if err != nil { - return err - } - - s.currentState = spec.initialState - s.data = spec.data - s.stateMessageHandlers = spec.stateMessageHandlers - s.stateCallHandlers = spec.stateCallHandlers - s.eventHandlers = spec.eventHandlers - s.stateEnterCallback = spec.stateEnterCallback - s.genericTimeouts = make(map[gen.Atom]*ActiveGenericTimeout) - - // Send a message to ourselves to start monitoring events if there are - // event handlers registerd. - if len(s.eventHandlers) > 0 { - s.Send(s.PID(), startMonitoringEvents{}) - } - s.Log().Debug("StateMachine: started in state %s", s.currentState) - - return nil -} - -func (s *StateMachine[D]) ProcessRun() (rr error) { - var message *gen.MailboxMessage - - if lib.Recover() { - defer func() { - if r := recover(); r != nil { - pc, fn, line, _ := runtime.Caller(2) - s.Log().Panic("StateMachine terminated. Panic reason: %#v at %s[%s:%d]", - r, runtime.FuncForPC(pc).Name(), fn, line) - rr = gen.TerminateReasonPanic - } - }() - } - - for { - if s.State() != gen.ProcessStateRunning { - // process was killed by the node. - return gen.TerminateReasonKill - } - - if message != nil { - gen.ReleaseMailboxMessage(message) - message = nil - } - - for { - // check queues - msg, ok := s.mailbox.Urgent.Pop() - if ok { - // got new urgent message. handle it - message = msg.(*gen.MailboxMessage) - break - } - - msg, ok = s.mailbox.System.Pop() - if ok { - // got new system message. handle it - message = msg.(*gen.MailboxMessage) - break - } - - msg, ok = s.mailbox.Main.Pop() - if ok { - // got new regular message. handle it - message = msg.(*gen.MailboxMessage) - break - } - - if _, ok := s.mailbox.Log.Pop(); ok { - panic("statemachne process can not be a logger") - } - - // no messages in the mailbox - return nil - } - - // Any message should cancel the active message timeout - if s.hasActiveMessageTimeout() { - s.messageTimeout.cancel() - s.messageTimeout.cancelled = true - } - - switch message.Type { - case gen.MailboxMessageTypeRegular: - switch message.Message.(type) { - case startMonitoringEvents: - // start monitoring - for event := range s.eventHandlers { - if _, err := s.MonitorEvent(event); err != nil { - panic(fmt.Sprintf("Error monitoring event: %v.", err)) - } - } - s.Log().Debug("StateMachine: monitoring events") - return nil - - default: - // check if there is a handler for the message in the current state - messageType := reflect.TypeOf(message.Message).String() - handler, ok := s.lookupMessageHandler(messageType) - if ok == false { - return fmt.Errorf("No handler for message %s in state %s", messageType, s.currentState) - } - return s.invokeMessageHandler(handler, message) - } - - case gen.MailboxMessageTypeRequest: - var reason error - var result any - - // check if there is a handler for the call in the current state - messageType := reflect.TypeOf(message.Message).String() - handler, ok := s.lookupCallHandler(messageType) - if ok == false { - return fmt.Errorf("No handler for message %s in state %s", messageType, s.currentState) - } - result, reason = s.invokeCallHandler(handler, message) - - if reason != nil { - // if reason is "normal" and we got response - send it before termination - if reason == gen.TerminateReasonNormal { - s.SendResponse(message.From, message.Ref, result) - } - return reason - } - // Note: we do not support async handling of sync request at the moment - s.SendResponse(message.From, message.Ref, result) - - case gen.MailboxMessageTypeEvent: - event := message.Message.(gen.MessageEvent) - handler, exists := s.eventHandlers[event.Event] - if exists == false { - return fmt.Errorf("No handler for event %v", event) - } - return s.invokeEventHandler(handler, &event) - - case gen.MailboxMessageTypeExit: - switch exit := message.Message.(type) { - case gen.MessageExitPID: - return fmt.Errorf("%s: %w", exit.PID, exit.Reason) - - case gen.MessageExitProcessID: - return fmt.Errorf("%s: %w", exit.ProcessID, exit.Reason) - - case gen.MessageExitAlias: - return fmt.Errorf("%s: %w", exit.Alias, exit.Reason) - - case gen.MessageExitEvent: - return fmt.Errorf("%s: %w", exit.Event, exit.Reason) - - case gen.MessageExitNode: - return fmt.Errorf("%s: %w", exit.Name, gen.ErrNoConnection) - - default: - panic(fmt.Sprintf("unknown exit message: %#v", exit)) - } - - case gen.MailboxMessageTypeInspect: - result := s.behavior.HandleInspect(message.From, message.Message.([]string)...) - s.SendResponse(message.From, message.Ref, result) - } - } -} - -func (s *StateMachine[D]) ProcessTerminate(reason error) { - s.behavior.Terminate(reason) -} - -// -// StateMachineBehavior default callbacks -// - -func (s *StateMachine[D]) HandleMessage(from gen.PID, message any) error { - s.Log().Warning("StateMachine.HandleMessage: unhandled message from %s", from) - return nil -} - -func (s *StateMachine[D]) HandleCall(from gen.PID, ref gen.Ref, request any) (any, error) { - s.Log().Warning("StateMachine.HandleCall: unhandled request from %s", from) - return nil, nil -} -func (s *StateMachine[D]) HandleEvent(message gen.MessageEvent) error { - s.Log().Warning("StateMachine.HandleEvent: unhandled event message %#v", message) - return nil -} - -func (s *StateMachine[D]) HandleInspect(from gen.PID, item ...string) map[string]string { - return nil -} - -func (s *StateMachine[D]) Terminate(reason error) {} - -// -// Internals -// - -func (s *StateMachine[D]) ProcessActions(actions []Action, state gen.Atom) { - for _, action := range actions { - switch action := action.(type) { - case StateTimeout: - if s.hasActiveStateTimeout() { - s.stateTimeout.cancel() - s.stateTimeout.cancelled = true - } - // Use SendAfter instead of manual goroutine + Send. - // In Ergo v3.2.0+, proc.Send() returns "not allowed" when called from - // a goroutine while the process is in Sleep state. SendAfter properly - // handles this by using the node's routing methods directly. - cancelFunc, err := s.SendAfter(s.PID(), action.Message, action.Duration) - if err != nil { - s.Log().Error("StateMachine: failed to schedule state timeout: %v", err) - continue - } - s.stateTimeout = &ActiveStateTimeout{ - state: state, - timeout: action, - cancel: func() { cancelFunc() }, - } - case GenericTimeout: - if s.hasActiveGenericTimeout(action.Name) { - s.genericTimeouts[action.Name].cancel() - s.genericTimeouts[action.Name].cancelled = true - } - // Use SendAfter instead of manual goroutine + Send. - cancelFunc, err := s.SendAfter(s.PID(), action.Message, action.Duration) - if err != nil { - s.Log().Error("StateMachine: failed to schedule generic timeout %s: %v", action.Name, err) - continue - } - s.genericTimeouts[action.Name] = &ActiveGenericTimeout{ - timeout: action, - cancel: func() { cancelFunc() }, - } - case MessageTimeout: - // Use SendAfter instead of manual goroutine + Send. - cancelFunc, err := s.SendAfter(s.PID(), action.Message, action.Duration) - if err != nil { - s.Log().Error("StateMachine: failed to schedule message timeout: %v", err) - continue - } - s.messageTimeout = &ActiveMessageTimeout{ - timeout: action, - cancel: func() { cancelFunc() }, - } - default: - panic("unsupported action") - } - } -} - -func (s *StateMachine[D]) lookupMessageHandler(messageType string) (any, bool) { - if stateMessageHandlers, exists := s.stateMessageHandlers[s.currentState]; exists == true { - if callback, exists := stateMessageHandlers[messageType]; exists == true { - return callback, true - } - } - return nil, false -} - -func (s *StateMachine[D]) invokeMessageHandler(handler any, message *gen.MailboxMessage) error { - stateMachineValue := reflect.ValueOf(s) - callbackValue := reflect.ValueOf(handler) - fromValue := reflect.ValueOf(message.From) - stateValue := reflect.ValueOf(s.currentState) - dataValue := reflect.ValueOf(s.Data()) - msgValue := reflect.ValueOf(message.Message) - messageType := reflect.TypeOf(message).String() - - results := callbackValue.Call([]reflect.Value{fromValue, stateValue, dataValue, msgValue, stateMachineValue}) - - validateResultSize(results, 4, messageType) - if isError, err := resultIsError(results); isError == true { - return err - } - - return updateStateMachineWithResults(stateMachineValue, results) -} - -func (s *StateMachine[D]) lookupCallHandler(messageType string) (any, bool) { - if stateCallHandlers, exists := s.stateCallHandlers[s.currentState]; exists == true { - if callback, exists := stateCallHandlers[messageType]; exists == true { - return callback, true - } - } - return nil, false -} - -func (s *StateMachine[D]) invokeCallHandler(handler any, message *gen.MailboxMessage) (any, error) { - stateMachineValue := reflect.ValueOf(s) - callbackValue := reflect.ValueOf(handler) - fromValue := reflect.ValueOf(message.From) - stateValue := reflect.ValueOf(s.currentState) - dataValue := reflect.ValueOf(s.Data()) - msgValue := reflect.ValueOf(message.Message) - messageType := reflect.TypeOf(message).String() - - results := callbackValue.Call([]reflect.Value{fromValue, stateValue, dataValue, msgValue, stateMachineValue}) - - validateResultSize(results, 5, messageType) - if isError, err := resultIsError(results); isError == true { - return nil, err - } - err := updateStateMachineWithResults(stateMachineValue, results) - if err != nil { - return nil, err - } - result := results[2].Interface() - return result, nil -} - -func (s *StateMachine[D]) invokeEventHandler(handler any, message *gen.MessageEvent) error { - stateMachineValue := reflect.ValueOf(s) - callbackValue := reflect.ValueOf(handler) - stateValue := reflect.ValueOf(s.currentState) - dataValue := reflect.ValueOf(s.Data()) - msgValue := reflect.ValueOf(message.Message) - messageType := reflect.TypeOf(message).String() - - results := callbackValue.Call([]reflect.Value{stateValue, dataValue, msgValue, stateMachineValue}) - - validateResultSize(results, 4, messageType) - if isError, err := resultIsError(results); isError == true { - return err - } - updateStateMachineWithResults(stateMachineValue, results) - - return nil -} - -func validateResultSize(results []reflect.Value, expectedSize int, messageType string) { - if len(results) != expectedSize { - panic(fmt.Sprintf("StateMachine terminated. Panic reason: unexpected "+ - "error when invoking call handler for %s", messageType)) - } -} - -func resultIsError(results []reflect.Value) (bool, error) { - errIndex := len(results) - 1 - if !results[errIndex].IsNil() { - err := results[errIndex].Interface().(error) - return true, err - } - return false, nil -} - -func updateStateMachineWithResults(s reflect.Value, results []reflect.Value) error { - // Check if any actions were returned. MessageHandler and EventHandler have - // the result tuple (gen.Atom, D, []Action, error) with the actions at index - // 2. CallHandler has the result typle (gen.Atom, D, R, []Action, error) - // with the actions at index 3. - var actionsIndex int - hasResult := len(results) == 5 - if hasResult { - actionsIndex = 3 - } else { - actionsIndex = 2 - } - if !isSliceNilOrEmpty(results[actionsIndex]) { - processActionsMethod := s.MethodByName("ProcessActions") - if processActionsMethod.IsNil() { - } - processActionsMethod.Call([]reflect.Value{results[actionsIndex], results[0]}) - } - - // Update the data - setDataMethod := s.MethodByName("SetData") - setDataMethod.Call([]reflect.Value{results[1]}) - - // It is important that we set the state last as this can potentially trigger - // a state enter callback. By design state enter callbacks are triggered - // after setting up state timeouts as state timeouts are tied to te state - // they are defined for. A state enter callback could transition to another - // state which then will cancel the state timeout. - setCurrentStateMethod := s.MethodByName("SetCurrentState") - err := setCurrentStateMethod.Call([]reflect.Value{results[0]})[0] - if !err.IsNil() { - return err.Interface().(error) - } - - return nil -} - -func isSliceNilOrEmpty(resultValue reflect.Value) bool { - if resultValue.IsNil() { - return true - } - - if resultValue.Len() == 0 { - return true - } - - return false -} diff --git a/statemachine/statemachine_test.go b/statemachine/statemachine_test.go index 9170e19..c44dced 100644 --- a/statemachine/statemachine_test.go +++ b/statemachine/statemachine_test.go @@ -1,8 +1,10 @@ -package statemachine +package statemachine_test import ( "testing" + "ergo.services/actor/statemachine" + "ergo.services/actor/statemachine/action" "ergo.services/ergo/gen" "ergo.services/ergo/testing/unit" ) @@ -10,36 +12,41 @@ import ( type Data struct{} type BasicStatemachine struct { - StateMachine[Data] + statemachine.StateMachine[Data] } +const ( + stateA = gen.Atom("StateA") + stateB = gen.Atom("StateB") +) + type StateChange struct{} func factoryBasicStatemachine() gen.ProcessBehavior { return &BasicStatemachine{} } -func (b *BasicStatemachine) Init(args ...any) (StateMachineSpec[Data], error) { - spec := NewStateMachineSpec( - gen.Atom("StateA"), +func (b *BasicStatemachine) Init(args ...any) (statemachine.StateMachineSpec[Data], error) { + spec := statemachine.NewStateMachineSpec( + stateA, - WithStateEnterCallback(stateEnter), - WithStateMessageHandler(gen.Atom("StateA"), changeState), - WithStateCallHandler(gen.Atom("StateA"), changeStateSync), + statemachine.WithStateEnterCallback(stateEnter), + statemachine.WithStateMessageHandler(stateA, changeState), + statemachine.WithStateCallHandler(stateA, changeStateSync), ) return spec, nil } -func changeState(state gen.Atom, data Data, msg StateChange, proc gen.Process) (gen.Atom, Data, []Action, error) { - return gen.Atom("StateB"), data, nil, nil +func changeState(_ gen.PID, state gen.Atom, data Data, msg StateChange, proc gen.Process) (gen.Atom, Data, []action.Action, error) { + return stateB, data, nil, nil } -func changeStateSync(state gen.Atom, data Data, msg StateChange, proc gen.Process) (gen.Atom, Data, gen.Atom, []Action, error) { - return gen.Atom("StateB"), data, gen.Atom("StateB"), nil, nil +func changeStateSync(_ gen.PID, state gen.Atom, data Data, msg StateChange, proc gen.Process) (gen.Atom, Data, gen.Atom, []action.Action, error) { + return stateB, data, stateB, nil, nil } func stateEnter(oldState gen.Atom, newState gen.Atom, data Data, proc gen.Process) (gen.Atom, Data, error) { - if newState == gen.Atom("StateB") { + if newState == stateB { return newState, data, gen.TerminateReasonNormal } return newState, data, nil diff --git a/statemachine/types.go b/statemachine/types.go new file mode 100644 index 0000000..f1e48a1 --- /dev/null +++ b/statemachine/types.go @@ -0,0 +1,28 @@ +package statemachine + +import ( + "ergo.services/actor/statemachine/action" + "ergo.services/ergo/gen" +) + +type activeStateTimeout struct { + state gen.Atom + timeout action.StateTimeout + cancel func() + cancelled bool +} + +type activeGenericTimeout struct { + timeout action.GenericTimeout + cancel func() + cancelled bool +} + +type activeMessageTimeout struct { + timeout action.MessageTimeout + cancel func() + cancelled bool +} + +// message to self to start events monitoring +type startMonitoringEvents struct{}