node

package
v2.6.3 Latest Latest
Warning

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

 Go to latest Published: Oct 21, 2026 License: MIT Imports: 34 Imported by: 9

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	InternalRoute   = RouteOptions{Internal: true}   // Internal route, messages flow between nodes only
	StatefulRoute   = RouteOptions{Stateful: true}   // Stateful route, always routed to the bound node
	AuthorizedRoute = RouteOptions{Authorized: true} // Authorized route, must carry the UID when flowing
)

Functions

This section is empty.

Types

type Actor added in v2.2.0 

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

Actor is an actor model.

An actor owns an independent message queue and task queue, which keeps its internal processing thread-safe.

func (*Actor) AddEventHandler added in v2.2.0 

func (a *Actor) AddEventHandler(event cluster.Event, handler EventHandler)

AddEventHandler registers handler for the given event. It does nothing when the actor is not started.

func (*Actor) AddRouteHandler added in v2.2.0 

func (a *Actor) AddRouteHandler[T constraints.Integer](route T, handler RouteHandler)

AddRouteHandler registers handler for the given route. It does nothing when the actor is not started.

func (*Actor) AfterFunc added in v2.2.4 

func (a *Actor) AfterFunc(d time.Duration, f func()) (*Timer, error)

AfterFunc schedules f to run after d, matching the semantics of time.AfterFunc. The returned Timer can be cancelled with Timer.Stop.

func (*Actor) AfterInvoke added in v2.2.4 

func (a *Actor) AfterInvoke(d time.Duration, f func()) (*Timer, error)

AfterInvoke schedules f to run after d in a thread-safe manner.

After the delay, f is executed serially through the task queue, which keeps the actor thread-safe. The returned Timer can be cancelled with Timer.Stop.

func (*Actor) Deliver added in v2.2.0 

func (a *Actor) Deliver(uid int64, message *cluster.Message) error

Deliver delivers message to the current actor for the given user.

The delivered message originates from the local node and has no source actor.

It reports an error when packing the message or delivering it fails.

func (*Actor) Destroy added in v2.2.0 

func (a *Actor) Destroy() bool

Destroy destroys the actor. It reports false when the actor does not exist or has already been destroyed.

It shares the implementation of [Scheduler.kill]: the actor is removed from the scheduler before being destroyed, so that exactly one of the concurrent Destroy/kill callers reports success.

func (*Actor) ID added in v2.2.0 

func (a *Actor) ID() string

ID returns the actor ID.

func (*Actor) Invoke added in v2.2.0 

func (a *Actor) Invoke(f func(), wait ...bool) error

Invoke calls f in a thread-safe manner inside the actor.

The call is written to the actor's task queue and executed serially; in blocking mode (wait is true) it waits for the function to complete and by default it does not wait. A synchronous call made from the actor's own dispatcher goroutine is executed directly, avoiding a deadlock caused by waiting on the queue the caller is running in. Do not wait synchronously across queues in the dispatch chain, for example an actor task that waits synchronously for a node task while that node task waits synchronously for an actor task; the mutual wait forms a cross-queue circular wait deadlock.

func (*Actor) Kind added in v2.2.0 

func (a *Actor) Kind() string

Kind returns the actor kind.

func (*Actor) Next added in v2.2.0 

func (a *Actor) Next(ctx Context) error

Next delivers ctx to the actor for processing.

The context is written to the actor's message queue and processed serially. It reports an error when the actor has not been started or when enqueuing the message fails.

func (*Actor) PID added in v2.2.0 

func (a *Actor) PID() string

PID returns the unique identifier of the actor.

The identifier is composed of the kind and the ID and is used to globally locate the actor.

func (*Actor) Processor added in v2.6.3 

func (a *Actor) Processor() Processor

Processor returns the processor of the actor.

It reports nil when the actor has been destroyed.

func (*Actor) Proxy added in v2.2.0 

func (a *Actor) Proxy() *Proxy

Proxy returns the node proxy.

func (*Actor) Push added in v2.2.1 

func (a *Actor) Push(uid int64, message *cluster.Message) error

Push pushes message to the local node queue for the given user.

It reports an error when packing the message fails.

func (*Actor) SetDefaultRouteHandler added in v2.3.0 

func (a *Actor) SetDefaultRouteHandler(handler RouteHandler)

SetDefaultRouteHandler sets the default route handler, which handles every route that has not been registered. It does nothing when the actor is not started.

func (*Actor) Spawn added in v2.2.0 

func (a *Actor) Spawn(creator Creator, opts ...ActorOption) (*Actor, error)

Spawn derives a new actor from the actor.

creator creates the processor of the new actor and opts configures it. It reports an error when an actor with the same kind and ID already exists or when creation fails.

type ActorOption added in v2.2.0 

type ActorOption func(o *actorOptions)

ActorOption is a configuration function of an Actor.

func WithActorArgs added in v2.2.0 

func WithActorArgs(args ...any) ActorOption

WithActorArgs sets the arguments passed to the Processor.

func WithActorID added in v2.2.0 

func WithActorID(id string) ActorOption

WithActorID sets the actor ID.

func WithActorKind added in v2.2.4 

func WithActorKind(kind string) ActorOption

WithActorKind sets the actor kind.

func WithActorMessageQueueSize added in v2.6.0 

func WithActorMessageQueueSize(size int32) ActorOption

WithActorMessageQueueSize sets the message queue capacity.

func WithActorMessageWriteTimeout added in v2.6.0 

func WithActorMessageWriteTimeout(timeout time.Duration) ActorOption

WithActorMessageWriteTimeout sets the timeout for writing a message to the message queue.

func WithActorNonDispatch added in v2.2.4 

func WithActorNonDispatch() ActorOption

WithActorNonDispatch makes the actor non-schedulable.

func WithActorNonWait added in v2.2.3 

func WithActorNonWait() ActorOption

WithActorNonWait marks the actor as not needing to be waited for, so that the Node component does not wait for this actor to finish when shutting down.

func WithActorTaskQueueSize added in v2.6.0 

func WithActorTaskQueueSize(size int32) ActorOption

WithActorTaskQueueSize sets the task queue capacity.

func WithActorTaskWriteTimeout added in v2.6.0 

func WithActorTaskWriteTimeout(timeout time.Duration) ActorOption

WithActorTaskWriteTimeout sets the timeout for writing a task to the task queue.

type BaseProcessor added in v2.2.0 

type BaseProcessor struct{}

BaseProcessor is a base processor.

It provides empty implementations of every callback so that a custom Processor can embed it to avoid implementing the empty methods.

func (*BaseProcessor) Destroy added in v2.2.0 

func (b *BaseProcessor) Destroy()

Destroy is the destroy callback.

func (*BaseProcessor) Init added in v2.2.0 

func (b *BaseProcessor) Init()

Init is the init callback.

func (*BaseProcessor) Start added in v2.2.0 

func (b *BaseProcessor) Start()

Start is the start callback.

type Context 

type Context interface {
	// GID returns the ID of the source gateway.
	GID() string
	// NID returns the ID of the source node.
	NID() string
	// CID returns the connection ID.
	CID() int64
	// UID returns the user ID.
	UID() int64
	// Seq returns the message sequence number.
	Seq() int32
	// Route returns the message route number.
	Route() int32
	// Event returns the event type.
	Event() cluster.Event
	// Kind returns the message kind, either an event or a request.
	Kind() Kind
	// Parse parses the message.
	//
	// By default it deserializes into the target value; when an encryptor is present and the
	// message comes from a gateway, it decrypts the message before deserializing. It reports an
	// error when the payload is empty, or when decryption or deserialization fails.
	Parse(v any) error
	// Defer adds a deferred call to the call stack.
	//
	// This method behaves like the go defer statement and its scope is limited to the current
	// handler function, so it is recommended over go defer. The difference is that Defer allows
	// the call stack to be cancelled. The stack is cancelled automatically when Task or Next is
	// called, and it can also be cancelled manually with Cancel. The optional bottom controls
	// whether fn is pushed to the bottom of the stack; it is pushed to the top by default.
	Defer(fn func(), bottom ...bool)
	// Cancel cancels the defer call stack.
	Cancel()
	// Clone clones the context.
	//
	// It creates a new independent context, which is used to deliver the message to another actor.
	Clone() Context
	// Task submits a task.
	//
	// It is recommended over task.Add and go func; calling it automatically cancels every function
	// in the defer call stack.
	//
	// Task must be called at most once per context and must not be combined with Next. The task
	// shares the context object, which is recycled and has its fields cleared once the latest
	// stage finishes; overlapping stages would otherwise observe a cleared context.
	Task(fn func(ctx Context))
	// Proxy returns the node proxy.
	Proxy() *Proxy
	// Context returns the underlying standard context.
	Context() context.Context
	// SetValue sets a value on the context.
	SetValue(key, val any)
	// GetValue returns the value associated with key, or nil when it does not exist.
	GetValue(key any) any
	// GetIP returns the client IP. It reports an error when the source is illegal or the IP cannot
	// be retrieved.
	GetIP() (string, error)
	// Deliver delivers a message to a node for processing.
	Deliver(args *cluster.DeliverArgs) error
	// Reply replies with a message.
	//
	// According to the message source (gateway, actor or node), the message is sent back to the
	// corresponding target.
	Reply(message *cluster.Message) error
	// Response responds to the current route message.
	//
	// The response reuses the route number and sequence number of the original request.
	Response(message any) error
	// Disconnect closes the connection from the gateway. The optional force controls whether the
	// connection is disconnected forcibly; it is not forced by default.
	Disconnect(force ...bool) error
	// BindGate binds the gateway. When uid is not specified, the uid of the current context user is
	// used.
	BindGate(uid ...int64) error
	// UnbindGate unbinds the gateway. When uid is not specified, the uid of the current context
	// user is used.
	UnbindGate(uid ...int64) error
	// BindNode binds the node. When uid is not specified, the uid of the current context user is
	// used.
	BindNode(uid ...int64) error
	// UnbindNode unbinds the node. When uid is not specified, the uid of the current context user
	// is used.
	UnbindNode(uid ...int64) error
	// Subscribe subscribes to a channel. When uids are specified, it subscribes the user channels;
	// otherwise it subscribes the channel of the current connection.
	Subscribe(channel string, uids ...int64) error
	// Unsubscribe unsubscribes from a channel. When uids are specified, it unsubscribes the user
	// channels; otherwise it unsubscribes the channel of the current connection.
	Unsubscribe(channel string, uids ...int64) error
	// BindActor binds an actor, establishing a binding between the current user and the given
	// actor.
	BindActor(kind, id string) error
	// UnbindActor unbinds the actor of the given kind.
	UnbindActor(kind string)
	// Next dispatches the message down to the node.
	//
	// It hands the current message to the node's scheduler for a second dispatch. Calling it
	// automatically cancels every function in the defer call stack.
	Next() error
	// Spawn derives a new actor.
	//
	// creator creates the actor processor and opts configures the new actor. It reports an error
	// when an actor with the same kind and ID already exists or when creation fails.
	Spawn(creator Creator, opts ...ActorOption) (*Actor, error)
	// Kill kills an existing actor, reporting whether it was killed successfully.
	Kill(kind, id string) bool
	// Actor returns the actor with the given kind and ID, and whether it exists.
	Actor(kind, id string) (*Actor, bool)
	// Invoke calls a function in a thread-safe manner.
	//
	// Inside a global handler, it calls proxy.Invoke; inside an actor processor, it calls
	// actor.Invoke. The optional isBlock controls whether the call is blocking; it is non-blocking
	// by default.
	Invoke(fn func(), isBlock ...bool) error
	// AfterFunc schedules f to run after d, matching the semantics of [time.AfterFunc].
	//
	// Inside a global handler, it calls proxy.AfterFunc; inside an actor processor, it calls
	// actor.AfterFunc. The returned [Timer] can be cancelled with Timer.Stop.
	AfterFunc(d time.Duration, f func()) (*Timer, error)
	// AfterInvoke schedules f to run after d in a thread-safe manner.
	//
	// After the delay, f is executed serially through the task queue, which keeps it thread-safe.
	// Inside a global handler, it calls proxy.AfterInvoke; inside an actor processor, it calls
	// actor.AfterInvoke. The returned [Timer] can be cancelled with Timer.Stop.
	AfterInvoke(d time.Duration, f func()) (*Timer, error)
	// NewMeshClient creates a new mesh service client.
	//
	// The target can be given in three forms:
	//   - direct connection: direct://127.0.0.1:8011
	//   - direct connection: direct://711baf8d-8a06-11ef-b7df-f4f19e1f0070
	//   - service discovery: discovery://service_name
	NewMeshClient(target string) (transport.Client, error)
	// contains filtered or unexported methods
}

Context is a message context.

It wraps the basic information about the gateway, node, connection and user, and provides a series of methods such as route handling, message delivery and actor operations.

type Creator added in v2.2.0 

type Creator func(actor *Actor, args ...any) Processor

Creator is the function that creates an actor processor.

type EventHandler 

type EventHandler func(ctx Context)

EventHandler is an event handler function.

type HookHandler added in v2.1.0 

type HookHandler func(proxy *Proxy)

HookHandler is the node hook handler.

type Kind added in v2.2.0 

type Kind int

Kind is the message kind. 

const (
	Event   Kind = iota // Event: an event message
	Request             // Request: a request message
)

The kinds of a message.

type Middleware 

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

Middleware is a middleware.

It is used to intercept and process messages uniformly before and after the route handlers run.

func (*Middleware) Next 

func (m *Middleware) Next(ctx Context)

Next runs the next middleware.

func (*Middleware) Skip 

func (m *Middleware) Skip(ctx Context, skip int)

Skip skips the given number of middlewares.

It runs the following middlewares in order and, once they are exhausted, runs the final route handler.

type MiddlewareHandler 

type MiddlewareHandler func(middleware *Middleware, ctx Context)

MiddlewareHandler is the middleware handler.

type Node 

type Node struct {
	component.Base
	// contains filtered or unexported fields
}

Node is a node server.

func NewNode 

func NewNode(opts ...Option) *Node

NewNode creates a node server.

It initializes internal components such as the proxy, tasker, router, trigger and scheduler, and starts in the shut state.

func (*Node) Close added in v2.2.3 

func (n *Node) Close()

Close closes the node.

It sets the state to hang, stops accepting new messages and waits for all in-flight asynchronous tasks to finish. The messages already in the route and event queues are not waited for and the dispatcher keeps processing them until Destroy; any residual messages not drained by Destroy are discarded.

func (*Node) Destroy 

func (n *Node) Destroy()

Destroy destroys the node server.

It sets the state to shut, deregisters the service instances, stops the linker and transport servers and releases the resources of the internal components. Destroy is idempotent and repeated calls have no side effects.

func (*Node) Init 

func (n *Node) Init()

Init initializes the node.

It validates the required configuration such as the instance ID, instance name, codec, locator and registry, and terminates the process when any of them is missing.

func (*Node) Name 

func (n *Node) Name() string

Name returns the component name.

func (*Node) Proxy 

func (n *Node) Proxy() *Proxy

Proxy returns the node proxy.

func (*Node) Start 

func (n *Node) Start()

Start starts the node.

It sets the state to working, then starts the linker server and the transport server, registers the service instances and begins dispatching messages.

type Option 

type Option func(o *options)

Option is a configuration function of a Node.

func WithAddr added in v2.1.0 

func WithAddr(addr string) Option

WithAddr sets the listen address.

func WithCodec 

func WithCodec(codec encoding.Codec) Option

WithCodec sets the codec.

func WithContext 

func WithContext(ctx context.Context) Option

WithContext sets the startup context.

func WithContextFunc added in v2.6.0 

func WithContextFunc(ctxFunc func() context.Context) Option

WithContextFunc sets the custom context generator.

func WithEncryptor 

func WithEncryptor(encryptor crypto.Encryptor) Option

WithEncryptor sets the message encryptor.

func WithExpose added in v2.3.0 

func WithExpose(expose bool) Option

WithExpose sets whether the internal communication address is exposed to the public network.

func WithID 

func WithID(id string) Option

WithID sets the instance ID.

func WithLinkerCallTimeout added in v2.6.0 

func WithLinkerCallTimeout(callTimeout time.Duration) Option

WithLinkerCallTimeout sets the RPC call timeout.

func WithLinkerCommandQueueSize added in v2.6.0 

func WithLinkerCommandQueueSize(commandQueueSize int32) Option

WithLinkerCommandQueueSize sets the message queue capacity.

func WithLinkerCommandWriteTimeout added in v2.6.0 

func WithLinkerCommandWriteTimeout(commandWriteTimeout time.Duration) Option

WithLinkerCommandWriteTimeout sets the write timeout.

func WithLinkerConnNum added in v2.6.0 

func WithLinkerConnNum(connNum int) Option

WithLinkerConnNum sets the number of connections.

func WithLinkerDialRetryTimes added in v2.6.0 

func WithLinkerDialRetryTimes(dialRetryTimes int) Option

WithLinkerDialRetryTimes sets the internal RPC dial retry times.

func WithLinkerDialTimeout added in v2.6.0 

func WithLinkerDialTimeout(dialTimeout time.Duration) Option

WithLinkerDialTimeout sets the internal RPC dial timeout.

func WithLinkerFaultRecoveryTime added in v2.6.0 

func WithLinkerFaultRecoveryTime(faultRecoveryTime time.Duration) Option

WithLinkerFaultRecoveryTime sets the internal RPC fault recovery time.

func WithLocator 

func WithLocator(locator locate.Locator) Option

WithLocator sets the user locator.

func WithMessageQueueSize added in v2.6.0 

func WithMessageQueueSize(messageQueueSize int32) Option

WithMessageQueueSize sets the message queue capacity.

func WithMessageWriteTimeout added in v2.6.0 

func WithMessageWriteTimeout(messageWriteTimeout time.Duration) Option

WithMessageWriteTimeout sets the message write timeout.

func WithMetadata added in v2.3.0 

func WithMetadata(metadata map[string]string) Option

WithMetadata sets the metadata.

func WithName 

func WithName(name string) Option

WithName sets the instance name.

func WithRegistry 

func WithRegistry(r registry.Registry) Option

WithRegistry sets the service registry.

func WithTaskQueueSize added in v2.6.0 

func WithTaskQueueSize(taskQueueSize int32) Option

WithTaskQueueSize sets the task queue capacity.

func WithTaskWriteTimeout added in v2.6.0 

func WithTaskWriteTimeout(taskWriteTimeout time.Duration) Option

WithTaskWriteTimeout sets the task timeout.

func WithTransporter 

func WithTransporter(transporter transport.Transporter) Option

WithTransporter sets the message transporter.

func WithWeight added in v2.2.4 

func WithWeight(weight int) Option

WithWeight sets the weight.

type Processor added in v2.2.0 

type Processor interface {
	// Init is the init callback.
	Init()
	// Start is the start callback.
	Start()
	// Destroy is the destroy callback.
	Destroy()
}

Processor is the processor interface of an actor.

It defines the init, start and destroy callbacks of the actor lifecycle.

type Proxy 

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

Proxy is a proxy.

It provides the complete API of a node server, including gateway/node links, routing, events, message pushing and actor management.

func (*Proxy) Actor added in v2.2.0 

func (p *Proxy) Actor(kind, id string) (*Actor, bool)

Actor returns an actor and whether it exists.

func (*Proxy) AddEventHandler added in v2.1.0 

func (p *Proxy) AddEventHandler(event cluster.Event, handler EventHandler)

AddEventHandler adds an event handler.

func (*Proxy) AddHookListener added in v2.1.0 

func (p *Proxy) AddHookListener(hook cluster.Hook, handler HookHandler)

AddHookListener adds a hook listener.

func (*Proxy) AddRouteHandler added in v2.1.0 

func (p *Proxy) AddRouteHandler[T constraints.Integer](route T, handler RouteHandler, opts ...RouteOptions)

AddRouteHandler adds a route handler.

func (*Proxy) AddServiceProvider added in v2.2.0 

func (p *Proxy) AddServiceProvider(name string, desc, provider any)

AddServiceProvider adds a service provider described by desc.

func (*Proxy) AfterFunc added in v2.2.4 

func (p *Proxy) AfterFunc(d time.Duration, f func()) (*Timer, error)

AfterFunc schedules a delayed call and is used the same way as time.AfterFunc. It returns a timer that can be cancelled with Stop, or an error when the node is shut down.

func (*Proxy) AfterInvoke added in v2.2.4 

func (p *Proxy) AfterInvoke(d time.Duration, f func()) (*Timer, error)

AfterInvoke schedules a thread-safe delayed call.

The function is executed serially through the task queue after the delay, which guarantees thread safety. It returns a timer that can be cancelled with Stop, or an error when the node is shut down.

func (*Proxy) AskGate 

func (p *Proxy) AskGate(ctx context.Context, gid string, uid int64) (string, bool, error)

AskGate reports whether the user is on the given gateway. It returns the gateway ID the user actually resides on, whether the user is on the given gateway, and an error when the node is shut down.

func (*Proxy) AskNode 

func (p *Proxy) AskNode(ctx context.Context, uid int64, name, nid string) (string, bool, error)

AskNode reports whether the user is on the given node. It returns the node ID the user actually resides on, whether the user is on the given node, and an error when the node is shut down.

func (*Proxy) BindActor added in v2.2.0 

func (p *Proxy) BindActor(uid int64, kind, id string) error

BindActor binds an actor. It returns the error reported when the node is shut down or the binding fails.

func (*Proxy) BindGate 

func (p *Proxy) BindGate(ctx context.Context, gid string, cid, uid int64) error

BindGate binds the gateway. Once bound, the user is associated with the connection. It returns the error reported when the node is shut down or the binding fails.

func (*Proxy) BindNode 

func (p *Proxy) BindNode(ctx context.Context, uid int64, nameAndNID ...string) error

BindNode binds the node.

A user can be bound to multiple node servers, but only one node server per name; binding to a node server with the same name again overrides the previous binding. The binding is synchronized to the gateway servers and other related node servers through publish and subscribe. It returns the error reported when the node is shut down or the binding fails.

func (*Proxy) Broadcast 

func (p *Proxy) Broadcast(ctx context.Context, args *cluster.BroadcastArgs) (int64, error)

Broadcast pushes a broadcast message.

Set args.Ack to true to get the number of targets the push succeeded on. It returns that number and the error reported when the node is shut down or the push fails.

func (*Proxy) Deliver 

func (p *Proxy) Deliver(ctx context.Context, args *cluster.DeliverArgs) error

Deliver delivers a message to a node for handling. It returns the error reported when the node is shut down, the target is the current node or the delivery fails.

func (*Proxy) Disconnect 

func (p *Proxy) Disconnect(ctx context.Context, args *cluster.DisconnectArgs) error

Disconnect disconnects a connection. It returns the error reported when the node is shut down or the disconnection fails.

func (*Proxy) FetchGateList 

func (p *Proxy) FetchGateList(ctx context.Context, states ...cluster.State) ([]*registry.ServiceInstance, error)

FetchGateList fetches the gateway list filtered by states. It returns the gateway service instances and an error when the node is shut down.

func (*Proxy) FetchNodeList 

func (p *Proxy) FetchNodeList(ctx context.Context, states ...cluster.State) ([]*registry.ServiceInstance, error)

FetchNodeList fetches the node list filtered by states. It returns the node service instances and an error when the node is shut down.

func (*Proxy) GetID added in v2.1.0 

func (p *Proxy) GetID() string

GetID returns the current node ID.

func (*Proxy) GetIP 

func (p *Proxy) GetIP(ctx context.Context, args *cluster.GetIPArgs) (string, error)

GetIP returns the client IP. It returns the error reported when the node is shut down or the query fails.

func (*Proxy) GetName added in v2.1.0 

func (p *Proxy) GetName() string

GetName returns the current node name.

func (*Proxy) GetState added in v2.1.0 

func (p *Proxy) GetState() cluster.State

GetState returns the current node state.

func (*Proxy) HasGate added in v2.2.0 

func (p *Proxy) HasGate(gid string) (bool, error)

HasGate reports whether the given gateway exists. It returns an error when the node is shut down.

func (*Proxy) HasNode added in v2.2.0 

func (p *Proxy) HasNode(nid string) bool

HasNode reports whether the given node exists.

func (*Proxy) Invoke 

func (p *Proxy) Invoke(f func(), wait ...bool) error

Invoke calls a function in a thread-safe way.

A synchronous call (wait=true) issued from within the node dispatch goroutine executes the function directly, avoiding a deadlock while waiting for the queue it runs in. Do not wait synchronously across queues in a dispatch chain (for example, a node task synchronously waiting for an actor task to finish while the actor task is synchronously waiting for the node task); the mutual wait forms a cross-queue circular wait and deadlocks. It returns the error reported when the node is shut down or the task fails to be enqueued.

func (*Proxy) IsOnline added in v2.1.0 

func (p *Proxy) IsOnline(ctx context.Context, args *cluster.IsOnlineArgs) (bool, error)

IsOnline reports whether the target is online. It returns the error reported when the node is shut down or the query fails.

func (*Proxy) Kill added in v2.2.0 

func (p *Proxy) Kill(kind, id string) bool

Kill kills an existing actor. It reports whether the actor was killed successfully, and returns false when the node is shut down or the actor does not exist.

func (*Proxy) LocateGate 

func (p *Proxy) LocateGate(ctx context.Context, uid int64) (string, error)

LocateGate locates the gateway the user resides on. It returns the gateway ID and an error when the node is shut down.

func (*Proxy) LocateNode 

func (p *Proxy) LocateNode(ctx context.Context, uid int64, name string) (string, error)

LocateNode locates the node the user resides on. It returns the node ID and an error when the node is shut down.

func (*Proxy) LocateNodes added in v2.5.0 

func (p *Proxy) LocateNodes(ctx context.Context, uid int64) (map[string]string, error)

LocateNodes locates the nodes the user resides on. It returns a map from node name to node ID the user is bound to, and an error when the node is shut down.

func (*Proxy) Multicast 

func (p *Proxy) Multicast(ctx context.Context, args *cluster.MulticastArgs) (int64, error)

Multicast pushes a multicast message.

Set args.Ack to true to get the number of targets the push succeeded on. It returns that number and the error reported when the node is shut down or the push fails.

func (*Proxy) NewMeshClient added in v2.1.0 

func (p *Proxy) NewMeshClient(target string) (transport.Client, error)

NewMeshClient creates a new microservice client.

target supports three modes: 

service direct mode:    direct://127.0.0.1:8011
service direct mode:    direct://711baf8d-8a06-11ef-b7df-f4f19e1f0070
service discovery mode: discovery://service_name

It returns the error reported when the node is shut down or no message transporter is configured.

func (*Proxy) PackBuffer added in v2.2.0 

func (p *Proxy) PackBuffer(message any) ([]byte, error)

PackBuffer packs the given message content. It returns the packed bytes and the error reported when packing fails.

func (*Proxy) PackMessage added in v2.2.0 

func (p *Proxy) PackMessage(message *cluster.Message) ([]byte, error)

PackMessage packs a message. It returns the packed bytes and the error reported when packing fails.

func (*Proxy) Publish added in v2.3.0 

func (p *Proxy) Publish(ctx context.Context, args *cluster.PublishArgs) (int64, error)

Publish publishes a message.

Set args.Ack to true to get the number of targets the publish succeeded on. It returns that number and the error reported when the node is shut down or the publish fails.

func (*Proxy) Push 

func (p *Proxy) Push(ctx context.Context, args *cluster.PushArgs) error

Push pushes a message.

Set args.Ack to true to get the actual sending result of the message. It returns the error reported when the node is shut down or the push fails.

func (*Proxy) RouteGroup added in v2.1.0 

func (p *Proxy) RouteGroup(groups ...func(group *RouterGroup)) *RouterGroup

RouteGroup returns a route group configured by the given functions.

func (*Proxy) Router 

func (p *Proxy) Router() *Router

Router returns the router.

func (*Proxy) SetDefaultRouteHandler added in v2.1.0 

func (p *Proxy) SetDefaultRouteHandler(handler RouteHandler)

SetDefaultRouteHandler sets the default route handler. Every unregistered route goes through the default route handler.

func (*Proxy) SetState added in v2.1.0 

func (p *Proxy) SetState(state cluster.State) error

SetState sets the current node state. It returns the error reported while setting the state.

func (*Proxy) Spawn added in v2.2.0 

func (p *Proxy) Spawn(creator Creator, opts ...ActorOption) (*Actor, error)

Spawn creates a new actor. It returns the error reported when the node is shut down or the actor fails to be created.

func (*Proxy) Stat 

func (p *Proxy) Stat(ctx context.Context, kind session.Kind) (int64, error)

Stat counts the sessions of the given kind. It returns the session count and an error when the node is shut down.

func (*Proxy) Subscribe added in v2.3.0 

func (p *Proxy) Subscribe(ctx context.Context, args *cluster.SubscribeArgs) error

Subscribe subscribes to a channel. It returns the error reported when the node is shut down or the subscription fails.

func (*Proxy) Trigger added in v2.1.0 

func (p *Proxy) Trigger() *Trigger

Trigger returns the event trigger.

func (*Proxy) UnbindActor added in v2.2.0 

func (p *Proxy) UnbindActor(uid int64, kind string) error

UnbindActor unbinds an actor. It returns the error reported when the node is shut down or the unbinding fails.

func (*Proxy) UnbindGate 

func (p *Proxy) UnbindGate(ctx context.Context, uid int64) error

UnbindGate unbinds the gateway. It returns the error reported when the node is shut down or the unbinding fails.

func (*Proxy) UnbindNode 

func (p *Proxy) UnbindNode(ctx context.Context, uid int64, nameAndNID ...string) error

UnbindNode unbinds the node.

It unbinds the node server with the corresponding name and verifies the node ID, so the unbinding fails on a mismatch. The unbinding is synchronized to the gateway servers and other related node servers through publish and subscribe. It returns the error reported when the node is shut down or the unbinding fails.

func (*Proxy) Unsubscribe added in v2.3.0 

func (p *Proxy) Unsubscribe(ctx context.Context, args *cluster.UnsubscribeArgs) error

Unsubscribe unsubscribes from a channel. It returns the error reported when the node is shut down or the unsubscription fails.

type RouteHandler 

type RouteHandler func(ctx Context)

RouteHandler is a route handler function.

type RouteOptions added in v2.1.0 

type RouteOptions struct {
	// Internal reports whether the route is internal, and it is not internal by default.
	//
	// External routes can carry messages across clients, gateways and nodes, while internal routes
	// can only carry messages between nodes.
	Internal bool

	// Stateful reports whether the route is stateful, and it is not stateful by default.
	//
	// Messages of a stateless route are dispatched to different node servers by the load balancing
	// strategy, while messages of a stateful route are always routed to the node server the user is
	// bound to.
	Stateful bool

	// Authorized reports whether the route is authorized, and it is not authorized by default.
	//
	// An authorized route must carry the UID when it flows across the cluster, otherwise it cannot
	// be delivered. The gateway can use this option to intercept unauthorized connections early and
	// reduce the attack handling pressure on node servers.
	Authorized bool

	// Middlewares are the route middlewares.
	Middlewares []MiddlewareHandler
}

RouteOptions are route options.

type Router 

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

Router is a router.

It carries the routing queue of a node and is responsible for receiving and handling routed messages.

func (*Router) AddRouteHandler 

func (r *Router) AddRouteHandler[T constraints.Integer](route T, handler RouteHandler, opts ...RouteOptions)

AddRouteHandler adds a route handler.

func (*Router) CheckRouteStateful 

func (r *Router) CheckRouteStateful(route int32) (stateful bool, exist bool)

CheckRouteStateful reports whether the route is stateful and whether the route exists.

func (*Router) Group 

func (r *Router) Group(groups ...func(group *RouterGroup)) *RouterGroup

Group groups routes into a RouterGroup configured by the given functions.

func (*Router) HasDefaultRouteHandler 

func (r *Router) HasDefaultRouteHandler() bool

HasDefaultRouteHandler reports whether a default route handler exists.

func (*Router) SetDefaultRouteHandler 

func (r *Router) SetDefaultRouteHandler(handler RouteHandler)

SetDefaultRouteHandler sets the default route handler. Every unregistered route goes through the default route handler.

func (*Router) SetPostRouteHandler added in v2.2.6 

func (r *Router) SetPostRouteHandler(handler RouteHandler)

SetPostRouteHandler sets the post-route handler.

func (*Router) SetPreRouteHandler added in v2.2.6 

func (r *Router) SetPreRouteHandler(handler RouteHandler)

SetPreRouteHandler sets the pre-route handler.

type RouterGroup 

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

RouterGroup is a route group.

It manages a set of routes and their middlewares together.

func (*RouterGroup) AddRouteHandler 

func (g *RouterGroup) AddRouteHandler(route int32, handler RouteHandler, opts ...RouteOptions) *RouterGroup

AddRouteHandler adds a route handler. It merges the route group middlewares with the middlewares of the route itself.

func (*RouterGroup) Middleware 

func (g *RouterGroup) Middleware(middlewares ...MiddlewareHandler) *RouterGroup

Middleware adds middlewares to the route group.

type Scheduler added in v2.2.0 

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

Scheduler is a scheduler.

It is responsible for creating and destroying actors as well as maintaining the relations between users and actors and dispatching messages.

type Timer added in v2.2.4 

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

Timer is a timer.

func (*Timer) Stop added in v2.2.4 

func (t *Timer) Stop() (ok bool)

Stop stops the timer.

When the timer is stopped successfully and the node is not nil, it decrements the node's wait count.

type Trigger added in v2.1.0 

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

Trigger is an event trigger.

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.