actors

Dec 19, 2024

Changes in this version

+ var ErrActorAlreadyExists = func(actorName string) error

type ActorSystem

+ Uptime func() int64

type BoundedMailbox

+ func (mailbox *BoundedMailbox) Dispose()

type Mailbox

+ Dispose func()

type PID

+ func (pid *PID) Uptime() int64

type UnboundedMailbox

+ func (m *UnboundedMailbox) Dispose()

type UnboundedPriorityMailBox

+ func (q *UnboundedPriorityMailBox) Dispose()

Dec 12, 2024

Changes in this version

+ var DefaultSupervisorStrategies = []*SupervisorStrategy

+ var ErrPriorityMessageRequired = errors.New("priority message type is required")

type PID

+ func (pid *PID) IsSuspended() bool

+ type PanicError struct

+ func NewPanicError(err error) PanicError

+ func (e PanicError) Error() string

+ type PriorityFunc func(msg1, msg2 proto.Message) bool

type SpawnOption

+ func WithSupervisorStrategies(supervisorStrategies ...*SupervisorStrategy) SpawnOption

+ type SupervisorStrategy struct

+ func NewSupervisorStrategy(err error, directive SupervisorDirective) *SupervisorStrategy

+ func (s *SupervisorStrategy) Directive() SupervisorDirective

+ func (s *SupervisorStrategy) Error() error

+ type UnboundedPriorityMailBox struct

+ func NewUnboundedPriorityMailBox(priorityFunc PriorityFunc) *UnboundedPriorityMailBox

+ func (q *UnboundedPriorityMailBox) Dequeue() (msg *ReceiveContext)

+ func (q *UnboundedPriorityMailBox) Enqueue(msg *ReceiveContext) error

+ func (q *UnboundedPriorityMailBox) IsEmpty() bool

+ func (q *UnboundedPriorityMailBox) Len() int64

Dec 1, 2024

Changes in this version

type ActorSystem

+ Host func() string

+ Port func() int32

Nov 28, 2024

Changes in this version

type ReceiveContext

+ func (rctx *ReceiveContext) ForwardTo(actorName string)

+ func (rctx *ReceiveContext) RemoteForward(to *address.Address)

Nov 24, 2024

Changes in this version

type PID

+ func (pid *PID) Parent() *PID

type ReceiveContext

+ func (rctx *ReceiveContext) ActorSystem() ActorSystem

+ func (rctx *ReceiveContext) Logger() log.Logger

+ func (rctx *ReceiveContext) Parent() *PID

+ func (rctx *ReceiveContext) UnWatch(cid *PID)

+ func (rctx *ReceiveContext) Watch(cid *PID)

Nov 21, 2024

Changes in this version

type SpawnOption

+ func WithSupervisor(directive SupervisorDirective) SpawnOption

Nov 11, 2024

Changes in this version

+ var ErrInvalidMessage = func(err error) error

+ var ErrInvalidTimeout = errors.New("invalid timeout")

+ var ErrSchedulerNotStarted = errors.New("scheduler has not started")

type ActorSystem

+ RemoteSchedule func(ctx context.Context, message proto.Message, address *address.Address, ...) error

+ Schedule func(ctx context.Context, message proto.Message, pid *PID, interval time.Duration) error

+ type Remoting struct

+ func NewRemoting() *Remoting

+ func (r *Remoting) Client(host string, port int) internalpbconnect.RemotingServiceClient

+ func (r *Remoting) Close()

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

+ func (r *Remoting) RemoteBatchAsk(ctx context.Context, to *address.Address, messages []proto.Message, ...) (responses []*anypb.Any, err error)

+ func (r *Remoting) RemoteBatchTell(ctx context.Context, to *address.Address, messages []proto.Message) error

+ func (r *Remoting) RemoteLookup(ctx context.Context, host string, port int, name string) (addr *address.Address, err error)

+ func (r *Remoting) RemoteReSpawn(ctx context.Context, host string, port int, name string) error

+ func (r *Remoting) RemoteSpawn(ctx context.Context, host string, port int, name, actorType string) error

+ func (r *Remoting) RemoteStop(ctx context.Context, host string, port int, name string) error

+ func (r *Remoting) RemoteTell(ctx context.Context, to *address.Address, message proto.Message) error

Sep 28, 2024

Changes in this version

+ type BoundedMailbox struct

+ func NewBoundedMailbox(cap int) *BoundedMailbox

+ func (mailbox *BoundedMailbox) Dequeue() (msg *ReceiveContext)

+ func (mailbox *BoundedMailbox) Enqueue(msg *ReceiveContext) error

+ func (mailbox *BoundedMailbox) IsEmpty() bool

+ func (mailbox *BoundedMailbox) Len() int64

type FuncOption

+ func WithFuncMailbox(mailbox Mailbox) FuncOption

type Mailbox

+ Dequeue func() (msg *ReceiveContext)

+ Enqueue func(msg *ReceiveContext) error

+ Len func() int64

type PID

+ func (pid *PID) ChildrenCount() int

+ func (pid *PID) LatestProcessedDuration() time.Duration

+ func (pid *PID) ProcessedCount() int

+ func (pid *PID) RestartCount() int

+ type SpawnOption interface

+ Apply func(config *spawnConfig)

+ type UnboundedMailbox struct

+ func NewUnboundedMailbox() *UnboundedMailbox

+ func (m *UnboundedMailbox) Dequeue() *ReceiveContext

+ func (m *UnboundedMailbox) Enqueue(value *ReceiveContext) error

+ func (m *UnboundedMailbox) IsEmpty() bool

+ func (m *UnboundedMailbox) Len() int64

Sep 24, 2024

Changes in this version

+ const DefaultJanitorInterval

type ClusterConfig

+ func (x *ClusterConfig) DiscoveryPort() int

+ func (x *ClusterConfig) WithDiscoveryPort(port int) *ClusterConfig

type Option

+ func WithJanitorInterval(interval time.Duration) Option

type PID

+ func (pid *PID) Actor() Actor

+ func (pid *PID) Address() *address.Address

+ func (pid *PID) Logger() log.Logger

+ func (pid *PID) SendAsync(ctx context.Context, actorName string, message proto.Message) error

+ func (pid *PID) SendSync(ctx context.Context, actorName string, message proto.Message) (response proto.Message, err error)

type ReceiveContext

+ func (c *ReceiveContext) SendAsync(actorName string, message proto.Message)

+ func (c *ReceiveContext) SendSync(actorName string, message proto.Message) (response proto.Message)

Aug 25, 2024

Changes in this version

type ReceiveContext

+ func (c *ReceiveContext) WithRemoteSender(remoteSender *goaktpb.Address) *ReceiveContext

Jul 26, 2024

Changes in this version

+ var ErrInvalidPath = func(err error) error

type ActorSystem

+ Logger func() log.Logger

+ SpawnNamedFromFunc func(ctx context.Context, name string, receiveFunc ReceiveFunc, opts ...FuncOption) (PID, error)

+ SpawnRouter func(ctx context.Context, poolSize int, routeesKind Actor, opts ...RouterOption) (PID, error)

type Option

+ func WithSupervisorDirective(directive SupervisorDirective) Option

type PID

+ Parents func() []PID

type Path

+ func (p *Path) Validate() error

type RestartDirective

+ func NewRestartDirective() *RestartDirective

+ func (x *RestartDirective) MaxNumRetries() uint32

+ func (x *RestartDirective) Timeout() time.Duration

+ func (x *RestartDirective) WithLimit(maxNumRetries uint32, timeout time.Duration)

+ type ResumeDirective struct

+ func NewResumeDirective() *ResumeDirective

+ type RouterOption interface

+ Apply func(cl *router)

+ func WithRoutingStrategy(strategy RoutingStrategy) RouterOption

+ type RouterOptionFunc func(router *router)

+ func (f RouterOptionFunc) Apply(c *router)

+ type RoutingStrategy int

+ const FanOutRouting

+ const RandomRouting

+ const RoundRobinRouting

type StopDirective

+ func NewStopDirective() *StopDirective

+ type SupervisorDirective interface

Jun 29, 2024

Changes in this version

+ const DefaultAskTimeout

+ const DefaultGCInterval

type Option

+ func WithGCInterval(interval time.Duration) Option

Jun 22, 2024

Changes in this version

+ const DefaultPeerStateLoopInterval

+ var ErrInvalidHost = errors.New("invalid host")

type Option

+ func WithPeerStateLoopInterval(interval time.Duration) Option

Jun 16, 2024

Changes in this version

+ const DefaultMailboxSize

+ const DefaultStashCapacity

+ var ErrUndefinedTask = errors.New("task is not defined")

type PID

+ PipeTo func(ctx context.Context, to PID, task future.Task) error

type ReceiveContext

+ PipeTo func(to PID, task future.Task)

Jun 4, 2024

Changes in this version

+ var ErrKindsRequired = errors.New("actor kinds are required")

+ var ErrPeerNotFound = errors.New("peer is not found")

+ type ClusterConfig struct

+ func NewClusterConfig() *ClusterConfig

+ func (x *ClusterConfig) Discovery() discovery.Provider

+ func (x *ClusterConfig) GossipPort() int

+ func (x *ClusterConfig) Kinds() []Actor

+ func (x *ClusterConfig) MinimumPeersQuorum() uint32

+ func (x *ClusterConfig) PartitionCount() uint64

+ func (x *ClusterConfig) PeersPort() int

+ func (x *ClusterConfig) ReplicaCount() uint32

+ func (x *ClusterConfig) Validate() error

+ func (x *ClusterConfig) WithDiscovery(discovery discovery.Provider) *ClusterConfig

+ func (x *ClusterConfig) WithGossipPort(port int) *ClusterConfig

+ func (x *ClusterConfig) WithKinds(kinds ...Actor) *ClusterConfig

+ func (x *ClusterConfig) WithMinimumPeersQuorum(minimumQuorum uint32) *ClusterConfig

+ func (x *ClusterConfig) WithPartitionCount(count uint64) *ClusterConfig

+ func (x *ClusterConfig) WithPeersPort(peersPort int) *ClusterConfig

+ func (x *ClusterConfig) WithReplicaCount(count uint32) *ClusterConfig

type Option

+ func WithCluster(config *ClusterConfig) Option

type PID

+ Name func() string

May 25, 2024

Changes in this version

+ const DefaultInitMaxRetries

+ const DefaultInitTimeout

+ const DefaultPassivationTimeout

+ const DefaultReplyTimeout

+ const DefaultShutdownTimeout

+ const DefaultSupervisoryStrategy

+ const RestartDirective

+ const StopDirective

+ var ErrActorNotFound = func(actorPath string) error

+ var ErrActorSystemNotStarted = errors.New("actor system has not started yet")

+ var ErrAddressNotFound = func(addr string) error

+ var ErrClusterDisabled = errors.New("cluster is not enabled")

+ var ErrDead = errors.New("actor is not alive")

+ var ErrEmptyMailbox = errors.New("mailbox is empty")

+ var ErrFullMailbox = errors.New("mailbox is full")

+ var ErrInitFailure = func(err error) error

+ var ErrInstanceNotAnActor = errors.New(...)

+ var ErrInvalidActorSystemName = errors.New(...)

+ var ErrInvalidInstance = errors.New("failed to create instance. Reason: invalid instance")

+ var ErrInvalidRemoteMessage = func(err error) error

+ var ErrLocalAddress = errors.New("address is a local address")

+ var ErrMethodCallNotAllowed = errors.New("method call not allowed")

+ var ErrNameRequired = errors.New("actor system is required")

+ var ErrRemoteSendFailure = func(err error) error

+ var ErrRemotingDisabled = errors.New("remoting is not enabled")

+ var ErrRequestTimeout = errors.New("request timed out")

+ var ErrStashBufferNotSet = errors.New("actor is not setup with a stash buffer")

+ var ErrTypeNotRegistered = errors.New("actor type is not registered")

+ var ErrUndefinedActor = errors.New("actor is not defined")

+ var ErrUnhandled = errors.New("unhandled message")

+ var RemoteNoSender = new(goaktpb.Address)

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

+ func BatchAsk(ctx context.Context, to PID, timeout time.Duration, messages ...proto.Message) (responses chan proto.Message, err error)

+ func BatchTell(ctx context.Context, to PID, messages ...proto.Message) error

+ func IsEOF(err error) bool

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

+ func RemoteBatchAsk(ctx context.Context, to *goaktpb.Address, messages ...proto.Message) (responses []*anypb.Any, err error)

+ func RemoteBatchTell(ctx context.Context, to *goaktpb.Address, messages ...proto.Message) error

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

+ func RemoteReSpawn(ctx context.Context, host string, port int, name string) error

+ func RemoteSpawn(ctx context.Context, host string, port int, name, actorType string) error

+ func RemoteStop(ctx context.Context, host string, port int, name string) error

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

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

+ type Actor interface

+ PostStop func(ctx context.Context) error

+ PreStart func(ctx context.Context) error

+ Receive func(ctx ReceiveContext)

+ type ActorSystem interface

+ ActorOf func(ctx context.Context, actorName string) (addr *goaktpb.Address, pid PID, err error)

+ Actors func() []PID

+ Deregister func(ctx context.Context, actor Actor) error

+ GetPartition func(actorName string) uint64

+ InCluster func() bool

+ Kill func(ctx context.Context, name string) error

+ LocalActor func(actorName string) (PID, error)

+ Name func() string

+ NumActors func() uint64

+ PeerAddress func() string

+ ReSpawn func(ctx context.Context, name string) (PID, error)

+ Register func(ctx context.Context, actor Actor) error

+ RemoteActor func(ctx context.Context, actorName string) (addr *goaktpb.Address, err error)

+ RemoteScheduleOnce func(ctx context.Context, message proto.Message, address *goaktpb.Address, ...) error

+ RemoteScheduleWithCron func(ctx context.Context, message proto.Message, address *goaktpb.Address, ...) error

+ ScheduleOnce func(ctx context.Context, message proto.Message, pid PID, interval time.Duration) error

+ ScheduleWithCron func(ctx context.Context, message proto.Message, pid PID, cronExpression string) error

+ Spawn func(ctx context.Context, name string, actor Actor) (PID, error)

+ SpawnFromFunc func(ctx context.Context, receiveFunc ReceiveFunc, opts ...FuncOption) (PID, error)

+ Start func(ctx context.Context) error

+ Stop func(ctx context.Context) error

+ Subscribe func() (eventstream.Subscriber, error)

+ Unsubscribe func(subscriber eventstream.Subscriber) error

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

+ type Address struct

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

+ func (a *Address) Host() string

+ func (a *Address) HostPort() string

+ func (a *Address) IsLocal() bool

+ func (a *Address) IsRemote() bool

+ func (a *Address) Port() int

+ func (a *Address) Protocol() string

+ func (a *Address) String() string

+ func (a *Address) System() string

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

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

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

+ type Behavior func(ctx ReceiveContext)

+ type FuncOption interface

+ Apply func(actor *fnActor)

+ func WithPostStop(fn PostStopFunc) FuncOption

+ func WithPreStart(fn PreStartFunc) FuncOption

+ type Mailbox interface

+ Capacity func() uint64

+ Clone func() Mailbox

+ IsEmpty func() bool

+ IsFull func() bool

+ Iterator func() <-chan ReceiveContext

+ Pop func() (msg ReceiveContext, err error)

+ Push func(msg ReceiveContext) error

+ Reset func()

+ Size func() uint64

+ type Option interface

+ Apply func(sys *actorSystem)

+ func WithActorInitMaxRetries(max int) Option

+ func WithActorInitTimeout(timeout time.Duration) Option

+ func WithClustering(provider discovery.Provider, partitionCount uint64, minimumPeersQuorum uint16, ...) Option

+ func WithExpireActorAfter(duration time.Duration) Option

+ func WithLogger(logger log.Logger) Option

+ func WithMailbox(mailbox Mailbox) Option

+ func WithMailboxSize(size uint64) Option

+ func WithMetric() Option

+ func WithPartitionHasher(hasher hash.Hasher) Option

+ func WithPassivationDisabled() Option

+ func WithRemoting(host string, port int32) Option

+ func WithReplyTimeout(timeout time.Duration) Option

+ func WithShutdownTimeout(timeout time.Duration) Option

+ func WithStash(capacity uint64) Option

+ func WithSupervisorStrategy(strategy StrategyDirective) Option

+ func WithTelemetry(telemetry *telemetry.Telemetry) Option

+ func WithTracing() Option

+ type OptionFunc func(*actorSystem)

+ func (f OptionFunc) Apply(c *actorSystem)

+ type PID interface

+ ActorHandle func() Actor

+ ActorPath func() *Path

+ ActorSystem func() ActorSystem

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

+ BatchAsk func(ctx context.Context, to PID, messages ...proto.Message) (responses chan proto.Message, err error)

+ BatchTell func(ctx context.Context, to PID, messages ...proto.Message) error

+ Child func(name string) (PID, error)

+ Children func() []PID

+ Equals func(to PID) bool

+ ID func() string

+ IsRunning func() bool

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

+ RemoteBatchAsk func(ctx context.Context, to *goaktpb.Address, messages ...proto.Message) (responses []*anypb.Any, err error)

+ RemoteBatchTell func(ctx context.Context, to *goaktpb.Address, messages ...proto.Message) error

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

+ RemoteReSpawn func(ctx context.Context, host string, port int, name string) error

+ RemoteSpawn func(ctx context.Context, host string, port int, name, actorType string) error

+ RemoteStop func(ctx context.Context, host string, port int, name string) error

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

+ Restart func(ctx context.Context) error

+ Shutdown func(ctx context.Context) error

+ SpawnChild func(ctx context.Context, name string, actor Actor) (PID, error)

+ StashSize func() uint64

+ Stop func(ctx context.Context, pid PID) error

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

+ UnWatch func(pid PID)

+ Watch func(pid PID)

+ var NoSender PID

+ type Path struct

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

+ func (p *Path) Address() *Address

+ func (p *Path) Equals(x *Path) bool

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

+ func (p *Path) Name() string

+ func (p *Path) Parent() *Path

+ func (p *Path) RemoteAddress() *goaktpb.Address

+ func (p *Path) String() string

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

+ type PostStopFunc = func(ctx context.Context) error

+ type PreStartFunc = func(ctx context.Context) error

+ type ReceiveContext interface

+ Ask func(to PID, message proto.Message) (response proto.Message)

+ BatchAsk func(to PID, messages ...proto.Message) (responses chan proto.Message)

+ BatchTell func(to PID, messages ...proto.Message)

+ Become func(behavior Behavior)

+ BecomeStacked func(behavior Behavior)

+ Child func(name string) PID

+ Children func() []PID

+ Context func() context.Context

+ Err func(err error)

+ Forward func(to PID)

+ Message func() proto.Message

+ RemoteAsk func(to *goaktpb.Address, message proto.Message) (response *anypb.Any)

+ RemoteBatchAsk func(to *goaktpb.Address, messages ...proto.Message) (responses []*anypb.Any)

+ RemoteBatchTell func(to *goaktpb.Address, messages ...proto.Message)

+ RemoteLookup func(host string, port int, name string) (addr *goaktpb.Address)

+ RemoteReSpawn func(host string, port int, name string)

+ RemoteSender func() *goaktpb.Address

+ RemoteTell func(to *goaktpb.Address, message proto.Message)

+ Response func(resp proto.Message)

+ Self func() PID

+ Sender func() PID

+ Shutdown func()

+ Spawn func(name string, actor Actor) PID

+ Stash func()

+ Stop func(child PID)

+ Tell func(to PID, message proto.Message)

+ UnBecome func()

+ UnBecomeStacked func()

+ Unhandled func()

+ Unstash func()

+ UnstashAll func()

+ type ReceiveFunc = func(ctx context.Context, message proto.Message) error

+ type StrategyDirective int