actor

package
v0.0.0-...-f6ceceb Latest Latest
Warning

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

 Go to latest Published: Jun 9, 2020 License: Apache-2.0 Imports: 19 Imported by: 0

Documentation

Overview

Package actor declares the types used to represent actors in the Actor Model.

The actors model provide a high level abstraction for writing concurrent and distributed systems. This approach simplifies the burden imposed on engineers, such as explicit locks and concurrent access to shared state, as actors receive messages synchronously.

The following quote from Wikipedia distills the definition of an actor down to its essence 

In response to a message that it receives, an actor can: make local decisions, create more actors,
send more messages, and determine how to respond to the next message received.

Creating Actors

Props provide the building blocks for declaring how actors should be created. The following example defines an actor using a function literal to process messages: 

var props Props = actor.PropsFromFunc(func(c Context) {
	// process messages
})

Alternatively, a type which conforms to the Actor interface, by defining a single Receive method, can be used. 

type MyActor struct {}

func (a *MyActor) Receive(c Context) {
	// process messages
}

var props Props = actor.PropsFromProducer(func() Actor { return &MyActor{} })

Spawn and SpawnNamed use the given props to create a running instances of an actor. Once spawned, the actor is ready to process incoming messages. To spawn an actor with a unique name, use 

pid := context.Spawn(props)

The result of calling Spawn is a unique PID or process identifier.

Each time an actor is spawned, a new mailbox is created and associated with the PID. Messages are sent to the mailbox and then forwarded to the actor to process.

Processing Messages

An actor processes messages via its Receive handler. The signature of this function is: 

Receive(c actor.Context)

The actor system guarantees that this method is called synchronously, therefore there is no requirement to protect shared state inside calls to this function.

Communicating With Actors

A PID is the primary interface for sending messages to actors. The PID.Tell method is used to send an asynchronous message to the actor associated with the PID: 

pid.Tell("Hello World")

Depending on the requirements, communication between actors can take place synchronously or asynchronously. Regardless of the circumstances, actors always communicate via a PID.

When sending a message using PID.Request or PID.RequestFuture, the actor which receives the message will respond using the Context.Sender method, which returns the PID of of the sender.

For synchronous communication, an actor will use a Future and wait for the result before continuing. To send a message to an actor and wait for a response, use the RequestFuture method, which returns a Future: 

f := actor.RequestFuture(pid,"Hello", 50 * time.Millisecond)
res, err := f.Result() // waits for pid to reply
Example

Demonstrates how to create an actor using a function literal and how to send a message asynchronously 

package main

import (
	"fmt"
	"time"

	"github.com/ChaokunChang/protoactor-go/actor"
)

func main() {
	var context = actor.EmptyRootContext
	var props = actor.PropsFromFunc(func(c actor.Context) {
		if msg, ok := c.Message().(string); ok {
			fmt.Println(msg) // outputs "Hello World"
		}
	})

	pid := context.Spawn(props)

	context.Send(pid, "Hello World")
	time.Sleep(time.Millisecond * 100)
	context.StopFuture(pid).Wait() // wait for the actor to stop

}

Output:

Hello World
Example (Synchronous)

Demonstrates how to send a message from one actor to another and for the caller to wait for a response before proceeding 

package main

import (
	"fmt"
	"sync"

	"github.com/ChaokunChang/protoactor-go/actor"
)

func main() {
	var wg sync.WaitGroup
	wg.Add(1)

	// callee will wait for the PING message
	callee := actor.EmptyRootContext.Spawn(actor.PropsFromFunc(func(c actor.Context) {
		if msg, ok := c.Message().(string); ok {
			fmt.Println(msg) // outputs PING
			c.Respond("PONG")
		}
	}))

	// caller will send a PING message and wait for the PONG
	caller := actor.EmptyRootContext.Spawn(actor.PropsFromFunc(func(c actor.Context) {
		switch msg := c.Message().(type) {
		// the first message an actor receives after it has started
		case *actor.Started:
			// send a PING to the callee, and specify the response
			// is sent to Self, which is this actor's PID
			c.Request(callee, "PING")

		case string:
			fmt.Println(msg) // PONG
			wg.Done()
		}
	}))

	wg.Wait()
	actor.EmptyRootContext.StopFuture(callee).Wait()
	actor.EmptyRootContext.StopFuture(caller).Wait()

}

Output:

PING
PONG

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	ErrInvalidLengthProtos        = fmt.Errorf("proto: negative length found during unmarshaling")
	ErrIntOverflowProtos          = fmt.Errorf("proto: integer overflow")
	ErrUnexpectedEndOfGroupProtos = fmt.Errorf("proto: unexpected end of group")
)
View Source 
var (
	EmptyMessageHeader = make(messageHeader)
)
View Source 
var EmptyRootContext = &RootContext{
	senderMiddleware: nil,
	spawnMiddleware:  nil,
	headers:          EmptyMessageHeader,
	guardianStrategy: nil,
}
View Source 
var ErrNameExists = errors.New("spawn: name exists")

ErrNameExists is the error used when an existing name is used for spawning an actor.

View Source 
var ErrTimeout = errors.New("future: timeout")

ErrTimeout is the error used when a future times out before receiving a result.

View Source 
var ProcessRegistry = &ProcessRegistryValue{
	Address:   localAddress,
	LocalPIDs: cmap.New(),
}

ProcessRegistry is a registry of all active processes.

NOTE: This should only be used for advanced scenarios

Functions

func SetLogLevel 

func SetLogLevel(level log.Level)

SetLogLevel sets the log level for the logger.

SetLogLevel is safe to call concurrently

func SetOptions 

func SetOptions(opts ...optionFn)

SetOptions is used to configure the actor system

func UnwrapEnvelope 

func UnwrapEnvelope(message interface{}) (ReadonlyMessageHeader, interface{}, *PID)

func UnwrapEnvelopeMessage 

func UnwrapEnvelopeMessage(message interface{}) interface{}

func WithDeadLetterSubscriber 

func WithDeadLetterSubscriber(fn func(evt interface{})) optionFn

WithDeadLetterSubscriber option replaces the default DeadLetterEvent event subscriber with fn.

fn will only receive *DeadLetterEvent messages

Specifying nil will clear the existing.

func WithSupervisorSubscriber 

func WithSupervisorSubscriber(fn func(evt interface{})) optionFn

WithSupervisorSubscriber option replaces the default SupervisorEvent event subscriber with fn.

fn will only receive *SupervisorEvent messages

Specifying nil will clear the existing.

Types

type Actor 

type Actor interface {
	Receive(c Context)
}

Actor is the interface that defines the Receive method.

Receive is sent messages to be processed from the mailbox associated with the instance of the actor

type ActorFunc 

type ActorFunc func(c Context)

The ActorFunc type is an adapter to allow the use of ordinary functions as actors to process messages

func (ActorFunc) Receive 

func (f ActorFunc) Receive(c Context)

Receive calls f(c)

type ActorProcess 

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

func NewActorProcess 

func NewActorProcess(mailbox mailbox.Mailbox) *ActorProcess

func (*ActorProcess) SendSystemMessage 

func (ref *ActorProcess) SendSystemMessage(pid *PID, message interface{})

func (*ActorProcess) SendUserMessage 

func (ref *ActorProcess) SendUserMessage(pid *PID, message interface{})

func (*ActorProcess) Stop 

func (ref *ActorProcess) Stop(pid *PID)

type AddressResolver 

type AddressResolver func(*PID) (Process, bool)

An AddressResolver is used to resolve remote actors

type AutoReceiveMessage 

type AutoReceiveMessage interface {
	AutoReceiveMessage()
}

An AutoReceiveMessage is a special kind of user message that will be handled in some way automatically by the actor

type Behavior 

type Behavior []ActorFunc

func NewBehavior 

func NewBehavior() Behavior

func (*Behavior) Become 

func (b *Behavior) Become(receive ActorFunc)

func (*Behavior) BecomeStacked 

func (b *Behavior) BecomeStacked(receive ActorFunc)

func (*Behavior) Receive 

func (b *Behavior) Receive(context Context)

func (*Behavior) UnbecomeStacked 

func (b *Behavior) UnbecomeStacked()

type Context 

type Context interface {
	// contains filtered or unexported methods
}

Context contains contextual information for actors

Example (SetReceiveTimeout)

SetReceiveTimeout allows an actor to be notified repeatedly if it does not receive any messages for a specified duration 

package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/ChaokunChang/protoactor-go/actor"
)

type setReceiveTimeoutActor struct {
	*sync.WaitGroup
}

// Receive is the default message handler when an actor is started
func (f *setReceiveTimeoutActor) Receive(context actor.Context) {
	switch context.Message().(type) {
	case *actor.Started:
		// when the actor starts, set the receive timeout to 10 milliseconds.
		//
		// the system will send an *actor.ReceiveTimeout message every time the timeout
		// expires until SetReceiveTimeout is called again with a duration < 1 ms]
		context.SetReceiveTimeout(10 * time.Millisecond)
	case *actor.ReceiveTimeout:
		fmt.Println("timed out")
		f.Done()
	}
}

// SetReceiveTimeout allows an actor to be notified repeatedly if it does not receive any messages
// for a specified duration
func main() {
	wg := &sync.WaitGroup{}
	wg.Add(1)

	pid := actor.EmptyRootContext.Spawn(actor.PropsFromProducer(func() actor.Actor { return &setReceiveTimeoutActor{wg} }))
	defer func() {
		actor.EmptyRootContext.StopFuture(pid).Wait()
	}()

	wg.Wait() // wait for the ReceiveTimeout message

}

Output:

timed out

type ContextDecorator 

type ContextDecorator func(next ContextDecoratorFunc) ContextDecoratorFunc

type ContextDecoratorFunc 

type ContextDecoratorFunc func(ctx Context) Context

type DeadLetterEvent 

type DeadLetterEvent struct {
	PID     *PID        // The invalid process, to which the message was sent
	Message interface{} // The message that could not be delivered
	Sender  *PID        // the process that sent the Message
}

A DeadLetterEvent is published via event.Publish when a message is sent to a nonexistent PID

type DeciderFunc 

type DeciderFunc func(reason interface{}) Directive

DeciderFunc is a function which is called by a SupervisorStrategy

type Directive 

type Directive int

Directive is an enum for supervision actions 

const (
	// ResumeDirective instructs the supervisor to resume the actor and continue processing messages
	ResumeDirective Directive = iota

	// RestartDirective instructs the supervisor to discard the actor, replacing it with a new instance,
	// before processing additional messages
	RestartDirective

	// StopDirective instructs the supervisor to stop the actor
	StopDirective

	// EscalateDirective instructs the supervisor to escalate handling of the failure to the actor's parent supervisor
	EscalateDirective
)

Directive determines how a supervisor should handle a faulting actor

func DefaultDecider 

func DefaultDecider(_ interface{}) Directive

DefaultDecider is a decider that will always restart the failing child actor

func (Directive) String 

func (i Directive) String() string

type Failure 

type Failure struct {
	Who          *PID
	Reason       interface{}
	RestartStats *RestartStatistics
	Message      interface{}
}

func (*Failure) SystemMessage 

func (*Failure) SystemMessage()

type Future 

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

func NewFuture 

func NewFuture(d time.Duration) *Future

NewFuture creates and returns a new actor.Future with a timeout of duration d

func (*Future) PID 

func (f *Future) PID() *PID

PID to the backing actor for the Future result

func (*Future) PipeTo 

func (f *Future) PipeTo(pids ...*PID)

PipeTo forwards the result or error of the future to the specified pids

Example 
package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/ChaokunChang/protoactor-go/actor"
)

func main() {
	var wg sync.WaitGroup
	wg.Add(1)

	// test actor that will be the target of the future PipeTo
	pid := actor.EmptyRootContext.Spawn(actor.PropsFromFunc(func(ctx actor.Context) {
		// check if the message is a string and therefore
		// the "hello world" message piped from the future
		if m, ok := ctx.Message().(string); ok {
			fmt.Println(m)
			wg.Done()
		}
	}))

	f := actor.NewFuture(50 * time.Millisecond)
	f.PipeTo(pid)
	// resolve the future and pipe to waiting actor
	actor.EmptyRootContext.Send(f.PID(), "hello world")
	wg.Wait()

}

Output:

hello world

func (*Future) Result 

func (f *Future) Result() (interface{}, error)

Result waits for the future to resolve

func (*Future) Wait 

func (f *Future) Wait() error

type InboundMiddleware deprecated 

type InboundMiddleware func(f ActorFunc) ActorFunc

Deprecated: Use ReceiverMiddleware instead

type MessageEnvelope 

type MessageEnvelope struct {
	Header  messageHeader
	Message interface{}
	Sender  *PID
}

func WrapEnvelope 

func WrapEnvelope(message interface{}) *MessageEnvelope

func (*MessageEnvelope) GetHeader 

func (me *MessageEnvelope) GetHeader(key string) string

func (*MessageEnvelope) SetHeader 

func (me *MessageEnvelope) SetHeader(key string, value string)

type NotInfluenceReceiveTimeout 

type NotInfluenceReceiveTimeout interface {
	NotInfluenceReceiveTimeout()
}

NotInfluenceReceiveTimeout messages will not reset the ReceiveTimeout timer of an actor that receives the message

type OutboundMiddleware deprecated 

type OutboundMiddleware func(next SenderFunc) SenderFunc

Deprecated: Use SenderMiddleware instead

type PID 

type PID struct {
	Address string `protobuf:"bytes,1,opt,name=Address,proto3" json:"Address,omitempty"`
	Id      string `protobuf:"bytes,2,opt,name=Id,proto3" json:"Id,omitempty"`
	// contains filtered or unexported fields
}

func NewLocalPID 

func NewLocalPID(id string) *PID

NewLocalPID returns a new instance of the PID struct with the address preset

func NewPID 

func NewPID(address, id string) *PID

NewPID returns a new instance of the PID struct

func Spawn 

func Spawn(props *Props) *PID

Spawn starts a new actor based on props and named with a unique id Deprecated: Use context.Spawn instead.

Example

Spawn creates instances of actors, similar to 'new' or 'make' but for actors. 

package main

import (
	"fmt"
	"sync"

	"github.com/ChaokunChang/protoactor-go/actor"
)

func main() {
	var wg sync.WaitGroup
	wg.Add(1)

	// create root context
	context := actor.EmptyRootContext

	// define the actor props
	// props define the creation process of an actor
	props := actor.PropsFromFunc(func(ctx actor.Context) {
		// check if the message is a *actor.Started message
		// this is the first message all actors get
		// actor.Started is received async and can be used
		// to initialize your actors initial state
		if _, ok := ctx.Message().(*actor.Started); ok {
			fmt.Println("hello world")
			wg.Done()
		}
	})

	// spawn the actor based on the props
	context.Spawn(props)
	wg.Wait()
}

Output:

hello world

func SpawnNamed 

func SpawnNamed(props *Props, name string) (*PID, error)

SpawnNamed starts a new actor based on props and named using the specified name

If name exists, error will be ErrNameExists Deprecated: Use context.SpawnNamed instead.

Example

Spawn creates instances of actors, similar to 'new' or 'make' but for actors. 

package main

import (
	"fmt"
	"log"
	"sync"

	"github.com/ChaokunChang/protoactor-go/actor"
)

func main() {
	var wg sync.WaitGroup
	wg.Add(1)

	// create root context
	context := actor.EmptyRootContext

	// define the actor props
	// props define the creation process of an actor
	props := actor.PropsFromFunc(func(ctx actor.Context) {
		// check if the message is a *actor.Started message
		// this is the first message all actors get
		// actor.Started is received async and can be used
		// to initialize your actors initial state
		if _, ok := ctx.Message().(*actor.Started); ok {
			fmt.Println("hello world")
			wg.Done()
		}
	})

	// spawn the actor based on the props
	_, err := context.SpawnNamed(props, "my-actor")
	if err != nil {
		log.Fatal("The actor name is already in use")
	}
	wg.Wait()
}

Output:

hello world

func SpawnPrefix 

func SpawnPrefix(props *Props, prefix string) *PID

SpawnPrefix starts a new actor based on props and named using a prefix followed by a unique id Deprecated: Use context.SpawnPrefix instead.

func UnwrapEnvelopeSender 

func UnwrapEnvelopeSender(message interface{}) *PID

func (*PID) Descriptor 

func (*PID) Descriptor() ([]byte, []int)

func (*PID) Equal 

func (this *PID) Equal(that interface{}) bool

func (*PID) GetAddress 

func (m *PID) GetAddress() string

func (*PID) GetId 

func (m *PID) GetId() string

func (*PID) GracefulPoison deprecated 

func (pid *PID) GracefulPoison()

GracefulPoison will tell and wait actor to stop after processing current user messages in mailbox.

Deprecated: Use Context.PoisonFuture(pid).Wait() instead

func (*PID) GracefulStop deprecated 

func (pid *PID) GracefulStop()

GracefulStop will stop actor immediately regardless of existing user messages in mailbox.

Deprecated: Use Context.StopFuture(pid).Wait() instead

func (*PID) Marshal 

func (m *PID) Marshal() (dAtA []byte, err error)

func (*PID) MarshalTo 

func (m *PID) MarshalTo(dAtA []byte) (int, error)

func (*PID) MarshalToSizedBuffer 

func (m *PID) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*PID) Poison deprecated 

func (pid *PID) Poison()

Poison will tell actor to stop after processing current user messages in mailbox.

Deprecated: Use Context.Poison instead

func (*PID) PoisonFuture deprecated 

func (pid *PID) PoisonFuture() *Future

PoisonFuture will tell actor to stop after processing current user messages in mailbox, and return its future.

Deprecated: Use Context.PoisonFuture instead

func (*PID) ProtoMessage 

func (*PID) ProtoMessage()

func (*PID) Request deprecated 

func (pid *PID) Request(message interface{}, respondTo *PID)

Deprecated: Use Context.Request or Context.RequestWithCustomSender instead

func (*PID) RequestFuture deprecated 

func (pid *PID) RequestFuture(message interface{}, timeout time.Duration) *Future

Deprecated: Use Context.RequestFuture instead

func (*PID) Reset 

func (m *PID) Reset()

func (*PID) Size 

func (m *PID) Size() (n int)

func (*PID) Stop deprecated 

func (pid *PID) Stop()

Stop will stop actor immediately regardless of existing user messages in mailbox.

Deprecated: Use Context.Stop instead

func (*PID) StopFuture deprecated 

func (pid *PID) StopFuture() *Future

StopFuture will stop actor immediately regardless of existing user messages in mailbox, and return its future.

Deprecated: Use Context.StopFuture instead

func (*PID) String 

func (pid *PID) String() string

func (*PID) Tell deprecated 

func (pid *PID) Tell(message interface{})

Deprecated: Use Context.Send instead

func (*PID) Unmarshal 

func (m *PID) Unmarshal(dAtA []byte) error

func (*PID) XXX_DiscardUnknown 

func (m *PID) XXX_DiscardUnknown()

func (*PID) XXX_Marshal 

func (m *PID) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*PID) XXX_Merge 

func (m *PID) XXX_Merge(src proto.Message)

func (*PID) XXX_Size 

func (m *PID) XXX_Size() int

func (*PID) XXX_Unmarshal 

func (m *PID) XXX_Unmarshal(b []byte) error

type PIDSet 

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

func NewPIDSet 

func NewPIDSet(pids ...*PID) *PIDSet

NewPIDSet returns a new PIDSet with the given pids.

func (*PIDSet) Add 

func (p *PIDSet) Add(v *PID)

Add adds the element v to the set

func (*PIDSet) Clear 

func (p *PIDSet) Clear()

Clear removes all the elements in the set

func (*PIDSet) Clone 

func (p *PIDSet) Clone() *PIDSet

func (*PIDSet) Contains 

func (p *PIDSet) Contains(v *PID) bool

Contains reports whether v is an element of the set

func (*PIDSet) Empty 

func (p *PIDSet) Empty() bool

Empty reports whether the set is empty

func (*PIDSet) ForEach 

func (p *PIDSet) ForEach(f func(i int, pid PID))

ForEach invokes f for every element of the set

func (*PIDSet) Len 

func (p *PIDSet) Len() int

Len returns the number of elements in the set

func (*PIDSet) Remove 

func (p *PIDSet) Remove(v *PID) bool

Remove removes v from the set and returns true if them element existed

func (*PIDSet) Values 

func (p *PIDSet) Values() []PID

Values returns all the elements of the set as a slice

type PoisonPill 

type PoisonPill struct {
}

user messages

func (*PoisonPill) AutoReceiveMessage 

func (*PoisonPill) AutoReceiveMessage()

func (*PoisonPill) Descriptor 

func (*PoisonPill) Descriptor() ([]byte, []int)

func (*PoisonPill) Equal 

func (this *PoisonPill) Equal(that interface{}) bool

func (*PoisonPill) Marshal 

func (m *PoisonPill) Marshal() (dAtA []byte, err error)

func (*PoisonPill) MarshalTo 

func (m *PoisonPill) MarshalTo(dAtA []byte) (int, error)

func (*PoisonPill) MarshalToSizedBuffer 

func (m *PoisonPill) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*PoisonPill) ProtoMessage 

func (*PoisonPill) ProtoMessage()

func (*PoisonPill) Reset 

func (m *PoisonPill) Reset()

func (*PoisonPill) Size 

func (m *PoisonPill) Size() (n int)

func (*PoisonPill) String 

func (this *PoisonPill) String() string

func (*PoisonPill) Unmarshal 

func (m *PoisonPill) Unmarshal(dAtA []byte) error

func (*PoisonPill) XXX_DiscardUnknown 

func (m *PoisonPill) XXX_DiscardUnknown()

func (*PoisonPill) XXX_Marshal 

func (m *PoisonPill) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*PoisonPill) XXX_Merge 

func (m *PoisonPill) XXX_Merge(src proto.Message)

func (*PoisonPill) XXX_Size 

func (m *PoisonPill) XXX_Size() int

func (*PoisonPill) XXX_Unmarshal 

func (m *PoisonPill) XXX_Unmarshal(b []byte) error

type Process 

type Process interface {
	SendUserMessage(pid *PID, message interface{})
	SendSystemMessage(pid *PID, message interface{})
	Stop(pid *PID)
}

A Process is an interface that defines the base contract for interaction of actors

type ProcessRegistryValue 

type ProcessRegistryValue struct {
	SequenceID     uint64
	Address        string
	LocalPIDs      cmap.ConcurrentMap
	RemoteHandlers []AddressResolver
}

func (*ProcessRegistryValue) Add 

func (pr *ProcessRegistryValue) Add(process Process, id string) (*PID, bool)

func (*ProcessRegistryValue) Get 

func (pr *ProcessRegistryValue) Get(pid *PID) (Process, bool)

func (*ProcessRegistryValue) GetLocal 

func (pr *ProcessRegistryValue) GetLocal(id string) (Process, bool)

func (*ProcessRegistryValue) NextId 

func (pr *ProcessRegistryValue) NextId() string

func (*ProcessRegistryValue) RegisterAddressResolver 

func (pr *ProcessRegistryValue) RegisterAddressResolver(handler AddressResolver)

func (*ProcessRegistryValue) Remove 

func (pr *ProcessRegistryValue) Remove(pid *PID)

type Producer 

type Producer func() Actor

The Producer type is a function that creates a new actor

type Props 

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

Props represents configuration to define how an actor should be created

func FromFunc deprecated 

func FromFunc(f ActorFunc) *Props

Deprecated: Use actor.PropsFromFunc instead.

func FromProducer deprecated 

func FromProducer(actorProducer Producer) *Props

Deprecated: Use actor.PropsFromProducer instead.

func FromSpawnFunc deprecated 

func FromSpawnFunc(spawn SpawnFunc) *Props

Deprecated: FromSpawnFunc is deprecated.

func PropsFromFunc 

func PropsFromFunc(f ActorFunc) *Props

PropsFromFunc creates a props with the given receive func assigned as the actor producer

func PropsFromProducer 

func PropsFromProducer(producer Producer) *Props

PropsFromProducer creates a props with the given actor producer assigned

func (*Props) WithContextDecorator 

func (props *Props) WithContextDecorator(contextDecorator ...ContextDecorator) *Props

WithContextDecorator assigns context decorator to the props

func (*Props) WithDispatcher 

func (props *Props) WithDispatcher(dispatcher mailbox.Dispatcher) *Props

WithDispatcher assigns a dispatcher to the props

func (*Props) WithFunc 

func (props *Props) WithFunc(f ActorFunc) *Props

WithFunc assigns a receive func to the props

func (*Props) WithGuardian 

func (props *Props) WithGuardian(guardian SupervisorStrategy) *Props

WithGuardian assigns a guardian strategy to the props

func (*Props) WithMailbox 

func (props *Props) WithMailbox(mailbox mailbox.Producer) *Props

WithMailbox assigns the desired mailbox producer to the props

func (*Props) WithMiddleware deprecated 

func (props *Props) WithMiddleware(middleware ...InboundMiddleware) *Props

Deprecated: Use WithReceiverMiddleware instead

func (*Props) WithOutboundMiddleware deprecated 

func (props *Props) WithOutboundMiddleware(middleware ...OutboundMiddleware) *Props

Deprecated: Use WithSenderMiddleware instead

func (*Props) WithProducer 

func (props *Props) WithProducer(p Producer) *Props

WithProducer assigns a actor producer to the props

func (*Props) WithReceiverMiddleware 

func (props *Props) WithReceiverMiddleware(middleware ...ReceiverMiddleware) *Props

Assign one or more middleware to the props

func (*Props) WithSenderMiddleware 

func (props *Props) WithSenderMiddleware(middleware ...SenderMiddleware) *Props

func (*Props) WithSpawnFunc 

func (props *Props) WithSpawnFunc(spawn SpawnFunc) *Props

WithSpawnFunc assigns a custom spawn func to the props, this is mainly for internal usage

func (*Props) WithSpawnMiddleware 

func (props *Props) WithSpawnMiddleware(middleware ...SpawnMiddleware) *Props

func (*Props) WithSupervisor 

func (props *Props) WithSupervisor(supervisor SupervisorStrategy) *Props

WithSupervisor assigns a supervision strategy to the props

type ReadonlyMessageHeader 

type ReadonlyMessageHeader interface {
	Get(key string) string
	Keys() []string
	Length() int
	ToMap() map[string]string
}

func UnwrapEnvelopeHeader 

func UnwrapEnvelopeHeader(message interface{}) ReadonlyMessageHeader

type ReceiveTimeout 

type ReceiveTimeout struct{}

A ReceiveTimeout message is sent to an actor after the Context.ReceiveTimeout duration has expired

type ReceiverContext 

type ReceiverContext interface {
	// contains filtered or unexported methods
}

type ReceiverFunc 

type ReceiverFunc func(c ReceiverContext, envelope *MessageEnvelope)

type ReceiverMiddleware 

type ReceiverMiddleware func(next ReceiverFunc) ReceiverFunc

type Restart 

type Restart struct{}

Restart is message sent by the actor system to control the lifecycle of an actor

func (*Restart) SystemMessage 

func (*Restart) SystemMessage()

type RestartStatistics 

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

RestartStatistics keeps track of how many times an actor have restarted and when

func NewRestartStatistics 

func NewRestartStatistics() *RestartStatistics

NewRestartStatistics construct a RestartStatistics

func (*RestartStatistics) Fail 

func (rs *RestartStatistics) Fail()

Fail increases the associated actors failure count

func (*RestartStatistics) FailureCount 

func (rs *RestartStatistics) FailureCount() int

FailureCount returns failure count

func (*RestartStatistics) NumberOfFailures 

func (rs *RestartStatistics) NumberOfFailures(withinDuration time.Duration) int

NumberOfFailures returns number of failures within a given duration

func (*RestartStatistics) Reset 

func (rs *RestartStatistics) Reset()

Reset the associated actors failure count

type Restarting 

type Restarting struct{}

A Restarting message is sent to an actor when the actor is being restarted by the system due to a failure

func (*Restarting) AutoReceiveMessage 

func (*Restarting) AutoReceiveMessage()

type RootContext 

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

func NewRootContext 

func NewRootContext(header map[string]string, middleware ...SenderMiddleware) *RootContext

func (*RootContext) Actor 

func (rc *RootContext) Actor() Actor

func (RootContext) Copy 

func (rc RootContext) Copy() *RootContext

func (*RootContext) Message 

func (rc *RootContext) Message() interface{}

func (*RootContext) MessageHeader 

func (rc *RootContext) MessageHeader() ReadonlyMessageHeader

func (*RootContext) Parent 

func (rc *RootContext) Parent() *PID

func (*RootContext) Poison 

func (rc *RootContext) Poison(pid *PID)

Poison will tell actor to stop after processing current user messages in mailbox.

func (*RootContext) PoisonFuture 

func (rc *RootContext) PoisonFuture(pid *PID) *Future

PoisonFuture will tell actor to stop after processing current user messages in mailbox, and return its future.

func (*RootContext) Request 

func (rc *RootContext) Request(pid *PID, message interface{})

func (*RootContext) RequestFuture 

func (rc *RootContext) RequestFuture(pid *PID, message interface{}, timeout time.Duration) *Future

RequestFuture sends a message to a given PID and returns a Future

func (*RootContext) RequestWithCustomSender 

func (rc *RootContext) RequestWithCustomSender(pid *PID, message interface{}, sender *PID)

func (*RootContext) Self 

func (rc *RootContext) Self() *PID

func (*RootContext) Send 

func (rc *RootContext) Send(pid *PID, message interface{})

func (*RootContext) Sender 

func (rc *RootContext) Sender() *PID

func (*RootContext) Spawn 

func (rc *RootContext) Spawn(props *Props) *PID

Spawn starts a new actor based on props and named with a unique id

func (*RootContext) SpawnNamed 

func (rc *RootContext) SpawnNamed(props *Props, name string) (*PID, error)

SpawnNamed starts a new actor based on props and named using the specified name

ErrNameExists will be returned if id already exists

Please do not use name sharing same pattern with system actors, for example "YourPrefix$1", "Remote$1", "future$1"

func (*RootContext) SpawnPrefix 

func (rc *RootContext) SpawnPrefix(props *Props, prefix string) *PID

SpawnPrefix starts a new actor based on props and named using a prefix followed by a unique id

func (*RootContext) Stop 

func (rc *RootContext) Stop(pid *PID)

Stop will stop actor immediately regardless of existing user messages in mailbox.

func (*RootContext) StopFuture 

func (rc *RootContext) StopFuture(pid *PID) *Future

StopFuture will stop actor immediately regardless of existing user messages in mailbox, and return its future.

func (*RootContext) WithGuardian 

func (rc *RootContext) WithGuardian(guardian SupervisorStrategy) *RootContext

func (*RootContext) WithHeaders 

func (rc *RootContext) WithHeaders(headers map[string]string) *RootContext

func (*RootContext) WithSenderMiddleware 

func (rc *RootContext) WithSenderMiddleware(middleware ...SenderMiddleware) *RootContext

func (*RootContext) WithSpawnMiddleware 

func (rc *RootContext) WithSpawnMiddleware(middleware ...SpawnMiddleware) *RootContext

type SenderContext 

type SenderContext interface {
	// contains filtered or unexported methods
}

type SenderFunc 

type SenderFunc func(c SenderContext, target *PID, envelope *MessageEnvelope)

type SenderMiddleware 

type SenderMiddleware func(next SenderFunc) SenderFunc

type SpawnFunc 

type SpawnFunc func(id string, props *Props, parentContext SpawnerContext) (*PID, error)

Props types 

var DefaultSpawner SpawnFunc = defaultSpawner

DefaultSpawner this is a hacking way to allow Proto.Router access default spawner func

type SpawnMiddleware 

type SpawnMiddleware func(next SpawnFunc) SpawnFunc

type SpawnerContext 

type SpawnerContext interface {
	// contains filtered or unexported methods
}

type Started 

type Started struct{}

A Started message is sent to an actor once it has been started and ready to begin receiving messages.

func (*Started) SystemMessage 

func (*Started) SystemMessage()

type Stop 

type Stop struct {
}

func (*Stop) Descriptor 

func (*Stop) Descriptor() ([]byte, []int)

func (*Stop) Equal 

func (this *Stop) Equal(that interface{}) bool

func (*Stop) Marshal 

func (m *Stop) Marshal() (dAtA []byte, err error)

func (*Stop) MarshalTo 

func (m *Stop) MarshalTo(dAtA []byte) (int, error)

func (*Stop) MarshalToSizedBuffer 

func (m *Stop) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*Stop) ProtoMessage 

func (*Stop) ProtoMessage()

func (*Stop) Reset 

func (m *Stop) Reset()

func (*Stop) Size 

func (m *Stop) Size() (n int)

func (*Stop) String 

func (this *Stop) String() string

func (*Stop) SystemMessage 

func (*Stop) SystemMessage()

func (*Stop) Unmarshal 

func (m *Stop) Unmarshal(dAtA []byte) error

func (*Stop) XXX_DiscardUnknown 

func (m *Stop) XXX_DiscardUnknown()

func (*Stop) XXX_Marshal 

func (m *Stop) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Stop) XXX_Merge 

func (m *Stop) XXX_Merge(src proto.Message)

func (*Stop) XXX_Size 

func (m *Stop) XXX_Size() int

func (*Stop) XXX_Unmarshal 

func (m *Stop) XXX_Unmarshal(b []byte) error

type Stopped 

type Stopped struct{}

A Stopped message is sent to the actor once it has been stopped. A stopped actor will receive no further messages

func (*Stopped) AutoReceiveMessage 

func (*Stopped) AutoReceiveMessage()

type Stopping 

type Stopping struct{}

A Stopping message is sent to an actor prior to the actor being stopped

func (*Stopping) AutoReceiveMessage 

func (*Stopping) AutoReceiveMessage()

type Supervisor 

type Supervisor interface {
	Children() []*PID
	EscalateFailure(reason interface{}, message interface{})
	RestartChildren(pids ...*PID)
	StopChildren(pids ...*PID)
	ResumeChildren(pids ...*PID)
}

Supervisor is an interface that is used by the SupervisorStrategy to manage child actor lifecycle

type SupervisorEvent 

type SupervisorEvent struct {
	Child     *PID
	Reason    interface{}
	Directive Directive
}

SupervisorEvent is sent on the EventStream when a supervisor have applied a directive to a failing child actor

type SupervisorStrategy 

type SupervisorStrategy interface {
	HandleFailure(supervisor Supervisor, child *PID, rs *RestartStatistics, reason interface{}, message interface{})
}

SupervisorStrategy is an interface that decides how to handle failing child actors

func DefaultSupervisorStrategy 

func DefaultSupervisorStrategy() SupervisorStrategy

func NewAllForOneStrategy 

func NewAllForOneStrategy(maxNrOfRetries int, withinDuration time.Duration, decider DeciderFunc) SupervisorStrategy

NewAllForOneStrategy returns a new SupervisorStrategy which applies the given fault Directive from the decider to the failing child and all its children.

This strategy is appropriate when the children have a strong dependency, such that and any single one failing would place them all into a potentially invalid state.

func NewExponentialBackoffStrategy 

func NewExponentialBackoffStrategy(backoffWindow time.Duration, initialBackoff time.Duration) SupervisorStrategy

NewExponentialBackoffStrategy creates a new Supervisor strategy that restarts a faulting child using an exponential back off algorithm: 

delay =

func NewOneForOneStrategy 

func NewOneForOneStrategy(maxNrOfRetries int, withinDuration time.Duration, decider DeciderFunc) SupervisorStrategy

NewOneForOneStrategy returns a new Supervisor strategy which applies the fault Directive from the decider to the failing child process.

This strategy is applicable if it is safe to handle a single child in isolation from its peers or dependents

func NewRestartingStrategy 

func NewRestartingStrategy() SupervisorStrategy

func RestartingSupervisorStrategy 

func RestartingSupervisorStrategy() SupervisorStrategy

type SystemMessage 

type SystemMessage interface {
	SystemMessage()
}

A SystemMessage message is reserved for specific lifecycle messages used by the actor system

type Terminated 

type Terminated struct {
	Who               *PID `protobuf:"bytes,1,opt,name=who,proto3" json:"who,omitempty"`
	AddressTerminated bool `protobuf:"varint,2,opt,name=address_terminated,json=addressTerminated,proto3" json:"address_terminated,omitempty"`
}

func (*Terminated) Descriptor 

func (*Terminated) Descriptor() ([]byte, []int)

func (*Terminated) Equal 

func (this *Terminated) Equal(that interface{}) bool

func (*Terminated) GetAddressTerminated 

func (m *Terminated) GetAddressTerminated() bool

func (*Terminated) GetWho 

func (m *Terminated) GetWho() *PID

func (*Terminated) Marshal 

func (m *Terminated) Marshal() (dAtA []byte, err error)

func (*Terminated) MarshalTo 

func (m *Terminated) MarshalTo(dAtA []byte) (int, error)

func (*Terminated) MarshalToSizedBuffer 

func (m *Terminated) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*Terminated) ProtoMessage 

func (*Terminated) ProtoMessage()

func (*Terminated) Reset 

func (m *Terminated) Reset()

func (*Terminated) Size 

func (m *Terminated) Size() (n int)

func (*Terminated) String 

func (this *Terminated) String() string

func (*Terminated) SystemMessage 

func (*Terminated) SystemMessage()

func (*Terminated) Unmarshal 

func (m *Terminated) Unmarshal(dAtA []byte) error

func (*Terminated) XXX_DiscardUnknown 

func (m *Terminated) XXX_DiscardUnknown()

func (*Terminated) XXX_Marshal 

func (m *Terminated) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Terminated) XXX_Merge 

func (m *Terminated) XXX_Merge(src proto.Message)

func (*Terminated) XXX_Size 

func (m *Terminated) XXX_Size() int

func (*Terminated) XXX_Unmarshal 

func (m *Terminated) XXX_Unmarshal(b []byte) error

type Unwatch 

type Unwatch struct {
	Watcher *PID `protobuf:"bytes,1,opt,name=watcher,proto3" json:"watcher,omitempty"`
}

func (*Unwatch) Descriptor 

func (*Unwatch) Descriptor() ([]byte, []int)

func (*Unwatch) Equal 

func (this *Unwatch) Equal(that interface{}) bool

func (*Unwatch) GetWatcher 

func (m *Unwatch) GetWatcher() *PID

func (*Unwatch) Marshal 

func (m *Unwatch) Marshal() (dAtA []byte, err error)

func (*Unwatch) MarshalTo 

func (m *Unwatch) MarshalTo(dAtA []byte) (int, error)

func (*Unwatch) MarshalToSizedBuffer 

func (m *Unwatch) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*Unwatch) ProtoMessage 

func (*Unwatch) ProtoMessage()

func (*Unwatch) Reset 

func (m *Unwatch) Reset()

func (*Unwatch) Size 

func (m *Unwatch) Size() (n int)

func (*Unwatch) String 

func (this *Unwatch) String() string

func (*Unwatch) SystemMessage 

func (*Unwatch) SystemMessage()

func (*Unwatch) Unmarshal 

func (m *Unwatch) Unmarshal(dAtA []byte) error

func (*Unwatch) XXX_DiscardUnknown 

func (m *Unwatch) XXX_DiscardUnknown()

func (*Unwatch) XXX_Marshal 

func (m *Unwatch) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Unwatch) XXX_Merge 

func (m *Unwatch) XXX_Merge(src proto.Message)

func (*Unwatch) XXX_Size 

func (m *Unwatch) XXX_Size() int

func (*Unwatch) XXX_Unmarshal 

func (m *Unwatch) XXX_Unmarshal(b []byte) error

type Watch 

type Watch struct {
	Watcher *PID `protobuf:"bytes,1,opt,name=watcher,proto3" json:"watcher,omitempty"`
}

system messages

func (*Watch) Descriptor 

func (*Watch) Descriptor() ([]byte, []int)

func (*Watch) Equal 

func (this *Watch) Equal(that interface{}) bool

func (*Watch) GetWatcher 

func (m *Watch) GetWatcher() *PID

func (*Watch) Marshal 

func (m *Watch) Marshal() (dAtA []byte, err error)

func (*Watch) MarshalTo 

func (m *Watch) MarshalTo(dAtA []byte) (int, error)

func (*Watch) MarshalToSizedBuffer 

func (m *Watch) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*Watch) ProtoMessage 

func (*Watch) ProtoMessage()

func (*Watch) Reset 

func (m *Watch) Reset()

func (*Watch) Size 

func (m *Watch) Size() (n int)

func (*Watch) String 

func (this *Watch) String() string

func (*Watch) SystemMessage 

func (*Watch) SystemMessage()

func (*Watch) Unmarshal 

func (m *Watch) Unmarshal(dAtA []byte) error

func (*Watch) XXX_DiscardUnknown 

func (m *Watch) XXX_DiscardUnknown()

func (*Watch) XXX_Marshal 

func (m *Watch) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Watch) XXX_Merge 

func (m *Watch) XXX_Merge(src proto.Message)

func (*Watch) XXX_Size 

func (m *Watch) XXX_Size() int

func (*Watch) XXX_Unmarshal 

func (m *Watch) XXX_Unmarshal(b []byte) error

Source Files

View all Source files

Directories

Path Synopsis
opentracing
propagator
protozip

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.