actors

package
v0.8.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 27, 2023 License: MIT Imports: 40 Imported by: 4

Documentation

Index

Constants

View Source 
const (

	// RestartDirective defines the restart strategy when handling actors failure
	RestartDirective StrategyDirective = iota
	// StopDirective defines the stop strategy when handling actors failure
	StopDirective

	// DefaultPassivationTimeout defines the default passivation timeout
	DefaultPassivationTimeout = 2 * time.Minute
	// DefaultReplyTimeout defines the default send/reply timeout
	DefaultReplyTimeout = 100 * time.Millisecond
	// DefaultInitMaxRetries defines the default value for retrying actor initialization
	DefaultInitMaxRetries = 5
	// DefaultSupervisoryStrategy defines the default supervisory strategy
	DefaultSupervisoryStrategy = StopDirective
	// DefaultShutdownTimeout defines the default shutdown timeout
	DefaultShutdownTimeout = 2 * time.Second
)

Variables

View Source 
var (
	// ErrInvalidActorSystemName is returned when the actor system name is invalid
	ErrInvalidActorSystemName = errors.New("invalid ActorSystem name, must contain only word characters (i.e. [a-zA-Z0-9] plus non-leading '-' or '_')")
	// ErrDead means that the given actor is not alive
	ErrDead = errors.New("actor is not alive")
	// ErrUnhandled is used when an actor can handle a given message
	ErrUnhandled = errors.New("unhandled message")
	// ErrClusterDisabled is returned when cluster mode is not enabled while accessing cluster features
	ErrClusterDisabled = errors.New("cluster is not enabled")
	// ErrUndefinedActor is returned when an actor is defined
	ErrUndefinedActor = errors.New("actor is not defined")
	// ErrRequestTimeout is returned when sending an Ask message times out
	ErrRequestTimeout = errors.New("request timed out")
	// ErrRemotingDisabled is returned when remoting is not enabled
	ErrRemotingDisabled = errors.New("remoting is not enabled")
	// ErrInvalidNode is returned when the given actor system node is not the one requested
	ErrInvalidNode = connect.NewError(connect.CodeFailedPrecondition, errors.New("invalid actor system node"))
	// ErrAddressNotFound is returned when an actor address is not found
	ErrAddressNotFound = func(addr string) error {
		return connect.NewError(connect.CodeNotFound, fmt.Errorf("actor=%s not found", addr))
	}
	// ErrRemoteSendFailure is returned when remote message fails
	ErrRemoteSendFailure = func(err error) error { return connect.NewError(connect.CodeInternal, err) }
	// ErrInvalidActorInterface is returned when an actor does not implement the Actor interface
	ErrInvalidActorInterface = errors.New("failed to create instance. Reason: instance does not implement the Actor interface")
	// ErrInvalidInstance is returned when the creation of an actor instance fails
	ErrInvalidInstance = errors.New("failed to create instance. Reason: invalid instance")
	// ErrTypeNotFound is returned when an actor type is not found
	ErrTypeNotFound = func(typeName string) error { return fmt.Errorf("typeName=%s not found", typeName) }
	// ErrActorNotFound is returned when an actor is not found
	ErrActorNotFound = func(actorPath string) error { return fmt.Errorf("actor=%s not found", actorPath) }
	// ErrMethodCallNotAllowed is returned when rpc call is not allowed
	ErrMethodCallNotAllowed = errors.New("method call not allowed")
	// ErrInvalidRemoteMessage is returned when an invalid remote message is sent
	ErrInvalidRemoteMessage = func(err error) error { return errors.Wrap(err, "invalid remote message") }
	// ErrStashBufferNotSet when stashing is not set while requesting for messages to be stashed
	ErrStashBufferNotSet = errors.New("actor is not setup with a stash buffer")
	// ErrInitFailure is returned when the initialization of an actor fails.
	ErrInitFailure = func(err error) error { return errors.Wrap(err, "failed to initialize") }
	// ErrActorSystemNotStarted is returned when the actor is not started while accessing the features of the actor system
	ErrActorSystemNotStarted = errors.New("actor system has not started yet")
	// ErrLocalAddress is returned when a remote address is used instead of a local address
	ErrLocalAddress = errors.New("address is a local address")
	// ErrEmptyMailbox is returned when the mailbox is empty
	ErrEmptyMailbox = errors.New("mailbox is empty")
	// ErrFullMailbox is returned when the mailbox is full
	ErrFullMailbox = errors.New("mailbox is full")
)
View Source 
var (
	ErrNameRequired = errors.New("actor system is required")
)
View Source 
var RemoteNoSender = new(addresspb.Address)

RemoteNoSender means that there is no sender

Functions

func Ask added in v0.2.0 

func Ask(ctx context.Context, to PID, message proto.Message, timeout time.Duration) (response proto.Message, err error)

Ask sends a synchronous message to another actor and expect a response. This block until a response is received or timed out.

func RemoteAsk added in v0.2.0 

func RemoteAsk(ctx context.Context, to *addresspb.Address, message proto.Message) (response *anypb.Any, err error)

RemoteAsk sends a synchronous message to another actor remotely and expect a response.

func RemoteLookup 

func RemoteLookup(ctx context.Context, host string, port int, name string) (addr *addresspb.Address, err error)

RemoteLookup look for an actor address on a remote node.

func RemoteTell added in v0.2.0 

func RemoteTell(ctx context.Context, to *addresspb.Address, message proto.Message) error

RemoteTell sends a message to an actor remotely without expecting any reply

func Tell added in v0.2.0 

func Tell(ctx context.Context, to PID, message proto.Message) error

Tell sends an asynchronous message to an actor

Types

type Actor 

type Actor interface {
	// PreStart pre-starts the actor. This function can be used to set up some database connections
	// or some sort of initialization before the actor start processing messages
	// when the initialization failed the actor will not be started
	PreStart(ctx context.Context) error
	// Receive processes any message dropped into the actor mailbox.
	// The receiver of any message can either reply to the sender of the message with a new message or reply to the message synchronously
	// by config the reply of the message. The latter approach is often used when an external service is communicating to the actor.
	// One thing to know is that actor can communicate synchronously as well, just that will hinder the performance of the system.
	Receive(ctx ReceiveContext)
	// PostStop is executed when the actor is shutting down.
	// The execution happens when every message that have not been processed yet will be processed before the actor shutdowns
	PostStop(ctx context.Context) error
}

Actor represents the Actor interface This will be implemented by any user who wants to create an actor

type ActorSystem 

type ActorSystem interface {
	// Name returns the actor system name
	Name() string
	// Actors returns the list of Actors that are alive in the actor system
	Actors() []PID
	// Start starts the actor system
	Start(ctx context.Context) error
	// Stop stops the actor system
	Stop(ctx context.Context) error
	// Spawn creates an actor in the system and starts it
	Spawn(ctx context.Context, name string, actor Actor) (PID, error)
	// Kill stops a given actor in the system
	Kill(ctx context.Context, name string) error
	// ReSpawn recreates a given actor in the system
	ReSpawn(ctx context.Context, name string) (PID, error)
	// NumActors returns the total number of active actors in the system
	NumActors() uint64
	// LocalActor returns the reference of a local actor.
	// A local actor is an actor that reside on the same node where the given actor system is running
	LocalActor(ctx context.Context, actorName string) (PID, error)
	// RemoteActor returns the address of a remote actor when cluster is enabled
	// When the cluster mode is not enabled an actor not found error will be returned
	// One can always check whether cluster is enabled before calling this method or just use the ActorOf method.
	RemoteActor(ctx context.Context, actorName string) (addr *addresspb.Address, err error)
	// ActorOf returns an existing actor in the local system or in the cluster when clustering is enabled
	// When cluster mode is activated, the PID will be nil.
	// When remoting is enabled this method will return and error
	// An actor not found error is return when the actor is not found.
	ActorOf(ctx context.Context, actorName string) (addr *addresspb.Address, pid PID, err error)
	// InCluster states whether the actor system is running within a cluster of nodes
	InCluster() bool
	// GetPartition returns the partition where a given actor is located
	GetPartition(ctx context.Context, actorName string) uint64
	// SubscribeToEvent creates an event subscriber.
	SubscribeToEvent(ctx context.Context, event eventspb.Event) (eventstream.Subscriber, error)
	// UnsubscribeToEvent unsubscribes a subscriber.
	UnsubscribeToEvent(ctx context.Context, event eventspb.Event, subscriber eventstream.Subscriber) error
	// contains filtered or unexported methods
}

ActorSystem defines the contract of an actor system

func NewActorSystem 

func NewActorSystem(name string, opts ...Option) (ActorSystem, error)

NewActorSystem creates an instance of ActorSystem

type Address added in v0.2.0 

type Address struct {
	// contains filtered or unexported fields
}

Address represents the physical location under which an Actor can be reached. Examples are local addresses, identified by the ActorSystem’s name, and remote addresses, identified by protocol, host and port.

func NewAddress added in v0.2.0 

func NewAddress(system string, host string, port int) *Address

NewAddress creates an instance of Address

func (*Address) Host added in v0.2.0 

func (a *Address) Host() string

Host returns the host

func (*Address) HostPort added in v0.2.0 

func (a *Address) HostPort() string

HostPort returns the host and port in the following string format @host:port

func (*Address) IsLocal added in v0.2.0 

func (a *Address) IsLocal() bool

IsLocal helps set the actor address locally

func (*Address) IsRemote added in v0.2.0 

func (a *Address) IsRemote() bool

IsRemote states whether the actor address is in the remote environment This happens when remoting is enabled

func (*Address) Port added in v0.2.0 

func (a *Address) Port() int

Port returns the port number

func (*Address) Protocol added in v0.2.0 

func (a *Address) Protocol() string

Protocol returns the protocol

func (*Address) String added in v0.2.0 

func (a *Address) String() string

String returns the canonical String representation of this Address formatted as: `protocol://system@host:port`

func (*Address) System added in v0.2.0 

func (a *Address) System() string

System returns the actor system name

func (*Address) WithHost added in v0.2.0 

func (a *Address) WithHost(host string) (*Address, error)

WithHost sets the hosts of a given Address and returns a new instance of the address

func (*Address) WithPort added in v0.2.0 

func (a *Address) WithPort(port int) (*Address, error)

WithPort sets the port of a given Address and returns a new instance of the address

func (*Address) WithSystem added in v0.2.0 

func (a *Address) WithSystem(system string) *Address

WithSystem sets the actor system of a given Address and returns a new instance of the address

type Behavior 

type Behavior func(ctx ReceiveContext)

Behavior defines an actor behavior

type Mailbox added in v0.6.0 

type Mailbox interface {
	// Push pushes a message into the mailbox. This returns an error
	// when the box is full
	Push(msg ReceiveContext) error
	// Pop fetches a message from the mailbox
	Pop() (msg ReceiveContext, err error)
	// IsEmpty returns true when the mailbox is empty
	IsEmpty() bool
	// IsFull returns true when the mailbox is full
	IsFull() bool
	// Size returns the size of the buffer atomically
	Size() uint64
	// Reset resets the mailbox
	Reset()
	// Clone clones the current mailbox and returns a new Mailbox with reset settings
	Clone() Mailbox
	// Capacity returns the mailbox capacity atomically
	Capacity() uint64
}

Mailbox defines the actor mailbox. Any implementation should be a thread-safe FIFO

type Option 

type Option interface {
	// Apply sets the Option value of a config.
	Apply(sys *actorSystem)
}

Option is the interface that applies a configuration option.

func WithActorInitMaxRetries 

func WithActorInitMaxRetries(max int) Option

WithActorInitMaxRetries sets the number of times to retry an actor init process

func WithClustering added in v0.2.0 

func WithClustering(serviceDiscovery *discovery.ServiceDiscovery, partitionCount uint64) Option

WithClustering enables clustering on the actor system. This enables remoting on the actor system as well and set the remotingHost to the cluster node host when the cluster is fully enabled.

func WithExpireActorAfter 

func WithExpireActorAfter(duration time.Duration) Option

WithExpireActorAfter sets the actor expiry duration. After such duration an idle actor will be expired and removed from the actor system

func WithLogger 

func WithLogger(logger log.Logger) Option

WithLogger sets the actor system custom log

func WithMailbox added in v0.6.0 

func WithMailbox(mailbox Mailbox) Option

WithMailbox sets the custom mailbox used by the actors in the actor system

func WithMailboxSize added in v0.6.0 

func WithMailboxSize(size uint64) Option

WithMailboxSize sets the actors mailbox size

func WithPassivationDisabled 

func WithPassivationDisabled() Option

WithPassivationDisabled disable the passivation mode

func WithRemoting 

func WithRemoting(host string, port int32) Option

WithRemoting enables remoting on the actor system

func WithReplyTimeout 

func WithReplyTimeout(timeout time.Duration) Option

WithReplyTimeout sets how long in seconds an actor should reply a command in a receive-reply pattern

func WithShutdownTimeout added in v0.4.0 

func WithShutdownTimeout(timeout time.Duration) Option

WithShutdownTimeout sets the shutdown timeout

func WithStash added in v0.7.0 

func WithStash(capacity uint64) Option

WithStash sets the stash buffer size

func WithSupervisorStrategy 

func WithSupervisorStrategy(strategy StrategyDirective) Option

WithSupervisorStrategy sets the supervisor strategy

func WithTelemetry 

func WithTelemetry(telemetry *telemetry.Telemetry) Option

WithTelemetry sets the custom telemetry

type OptionFunc 

type OptionFunc func(*actorSystem)

OptionFunc implements the Option interface.

func (OptionFunc) Apply 

func (f OptionFunc) Apply(c *actorSystem)

type PID 

type PID interface {
	// Shutdown gracefully shuts down the given actor
	// All current messages in the mailbox will be processed before the actor shutdown after a period of time
	// that can be configured. All child actors will be gracefully shutdown.
	Shutdown(ctx context.Context) error
	// IsRunning returns true when the actor is running ready to process public and false
	// when the actor is stopped or not started at all
	IsRunning() bool
	// ReceivedCount returns the total number of public processed by the actor
	// at a given point in time while the actor heart is still beating
	ReceivedCount(ctx context.Context) uint64
	// ErrorsCount returns the total number of panic attacks that occur while the actor is processing public
	// at a given point in time while the actor heart is still beating
	ErrorsCount(ctx context.Context) uint64
	// SpawnChild creates a child actor
	// When the given child actor already exists its PID will only be returned
	SpawnChild(ctx context.Context, name string, actor Actor) (PID, error)
	// Restart restarts the actor
	Restart(ctx context.Context) error
	// Watch an actor
	Watch(pid PID)
	// UnWatch stops watching a given actor
	UnWatch(pid PID)
	// ActorSystem returns the underlying actor system
	ActorSystem() ActorSystem
	// ActorPath returns the path of the actor
	ActorPath() *Path
	// ActorHandle returns the underlying actor
	ActorHandle() Actor
	// Tell sends an asynchronous message to another PID
	Tell(ctx context.Context, to PID, message proto.Message) error
	// Ask sends a synchronous message to another actor and expect a response.
	Ask(ctx context.Context, to PID, message proto.Message) (response proto.Message, err error)
	// RemoteTell sends a message to an actor remotely without expecting any reply
	RemoteTell(ctx context.Context, to *addresspb.Address, message proto.Message) error
	// RemoteAsk is used to send a message to an actor remotely and expect a response
	// immediately. With this type of message the receiver cannot communicate back to Sender
	// except reply the message with a response. This one-way communication.
	RemoteAsk(ctx context.Context, to *addresspb.Address, message proto.Message) (response *anypb.Any, err error)
	// RemoteLookup look for an actor address on a remote node. If the actorSystem is nil then the lookup will be done
	// using the same actor system as the PID actor system
	RemoteLookup(ctx context.Context, host string, port int, name string) (addr *addresspb.Address, err error)
	// StartCount returns the number of times the actor has started
	StartCount(ctx context.Context) uint64
	// MailboxSize returns the mailbox size a given time
	MailboxSize(ctx context.Context) uint64
	// Children returns the list of all the children of the given actor that are still alive
	// or an empty list.
	Children(ctx context.Context) []PID
	// Child returns the named child actor if it is alive
	Child(ctx context.Context, name string) (PID, error)
	// Stop forces the child Actor under the given name to terminate after it finishes processing its current message.
	// Nothing happens if child is already stopped.
	Stop(ctx context.Context, pid PID) error
	// StashSize returns the stash buffer size
	StashSize(ctx context.Context) uint64
	// contains filtered or unexported methods
}

PID defines the various actions one can perform on a given actor 

var NoSender PID

NoSender means that there is no sender

type Path 

type Path struct {
	// contains filtered or unexported fields
}

Path is a unique path to an actor

func NewPath 

func NewPath(name string, address *Address) *Path

NewPath creates an immutable Path

func (*Path) Address added in v0.2.0 

func (p *Path) Address() *Address

Address returns the address of the path

func (*Path) ID 

func (p *Path) ID() uuid.UUID

ID returns the internal unique id of the actor that this path refer to.

func (*Path) Name 

func (p *Path) Name() string

Name returns the name of the actor that this path refers to.

func (*Path) Parent 

func (p *Path) Parent() *Path

Parent returns the parent path

func (*Path) RemoteAddress 

func (p *Path) RemoteAddress() *addresspb.Address

RemoteAddress returns the remote from path

func (*Path) String 

func (p *Path) String() string

String returns the string representation of an actorPath

func (*Path) WithParent 

func (p *Path) WithParent(parent *Path) *Path

WithParent sets the parent actor path and returns a new path This function is immutable

type ReceiveContext 

type ReceiveContext interface {
	// Context returns the context attached to the message
	Context() context.Context
	// Sender of the message. In the case of remote message this will be set to NoSender
	Sender() PID
	// Self represents the actor receiving the message.
	Self() PID
	// Message is the actual message sent
	Message() proto.Message
	// Response sets the message response
	// Use this method within the Actor.Receive method of the actor to sets a reply
	// This can only be used when we are request-response pattern. When it is an async communication
	// this operation will amount to nothing.
	Response(resp proto.Message)
	// RemoteSender defines the remote sender of the message if it is a remote message
	// This is set to RemoteNoSender when the message is not a remote message
	RemoteSender() *addresspb.Address
	// Become switch the current behavior of the actor to a new behavior
	// The current message in process during the transition will still be processed with the current
	// behavior before the transition. However, subsequent messages will be processed with the new behavior.
	// One needs to call UnBecome to reset the actor behavior to the default one which is the Actor.Receive method
	// which is the default behavior.
	Become(behavior Behavior)
	// UnBecome reset the actor behavior to the default one which is the
	// Actor.Receive method
	UnBecome()
	// BecomeStacked sets a new behavior to the actor to the top of the behavior stack, while maintaining the previous ones.
	// The current message in process during the transition will still be processed with the current
	// behavior before the transition. However, subsequent messages will be processed with the new behavior.
	// One needs to call UnBecomeStacked to go the previous the actor's behavior.
	// which is the default behavior.
	BecomeStacked(behavior Behavior)
	// UnBecomeStacked sets the actor behavior to the previous behavior before BecomeStacked was called
	UnBecomeStacked()
	// Stash adds the current message to the stash buffer
	Stash()
	// Unstash unstashes the oldest message in the stash and prepends to the mailbox
	Unstash()
	// UnstashAll unstashes all messages from the stash buffer  and prepends in the mailbox
	// it keeps the messages in the same order as received, unstashing older messages before newer
	UnstashAll()
	// Tell sends an asynchronous message to another PID
	Tell(to PID, message proto.Message)
	// Ask sends a synchronous message to another actor and expect a response. This method is good when interacting with a child actor.
	// Ask has a timeout which can cause the sender to panic. When ask times out, the receiving actor does not know and may still process the message.
	// It is recommended to set a good timeout to quickly receive response and try to avoid false positives
	Ask(to PID, message proto.Message) (response proto.Message)
	// Forward method works similarly to the Tell() method except that the sender of a forwarded message is kept as the original sender.
	// As a result, the actor receiving the forwarded messages knows who the actual sender of the message is.
	// The message that is forwarded is the current message received by the received context.
	// This operation does nothing when the receiving actor is not running
	Forward(to PID)
	// RemoteTell sends a message to an actor remotely without expecting any reply
	RemoteTell(to *addresspb.Address, message proto.Message)
	// RemoteAsk is used to send a message to an actor remotely and expect a response
	// immediately. This executed within an actor can hinder performance because this is a blocking call.
	RemoteAsk(to *addresspb.Address, message proto.Message) (response *anypb.Any)
	// RemoteLookup look for an actor address on a remote node. If the actorSystem is nil then the lookup will be done
	// using the same actor system as the PID actor system
	RemoteLookup(host string, port int, name string) (addr *addresspb.Address)
	// Shutdown gracefully shuts down the given actor
	// All current messages in the mailbox will be processed before the actor shutdown after a period of time
	// that can be configured. All child actors will be gracefully shutdown.
	Shutdown()
	// Spawn creates a child actor or panic
	Spawn(name string, actor Actor) PID
	// Children returns the list of all the children of the given actor
	Children() []PID
	// Child returns the named child actor if it is alive
	Child(name string) PID
	// Stop forces the child Actor under the given name to terminate after it finishes processing its current message.
	// Nothing happens if child is already stopped.
	Stop(child PID)
	// Unhandled is used to handle unhandled messages instead of throwing error
	// This will push the given message into the deadletter queue
	Unhandled()
}

ReceiveContext is the context that is used by the actor to receive messages

type Reflection 

type Reflection interface {
	// ActorOf creates a new instance of Actor from its concrete type
	ActorOf(rtype reflect.Type) (actor Actor, err error)
	// ActorFrom creates a new instance of Actor from its FQN
	ActorFrom(name string) (actor Actor, err error)
}

Reflection helps create an instance dynamically

func NewReflection 

func NewReflection(loader TypesLoader) Reflection

NewReflection creates an instance of Reflection

type StrategyDirective 

type StrategyDirective int

StrategyDirective represents the supervisor strategy directive

type TypesLoader 

type TypesLoader interface {
	// Register an object with its fully qualified name
	Register(name string, v any)
	// Type returns the type of object,
	Type(v any) (reflect.Type, bool)
	// TypeByName returns the type of object given its name
	TypeByName(name string) (reflect.Type, bool)
}

TypesLoader represents reflection typesLoader for dynamic loading and creation of actors at run-time

func NewTypesLoader 

func NewTypesLoader() TypesLoader

NewTypesLoader creates an instance of TypesLoader

type Unit 

type Unit struct{}

Unit type

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.