pkgsite style docs for statemachine actor

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()
+}

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() {}

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")
+		}
+	}
+}

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
+}

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,
+)

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]