Documentation ¶
Index ¶
- Constants
- Variables
- func CalculateWeight(rtt int) int
- func Hash(key uint64, buckets int32) int32
- func HashString(s string) uint64
- func JumpConsistentHash(len int, options ...interface{}) int
- func Ping(host string) (rtt int, err error)
- func RegisterCacheClientBuilder(network string, builder CacheClientBuilder)
- type Breaker
- type CacheClientBuilder
- type Call
- type Client
- func (client *Client) Call(ctx context.Context, servicePath, serviceMethod string, args interface{}, ...) error
- func (client *Client) Close() error
- func (client *Client) Connect(network, address string) error
- func (client *Client) GetConn() net.Conn
- func (client *Client) Go(ctx context.Context, servicePath, serviceMethod string, args interface{}, ...) *Call
- func (client *Client) IsClosing() bool
- func (client *Client) IsShutdown() bool
- func (client *Client) RegisterServerMessageChan(ch chan<- *protocol.Message)
- func (client *Client) RemoteAddr() string
- func (client *Client) SendRaw(ctx context.Context, r *protocol.Message) (map[string]string, []byte, error)
- func (client *Client) UnregisterServerMessageChan()
- type ClientAfterDecodePlugin
- type ClientBeforeEncodePlugin
- type ClientConnectedPlugin
- type ClientConnectionClosePlugin
- type ConnCreateFailedPlugin
- type ConnCreatedPlugin
- type ConnFactoryFn
- type ConsecCircuitBreaker
- type ConsistentAddrStrFunction
- type DNSDiscovery
- func (d *DNSDiscovery) Clone(servicePath string) (ServiceDiscovery, error)
- func (d *DNSDiscovery) Close()
- func (d *DNSDiscovery) GetServices() []*KVPair
- func (d *DNSDiscovery) RemoveWatcher(ch chan []*KVPair)
- func (d *DNSDiscovery) SetFilter(filter ServiceDiscoveryFilter)
- func (d *DNSDiscovery) WatchService() chan []*KVPair
- type FailMode
- type HashServiceAndArgs
- type KVPair
- type MDNSDiscovery
- func (d *MDNSDiscovery) Clone(servicePath string) (ServiceDiscovery, error)
- func (d *MDNSDiscovery) Close()
- func (d *MDNSDiscovery) GetServices() []*KVPair
- func (d *MDNSDiscovery) RemoveWatcher(ch chan []*KVPair)
- func (d *MDNSDiscovery) SetFilter(filter ServiceDiscoveryFilter)
- func (d *MDNSDiscovery) WatchService() chan []*KVPair
- type MultipleServersDiscovery
- func (d *MultipleServersDiscovery) Clone(servicePath string) (ServiceDiscovery, error)
- func (d *MultipleServersDiscovery) Close()
- func (d *MultipleServersDiscovery) GetServices() []*KVPair
- func (d *MultipleServersDiscovery) RemoveWatcher(ch chan []*KVPair)
- func (d *MultipleServersDiscovery) SetFilter(filter ServiceDiscoveryFilter)
- func (d *MultipleServersDiscovery) Update(pairs []*KVPair)
- func (d *MultipleServersDiscovery) WatchService() chan []*KVPair
- type OneClient
- func (c *OneClient) Auth(auth string)
- func (c *OneClient) Broadcast(ctx context.Context, servicePath string, serviceMethod string, ...) error
- func (c *OneClient) Call(ctx context.Context, servicePath string, serviceMethod string, ...) error
- func (c *OneClient) Close() error
- func (c *OneClient) ConfigGeoSelector(latitude, longitude float64)
- func (c *OneClient) DownloadFile(ctx context.Context, requestFileName string, saveTo io.Writer, ...) error
- func (c *OneClient) Fork(ctx context.Context, servicePath string, serviceMethod string, ...) error
- func (c *OneClient) GetPlugins() PluginContainer
- func (c *OneClient) Go(ctx context.Context, servicePath string, serviceMethod string, ...) (*Call, error)
- func (c *OneClient) SendFile(ctx context.Context, fileName string, rateInBytesPerSecond int64, ...) error
- func (c *OneClient) SendRaw(ctx context.Context, r *protocol.Message) (map[string]string, []byte, error)
- func (c *OneClient) SetPlugins(plugins PluginContainer)
- func (c *OneClient) SetSelector(servicePath string, s Selector)
- func (c *OneClient) Stream(ctx context.Context, meta map[string]string) (net.Conn, error)
- type OneClientPool
- type Option
- type Peer2PeerDiscovery
- func (d *Peer2PeerDiscovery) Clone(servicePath string) (ServiceDiscovery, error)
- func (d *Peer2PeerDiscovery) Close()
- func (d *Peer2PeerDiscovery) GetServices() []*KVPair
- func (d *Peer2PeerDiscovery) RemoveWatcher(ch chan []*KVPair)
- func (d *Peer2PeerDiscovery) SetFilter(filter ServiceDiscoveryFilter)
- func (d *Peer2PeerDiscovery) WatchService() chan []*KVPair
- type Plugin
- type PluginContainer
- type PostCallPlugin
- type PreCallPlugin
- type RPCClient
- type Receipt
- type SelectFunc
- type SelectMode
- type SelectNodePlugin
- type Selector
- type ServiceDiscovery
- type ServiceDiscoveryFilter
- type ServiceError
- type Weighted
- type XClient
- type XClientPool
Constants ¶
const ( XVersion = "X-RPCX-Version" XMessageType = "X-RPCX-MesssageType" XHeartbeat = "X-RPCX-Heartbeat" XOneway = "X-RPCX-Oneway" XMessageStatusType = "X-RPCX-MessageStatusType" XSerializeType = "X-RPCX-SerializeType" XMessageID = "X-RPCX-MessageID" XServicePath = "X-RPCX-ServicePath" XServiceMethod = "X-RPCX-ServiceMethod" XMeta = "X-RPCX-Meta" XErrorMessage = "X-RPCX-ErrorMessage" )
const ( // ReaderBuffsize is used for bufio reader. ReaderBuffsize = 16 * 1024 // WriterBuffsize is used for bufio writer. WriterBuffsize = 16 * 1024 )
const (
FileTransferBufferSize = 1024
)
Variables ¶
var ( ErrBreakerOpen = errors.New("breaker open") ErrBreakerTimeout = errors.New("breaker time out") )
var ( ErrShutdown = errors.New("connection is shut down") ErrUnsupportedCodec = errors.New("unsupported codec") )
ErrShutdown connection is closed.
var ( // ErrXClientShutdown xclient is shutdown. ErrXClientShutdown = errors.New("xClient is shut down") // ErrXClientNoServer selector can't found one server. ErrXClientNoServer = errors.New("can not found any server") ErrServerUnavailable = errors.New("selected server is unavailable") )
var ClientErrorFunc func(res *protocol.Message, e string) ServiceError
ClientErrorFunc is a function to create a customized error.
var ConnFactories = map[string]ConnFactoryFn{
"http": newDirectHTTPConn,
"kcp": newDirectKCPConn,
"quic": newDirectQuicConn,
"unix": newDirectConn,
"memu": newMemuConn,
"iouring": newIOUringConn,
}
var DefaultOption = Option{ Retries: 3, RPCPath: share.DefaultRPCPath, ConnectTimeout: time.Second, SerializeType: protocol.MsgPack, CompressType: protocol.None, BackupLatency: 10 * time.Millisecond, MaxWaitForHeartbeat: 30 * time.Second, TCPKeepAlivePeriod: time.Minute, BidirectionalBlock: false, TimeToDisallow: time.Minute, }
DefaultOption is a common option configuration for client.
Functions ¶
func CalculateWeight ¶ added in v1.6.2
CalculateWeight converts the rtt to weighted by:
- weight=191 if t <= 10
- weight=201 -t if 10 < t <=200
- weight=1 if 200 < t < 1000
- weight = 0 if t >= 1000
It means servers that ping time t < 10 will be preferred and servers won't be selected if t > 1000. It is hard coded based on Ops experience.
func Hash ¶
Hash consistently chooses a hash bucket number in the range [0, numBuckets) for the given key. numBuckets must be >= 1.
func JumpConsistentHash ¶
JumpConsistentHash selects a server by serviceMethod and args
func RegisterCacheClientBuilder ¶ added in v1.6.3
func RegisterCacheClientBuilder(network string, builder CacheClientBuilder)
Types ¶
type Breaker ¶
Breaker is a CircuitBreaker interface.
var CircuitBreaker Breaker = circuit.NewRateBreaker(0.95, 100)
CircuitBreaker is a default circuit breaker (RateBreaker(0.95, 100)).
type CacheClientBuilder ¶ added in v1.6.3
type CacheClientBuilder interface { SetCachedClient(client RPCClient, k, servicePath, serviceMethod string) FindCachedClient(k, servicePath, serviceMethod string) RPCClient DeleteCachedClient(client RPCClient, k, servicePath, serviceMethod string) GenerateClient(k, servicePath, serviceMethod string) (client RPCClient, err error) }
CacheClientBuilder defines builder interface to generate RPCCient.
type Call ¶
type Call struct { ServicePath string // The name of the service and method to call. ServiceMethod string // The name of the service and method to call. Metadata map[string]string // metadata ResMetadata map[string]string Args interface{} // The argument to the function (*struct). Reply interface{} // The reply from the function (*struct). Error error // After completion, the error status. Done chan *Call // Strobes when call is complete. Raw bool // raw message or not }
Call represents an active RPC.
type Client ¶
type Client struct { Conn net.Conn Plugins PluginContainer ServerMessageChan chan<- *protocol.Message // contains filtered or unexported fields }
Client represents a RPC client.
func (*Client) Call ¶
func (client *Client) Call(ctx context.Context, servicePath, serviceMethod string, args interface{}, reply interface{}) error
Call invokes the named function, waits for it to complete, and returns its error status.
func (*Client) Close ¶
Close calls the underlying connection's Close method. If the connection is already shutting down, ErrShutdown is returned.
func (*Client) Go ¶
func (client *Client) Go(ctx context.Context, servicePath, serviceMethod string, args interface{}, reply interface{}, done chan *Call) *Call
Go invokes the function asynchronously. It returns the Call structure representing the invocation. The done channel will signal when the call is complete by returning the same Call object. If done is nil, Go will allocate a new channel. If non-nil, done must be buffered or Go will deliberately crash.
func (*Client) IsShutdown ¶
IsShutdown client is shutdown or not.
func (*Client) RegisterServerMessageChan ¶
RegisterServerMessageChan registers the channel that receives server requests.
func (*Client) RemoteAddr ¶ added in v1.6.2
RemoteAddr returns the remote address.
func (*Client) SendRaw ¶
func (client *Client) SendRaw(ctx context.Context, r *protocol.Message) (map[string]string, []byte, error)
SendRaw sends raw messages. You don't care args and replies.
func (*Client) UnregisterServerMessageChan ¶
func (client *Client) UnregisterServerMessageChan()
UnregisterServerMessageChan removes ServerMessageChan.
type ClientAfterDecodePlugin ¶ added in v1.4.1
ClientAfterDecodePlugin is invoked when the message is decoded.
type ClientBeforeEncodePlugin ¶ added in v1.4.1
ClientBeforeEncodePlugin is invoked when the message is encoded and sent.
type ClientConnectedPlugin ¶ added in v1.4.1
ClientConnectedPlugin is invoked when the client has connected the server.
type ClientConnectionClosePlugin ¶ added in v1.4.1
ClientConnectionClosePlugin is invoked when the connection is closing.
type ConnCreateFailedPlugin ¶ added in v1.7.4
type ConnCreateFailedPlugin interface {
ConnCreateFailed(network, address string)
}
type ConnCreatedPlugin ¶ added in v1.6.2
ConnCreatedPlugin is invoked when the client connection has created.
type ConnFactoryFn ¶ added in v1.6.2
type ConsecCircuitBreaker ¶
type ConsecCircuitBreaker struct {
// contains filtered or unexported fields
}
ConsecCircuitBreaker is window sliding CircuitBreaker with failure threshold.
func NewConsecCircuitBreaker ¶
func NewConsecCircuitBreaker(failureThreshold uint64, window time.Duration) *ConsecCircuitBreaker
NewConsecCircuitBreaker returns a new ConsecCircuitBreaker.
func (*ConsecCircuitBreaker) Call ¶
func (cb *ConsecCircuitBreaker) Call(fn func() error, d time.Duration) error
Call Circuit function
func (*ConsecCircuitBreaker) Fail ¶ added in v1.4.1
func (cb *ConsecCircuitBreaker) Fail()
func (*ConsecCircuitBreaker) Ready ¶ added in v1.4.1
func (cb *ConsecCircuitBreaker) Ready() bool
func (*ConsecCircuitBreaker) Success ¶ added in v1.4.1
func (cb *ConsecCircuitBreaker) Success()
type ConsistentAddrStrFunction ¶
type ConsistentAddrStrFunction func(options ...interface{}) string
ConsistentFunction define a hash function Return service address, like "tcp@127.0.0.1:8970"
type DNSDiscovery ¶ added in v1.6.2
type DNSDiscovery struct {
// contains filtered or unexported fields
}
DNSDiscovery is based on DNS a record. You must set port and network info when you create the DNSDiscovery.
func NewDNSDiscovery ¶ added in v1.6.2
func NewDNSDiscovery(domain string, network string, port int, d time.Duration) (*DNSDiscovery, error)
NewDNSDiscovery returns a new DNSDiscovery.
func (*DNSDiscovery) Clone ¶ added in v1.6.2
func (d *DNSDiscovery) Clone(servicePath string) (ServiceDiscovery, error)
Clone clones this ServiceDiscovery with new servicePath.
func (*DNSDiscovery) Close ¶ added in v1.6.2
func (d *DNSDiscovery) Close()
func (*DNSDiscovery) GetServices ¶ added in v1.6.2
func (d *DNSDiscovery) GetServices() []*KVPair
GetServices returns the static server
func (*DNSDiscovery) RemoveWatcher ¶ added in v1.6.2
func (d *DNSDiscovery) RemoveWatcher(ch chan []*KVPair)
func (*DNSDiscovery) SetFilter ¶ added in v1.6.2
func (d *DNSDiscovery) SetFilter(filter ServiceDiscoveryFilter)
SetFilter sets the filer.
func (*DNSDiscovery) WatchService ¶ added in v1.6.2
func (d *DNSDiscovery) WatchService() chan []*KVPair
WatchService returns a nil chan.
type FailMode ¶
type FailMode int
FailMode decides how clients action when clients fail to invoke services
const ( // Failover selects another server automaticaly Failover FailMode = iota // Failfast returns error immediately Failfast // Failtry use current client again Failtry // Failbackup select another server if the first server doesn't respond in specified time and use the fast response. Failbackup )
func FailModeString ¶ added in v1.4.1
FailModeString retrieves an enum value from the enum constants string name. Throws an error if the param is not part of the enum.
func FailModeValues ¶ added in v1.4.1
func FailModeValues() []FailMode
FailModeValues returns all values of the enum
func (FailMode) IsAFailMode ¶ added in v1.4.1
IsAFailMode returns "true" if the value is listed in the enum definition. "false" otherwise
type HashServiceAndArgs ¶
HashServiceAndArgs define a hash function
type MDNSDiscovery ¶
type MDNSDiscovery struct { Timeout time.Duration WatchInterval time.Duration // contains filtered or unexported fields }
MDNSDiscovery is a mdns service discovery. It always returns the registered servers in mdns.
func NewMDNSDiscovery ¶
func NewMDNSDiscovery(service string, timeout time.Duration, watchInterval time.Duration, domain string) (*MDNSDiscovery, error)
NewMDNSDiscovery returns a new MDNSDiscovery. If domain is empty, use "local." in default.
func (*MDNSDiscovery) Clone ¶
func (d *MDNSDiscovery) Clone(servicePath string) (ServiceDiscovery, error)
Clone clones this ServiceDiscovery with new servicePath.
func (*MDNSDiscovery) Close ¶
func (d *MDNSDiscovery) Close()
func (*MDNSDiscovery) GetServices ¶
func (d *MDNSDiscovery) GetServices() []*KVPair
GetServices returns the servers
func (*MDNSDiscovery) RemoveWatcher ¶
func (d *MDNSDiscovery) RemoveWatcher(ch chan []*KVPair)
func (*MDNSDiscovery) SetFilter ¶ added in v1.6.2
func (d *MDNSDiscovery) SetFilter(filter ServiceDiscoveryFilter)
SetFilter sets the filer.
func (*MDNSDiscovery) WatchService ¶
func (d *MDNSDiscovery) WatchService() chan []*KVPair
WatchService returns a nil chan.
type MultipleServersDiscovery ¶
type MultipleServersDiscovery struct {
// contains filtered or unexported fields
}
MultipleServersDiscovery is a multiple servers service discovery. It always returns the current servers and users can change servers dynamically.
func NewMultipleServersDiscovery ¶
func NewMultipleServersDiscovery(pairs []*KVPair) (*MultipleServersDiscovery, error)
NewMultipleServersDiscovery returns a new MultipleServersDiscovery.
func (*MultipleServersDiscovery) Clone ¶
func (d *MultipleServersDiscovery) Clone(servicePath string) (ServiceDiscovery, error)
Clone clones this ServiceDiscovery with new servicePath.
func (*MultipleServersDiscovery) Close ¶
func (d *MultipleServersDiscovery) Close()
func (*MultipleServersDiscovery) GetServices ¶
func (d *MultipleServersDiscovery) GetServices() []*KVPair
GetServices returns the configured server
func (*MultipleServersDiscovery) RemoveWatcher ¶
func (d *MultipleServersDiscovery) RemoveWatcher(ch chan []*KVPair)
func (*MultipleServersDiscovery) SetFilter ¶ added in v1.6.2
func (d *MultipleServersDiscovery) SetFilter(filter ServiceDiscoveryFilter)
SetFilter sets the filer.
func (*MultipleServersDiscovery) Update ¶
func (d *MultipleServersDiscovery) Update(pairs []*KVPair)
Update is used to update servers at runtime.
func (*MultipleServersDiscovery) WatchService ¶
func (d *MultipleServersDiscovery) WatchService() chan []*KVPair
WatchService returns a nil chan.
type OneClient ¶ added in v1.4.1
type OneClient struct { Plugins PluginContainer // contains filtered or unexported fields }
OneClient wraps servicesPath and XClients. Users can use a shared oneclient to access multiple services.
func NewBidirectionalOneClient ¶ added in v1.4.1
func NewBidirectionalOneClient(failMode FailMode, selectMode SelectMode, discovery ServiceDiscovery, option Option, serverMessageChan chan<- *protocol.Message) *OneClient
NewBidirectionalOneClient creates a new xclient that can receive notifications from servers.
func NewOneClient ¶ added in v1.4.1
func NewOneClient(failMode FailMode, selectMode SelectMode, discovery ServiceDiscovery, option Option) *OneClient
NewOneClient creates a OneClient that supports service discovery and service governance.
func (*OneClient) Broadcast ¶ added in v1.4.1
func (c *OneClient) Broadcast(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) error
Broadcast sends requests to all servers and Success only when all servers return OK. FailMode and SelectMode are meanless for this method. Please set timeout to avoid hanging.
func (*OneClient) Call ¶ added in v1.4.1
func (c *OneClient) Call(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) error
Call invokes the named function, waits for it to complete, and returns its error status. It handles errors base on FailMode.
func (*OneClient) Close ¶ added in v1.4.1
Close closes all xclients and its underlying connections to services.
func (*OneClient) ConfigGeoSelector ¶ added in v1.4.1
ConfigGeoSelector sets location of client's latitude and longitude, and use newGeoSelector.
func (*OneClient) DownloadFile ¶ added in v1.4.1
func (*OneClient) Fork ¶ added in v1.4.1
func (c *OneClient) Fork(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) error
Fork sends requests to all servers and Success once one server returns OK. FailMode and SelectMode are meanless for this method.
func (*OneClient) GetPlugins ¶ added in v1.6.2
func (c *OneClient) GetPlugins() PluginContainer
func (*OneClient) Go ¶ added in v1.4.1
func (c *OneClient) Go(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}, done chan *Call) (*Call, error)
Go invokes the function asynchronously. It returns the Call structure representing the invocation. The done channel will signal when the call is complete by returning the same Call object. If done is nil, Go will allocate a new channel. If non-nil, done must be buffered or Go will deliberately crash. It does not use FailMode.
func (*OneClient) SetPlugins ¶ added in v1.4.1
func (c *OneClient) SetPlugins(plugins PluginContainer)
SetPlugins sets client's plugins.
func (*OneClient) SetSelector ¶ added in v1.4.1
SetSelector sets customized selector by users.
type OneClientPool ¶ added in v1.4.1
type OneClientPool struct {
// contains filtered or unexported fields
}
OneClientPool is a oneclient pool with fixed size. It uses roundrobin algorithm to call its xclients. All oneclients share the same configurations such as ServiceDiscovery and serverMessageChan.
func NewBidirectionalOneClientPool ¶ added in v1.4.1
func NewBidirectionalOneClientPool(count int, failMode FailMode, selectMode SelectMode, discovery ServiceDiscovery, option Option, serverMessageChan chan<- *protocol.Message) *OneClientPool
NewBidirectionalOneClientPool creates a BidirectionalOneClient pool with fixed size.
func NewOneClientPool ¶ added in v1.4.1
func NewOneClientPool(count int, failMode FailMode, selectMode SelectMode, discovery ServiceDiscovery, option Option) *OneClientPool
NewOneClientPool creates a fixed size OneClient pool.
func (*OneClientPool) Auth ¶ added in v1.6.2
func (p *OneClientPool) Auth(auth string)
Auth sets s token for Authentication.
func (*OneClientPool) Close ¶ added in v1.4.1
func (p *OneClientPool) Close()
Close this pool. Please make sure it won't be used any more.
func (*OneClientPool) Get ¶ added in v1.4.1
func (p *OneClientPool) Get() *OneClient
Get returns a OneClient. It does not remove this OneClient from its cache so you don't need to put it back. Don't close this OneClient because maybe other goroutines are using this OneClient.
type Option ¶
type Option struct { // Group is used to select the services in the same group. Services set group info in their meta. // If it is empty, clients will ignore group. Group string // Retries retries to send Retries int // Time to disallow the bad server not to be selected TimeToDisallow time.Duration // TLSConfig for tcp and quic TLSConfig *tls.Config // kcp.BlockCrypt Block interface{} // RPCPath for http connection RPCPath string // ConnectTimeout sets timeout for dialing ConnectTimeout time.Duration // IdleTimeout sets max idle time for underlying net.Conns IdleTimeout time.Duration // BackupLatency is used for Failbackup mode. rpcx will sends another request if the first response doesn't return in BackupLatency time. BackupLatency time.Duration // Breaker is used to config CircuitBreaker GenBreaker func() Breaker SerializeType protocol.SerializeType CompressType protocol.CompressType // send heartbeat message to service and check responses Heartbeat bool // interval for heartbeat HeartbeatInterval time.Duration MaxWaitForHeartbeat time.Duration // TCPKeepAlive, if it is zero we don't set keepalive TCPKeepAlivePeriod time.Duration // bidirectional mode, if true serverMessageChan will block to wait message for consume. default false. BidirectionalBlock bool // alaways use the selected server until it is bad Sticky bool // not call server message handler NilCallServerMessageHandler func(msg *protocol.Message) }
Option contains all options for creating clients.
type Peer2PeerDiscovery ¶
type Peer2PeerDiscovery struct {
// contains filtered or unexported fields
}
Peer2PeerDiscovery is a peer-to-peer service discovery. It always returns the static server.
func NewPeer2PeerDiscovery ¶
func NewPeer2PeerDiscovery(server, metadata string) (*Peer2PeerDiscovery, error)
NewPeer2PeerDiscovery returns a new Peer2PeerDiscovery.
func (*Peer2PeerDiscovery) Clone ¶
func (d *Peer2PeerDiscovery) Clone(servicePath string) (ServiceDiscovery, error)
Clone clones this ServiceDiscovery with new servicePath.
func (*Peer2PeerDiscovery) Close ¶
func (d *Peer2PeerDiscovery) Close()
func (*Peer2PeerDiscovery) GetServices ¶
func (d *Peer2PeerDiscovery) GetServices() []*KVPair
GetServices returns the static server
func (*Peer2PeerDiscovery) RemoveWatcher ¶
func (d *Peer2PeerDiscovery) RemoveWatcher(ch chan []*KVPair)
func (*Peer2PeerDiscovery) SetFilter ¶ added in v1.6.2
func (d *Peer2PeerDiscovery) SetFilter(filter ServiceDiscoveryFilter)
SetFilter sets the filer.
func (*Peer2PeerDiscovery) WatchService ¶
func (d *Peer2PeerDiscovery) WatchService() chan []*KVPair
WatchService returns a nil chan.
type PluginContainer ¶
type PluginContainer interface { Add(plugin Plugin) Remove(plugin Plugin) All() []Plugin DoConnCreated(net.Conn) (net.Conn, error) DoConnCreateFailed(network, address string) DoClientConnected(net.Conn) (net.Conn, error) DoClientConnectionClose(net.Conn) error DoPreCall(ctx context.Context, servicePath, serviceMethod string, args interface{}) error DoPostCall(ctx context.Context, servicePath, serviceMethod string, args interface{}, reply interface{}, err error) error DoClientBeforeEncode(*protocol.Message) error DoClientAfterDecode(*protocol.Message) error DoWrapSelect(SelectFunc) SelectFunc }
PluginContainer represents a plugin container that defines all methods to manage plugins. And it also defines all extension points.
func NewPluginContainer ¶ added in v1.4.1
func NewPluginContainer() PluginContainer
type PostCallPlugin ¶
type PostCallPlugin interface {
PostCall(ctx context.Context, servicePath, serviceMethod string, args interface{}, reply interface{}, err error) error
}
PostCallPlugin is invoked after the client calls a server.
type PreCallPlugin ¶
type PreCallPlugin interface {
PreCall(ctx context.Context, servicePath, serviceMethod string, args interface{}) error
}
PreCallPlugin is invoked before the client calls a server.
type RPCClient ¶
type RPCClient interface { Connect(network, address string) error Go(ctx context.Context, servicePath, serviceMethod string, args interface{}, reply interface{}, done chan *Call) *Call Call(ctx context.Context, servicePath, serviceMethod string, args interface{}, reply interface{}) error SendRaw(ctx context.Context, r *protocol.Message) (map[string]string, []byte, error) Close() error RemoteAddr() string RegisterServerMessageChan(ch chan<- *protocol.Message) UnregisterServerMessageChan() IsClosing() bool IsShutdown() bool GetConn() net.Conn }
RPCClient is interface that defines one client to call one server.
type SelectFunc ¶ added in v1.6.2
type SelectMode ¶
type SelectMode int
SelectMode defines the algorithm of selecting a services from candidates.
const ( // RandomSelect is selecting randomly RandomSelect SelectMode = iota // RoundRobin is selecting by round robin RoundRobin // WeightedRoundRobin is selecting by weighted round robin WeightedRoundRobin // WeightedICMP is selecting by weighted Ping time WeightedICMP // ConsistentHash is selecting by hashing ConsistentHash // Closest is selecting the closest server Closest // SelectByUser is selecting by implementation of users SelectByUser = 1000 )
func SelectModeString ¶ added in v1.4.1
func SelectModeString(s string) (SelectMode, error)
SelectModeString retrieves an enum value from the enum constants string name. Throws an error if the param is not part of the enum.
func SelectModeValues ¶ added in v1.4.1
func SelectModeValues() []SelectMode
SelectModeValues returns all values of the enum
func (SelectMode) IsASelectMode ¶ added in v1.4.1
func (i SelectMode) IsASelectMode() bool
IsASelectMode returns "true" if the value is listed in the enum definition. "false" otherwise
func (SelectMode) String ¶
func (i SelectMode) String() string
type SelectNodePlugin ¶ added in v1.6.2
type SelectNodePlugin interface {
WrapSelect(SelectFunc) SelectFunc
}
SelectNodePlugin can interrupt selecting of xclient and add customized logics such as skipping some nodes.
type Selector ¶
type Selector interface { Select(ctx context.Context, servicePath, serviceMethod string, args interface{}) string // SelectFunc UpdateServer(servers map[string]string) }
Selector defines selector that selects one service from candidates.
type ServiceDiscovery ¶
type ServiceDiscovery interface { GetServices() []*KVPair // return all services in the registry WatchService() chan []*KVPair // watch the change of services, it's a golang channel RemoveWatcher(ch chan []*KVPair) Clone(servicePath string) (ServiceDiscovery, error) SetFilter(ServiceDiscoveryFilter) // set customized filter to filter services Close() }
ServiceDiscovery defines ServiceDiscovery of zookeeper, etcd and consul
func CacheDiscovery ¶ added in v1.8.12
func CacheDiscovery(threshold int, cachedFile string, discovery ServiceDiscovery) ServiceDiscovery
CacheDiscovery caches the services in a file, it will return the cached services if the number of services is greater than threshold. It is very useful when the register center is lost.
type ServiceDiscoveryFilter ¶ added in v1.6.2
ServiceDiscoveryFilter can be used to filter services with customized logics. Servers can register its services but clients can use the customized filter to select some services. It returns true if ServiceDiscovery wants to use this service, otherwise it returns false.
type ServiceError ¶
ServiceError is an error from server.
func NewServiceError ¶ added in v1.8.9
func NewServiceError(s string) ServiceError
NewServiceError creates a ServiceError with the error message.
type XClient ¶
type XClient interface { SetPlugins(plugins PluginContainer) GetPlugins() PluginContainer SetSelector(s Selector) ConfigGeoSelector(latitude, longitude float64) Auth(auth string) Go(ctx context.Context, serviceMethod string, args interface{}, reply interface{}, done chan *Call) (*Call, error) Call(ctx context.Context, serviceMethod string, args interface{}, reply interface{}) error Oneshot(ctx context.Context, serviceMethod string, args interface{}) error Broadcast(ctx context.Context, serviceMethod string, args interface{}, reply interface{}) error Fork(ctx context.Context, serviceMethod string, args interface{}, reply interface{}) error Inform(ctx context.Context, serviceMethod string, args interface{}, reply interface{}) ([]Receipt, error) SendRaw(ctx context.Context, r *protocol.Message) (map[string]string, []byte, error) SendFile(ctx context.Context, fileName string, rateInBytesPerSecond int64, meta map[string]string) error DownloadFile(ctx context.Context, requestFileName string, saveTo io.Writer, meta map[string]string) error Stream(ctx context.Context, meta map[string]string) (net.Conn, error) Close() error }
XClient is an interface that used by client with service discovery and service governance. One XClient is used only for one service. You should create multiple XClient for multiple services.
func NewBidirectionalXClient ¶
func NewBidirectionalXClient(servicePath string, failMode FailMode, selectMode SelectMode, discovery ServiceDiscovery, option Option, serverMessageChan chan<- *protocol.Message) XClient
NewBidirectionalXClient creates a new xclient that can receive notifications from servers.
func NewXClient ¶
func NewXClient(servicePath string, failMode FailMode, selectMode SelectMode, discovery ServiceDiscovery, option Option) XClient
NewXClient creates a XClient that supports service discovery and service governance.
type XClientPool ¶ added in v1.4.1
type XClientPool struct {
// contains filtered or unexported fields
}
XClientPool is a xclient pool with fixed size. It uses roundrobin algorithm to call its xclients. All xclients share the same configurations such as ServiceDiscovery and serverMessageChan.
func NewBidirectionalXClientPool ¶ added in v1.4.1
func NewBidirectionalXClientPool(count int, servicePath string, failMode FailMode, selectMode SelectMode, discovery ServiceDiscovery, option Option, serverMessageChan chan<- *protocol.Message) *XClientPool
NewBidirectionalXClientPool creates a BidirectionalXClient pool with fixed size.
func NewXClientPool ¶ added in v1.4.1
func NewXClientPool(count int, servicePath string, failMode FailMode, selectMode SelectMode, discovery ServiceDiscovery, option Option) *XClientPool
NewXClientPool creates a fixed size XClient pool.
func (*XClientPool) Auth ¶ added in v1.6.2
func (c *XClientPool) Auth(auth string)
Auth sets s token for Authentication.
func (*XClientPool) Close ¶ added in v1.4.1
func (p *XClientPool) Close()
Close this pool. Please make sure it won't be used any more.
func (*XClientPool) Get ¶ added in v1.4.1
func (p *XClientPool) Get() XClient
Get returns a xclient. It does not remove this xclient from its cache so you don't need to put it back. Don't close this xclient because maybe other goroutines are using this xclient.
Source Files ¶
- cache_client_builder.go
- circuit_breaker.go
- client.go
- connection.go
- connection_iouring.go
- connection_memu.go
- connection_nonkcp.go
- connection_nonquic.go
- discovery.go
- dns_discovery.go
- failmode_enumer.go
- geo_utils.go
- hash_utils.go
- mdns_discovery.go
- mode.go
- multiple_servers_discovery.go
- oneclient.go
- oneclient_pool.go
- peer2peer_discovery.go
- ping_utils.go
- plugin.go
- selectmode_enumer.go
- selector.go
- smooth_weighted_round_robin.go
- xclient.go
- xclient_pool.go