Versions in this module Expand all Collapse all v1 v1.2.1 May 9, 2023 v1.2.0 Jun 7, 2022 v1.1.7 Jun 7, 2022 Changes in this version + const FileTransferBufferSize + const ReaderBuffsize + const WriterBuffsize + const XErrorMessage + const XHeartbeat + const XMessageID + const XMessageStatusType + const XMessageType + const XMeta + const XOneway + const XSerializeType + const XServiceMethod + const XServicePath + const XVersion + var ConnFactories = map[string]ConnFactoryFn + var DefaultOption = Option + var ErrBreakerOpen = errors.New("breaker open") + var ErrBreakerTimeout = errors.New("breaker time out") + var ErrServerUnavailable = errors.New("selected server is unavilable") + var ErrShutdown = errors.New("connection is shut down") + var ErrUnsupportedCodec = errors.New("unsupported codec") + var ErrXClientNoServer = errors.New("can not found any server") + var ErrXClientShutdown = errors.New("xClient is shut down") + 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 interface + Call func(func() error, time.Duration) error + Fail func() + Ready func() bool + Success func() + var CircuitBreaker Breaker = circuit.NewRateBreaker(0.95, 100) + type CacheClientBuilder interface + DeleteCachedClient func(client RPCClient, k, servicePath, serviceMethod string) + FindCachedClient func(k, servicePath, serviceMethod string) RPCClient + GenerateClient func(k, servicePath, serviceMethod string) (client RPCClient, err error) + SetCachedClient func(client RPCClient, k, servicePath, serviceMethod string) + type Call struct + Args interface{} + Done chan *Call + Error error + Metadata map[string]string + Raw bool + Reply interface{} + ResMetadata map[string]string + ServiceMethod string + ServicePath string + type Client struct + Conn net.Conn + Plugins PluginContainer + ServerMessageChan chan<- *protocol.Message + ServerMessageChanMu sync.RWMutex + func NewClient(option Option) *Client + func (c *Client) Connect(network, address string) error + func (c *Client) GetConn() net.Conn + func (c *Client) RemoteAddr() string + func (client *Client) Call(ctx context.Context, servicePath, serviceMethod string, args interface{}, ...) error + func (client *Client) Close() error + 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) SendRaw(ctx context.Context, r *protocol.Message) (map[string]string, []byte, error) + func (client *Client) UnregisterServerMessageChan() + type ClientAfterDecodePlugin interface + ClientAfterDecode func(*protocol.Message) error + type ClientBeforeEncodePlugin interface + ClientBeforeEncode func(*protocol.Message) error + type ClientConnectedPlugin interface + ClientConnected func(net.Conn) (net.Conn, error) + type ClientConnectionClosePlugin interface + ClientConnectionClose func(net.Conn) error + type ConnCreatedPlugin interface + ConnCreated func(net.Conn) (net.Conn, error) + type ConnFactoryFn func(c *Client, network, address string) (net.Conn, error) + type ConsecCircuitBreaker struct + func NewConsecCircuitBreaker(failureThreshold uint64, window time.Duration) *ConsecCircuitBreaker + func (cb *ConsecCircuitBreaker) Call(fn func() error, d time.Duration) error + func (cb *ConsecCircuitBreaker) Fail() + func (cb *ConsecCircuitBreaker) Ready() bool + func (cb *ConsecCircuitBreaker) Success() + type ConsistentAddrStrFunction func(options ...interface{}) string + type ConsulDiscovery struct + RetriesAfterWatchFailed int + func (d *ConsulDiscovery) Clone(servicePath string) (ServiceDiscovery, error) + func (d *ConsulDiscovery) Close() + func (d *ConsulDiscovery) GetServices() []*KVPair + func (d *ConsulDiscovery) RemoveWatcher(ch chan []*KVPair) + func (d *ConsulDiscovery) SetFilter(filter ServiceDiscoveryFilter) + func (d *ConsulDiscovery) WatchService() chan []*KVPair + type DNSDiscovery struct + 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 int + const Failbackup + const Failfast + const Failover + const Failtry + func FailModeString(s string) (FailMode, error) + func FailModeValues() []FailMode + func (i FailMode) IsAFailMode() bool + func (i FailMode) String() string + type HashServiceAndArgs func(len int, options ...interface{}) int + type KVPair struct + Key string + Value string + type MDNSDiscovery struct + Timeout time.Duration + WatchInterval time.Duration + 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 struct + 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 struct + Plugins PluginContainer + func NewBidirectionalOneClient(failMode FailMode, selectMode SelectMode, discovery ServiceDiscovery, ...) *OneClient + func NewOneClient(failMode FailMode, selectMode SelectMode, discovery ServiceDiscovery, ...) *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 struct + func NewBidirectionalOneClientPool(count int, failMode FailMode, selectMode SelectMode, ...) *OneClientPool + func NewOneClientPool(count int, failMode FailMode, selectMode SelectMode, ...) *OneClientPool + func (c *OneClientPool) Auth(auth string) + func (p *OneClientPool) Get() *OneClient + func (p OneClientPool) Close() + type OpenCensusPlugin struct + func (p *OpenCensusPlugin) PostCall(ctx context.Context, servicePath, serviceMethod string, args interface{}, ...) error + func (p *OpenCensusPlugin) PreCall(ctx context.Context, servicePath, serviceMethod string, args interface{}) error + type OpenTracingPlugin struct + func (p *OpenTracingPlugin) PostCall(ctx context.Context, servicePath, serviceMethod string, args interface{}, ...) error + func (p *OpenTracingPlugin) PreCall(ctx context.Context, servicePath, serviceMethod string, args interface{}) error + type Option struct + BackupLatency time.Duration + Block interface{} + CompressType protocol.CompressType + ConnectTimeout time.Duration + GenBreaker func() Breaker + Group string + Heartbeat bool + HeartbeatInterval time.Duration + IdleTimeout time.Duration + MaxWaitForHeartbeat time.Duration + RPCPath string + Retries int + SerializeType protocol.SerializeType + TCPKeepAlivePeriod time.Duration + TLSConfig *tls.Config + type Peer2PeerDiscovery struct + 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 interface + type PluginContainer interface + Add func(plugin Plugin) + All func() []Plugin + DoClientAfterDecode func(*protocol.Message) error + DoClientBeforeEncode func(*protocol.Message) error + DoClientConnected func(net.Conn) (net.Conn, error) + DoClientConnectionClose func(net.Conn) error + DoConnCreated func(net.Conn) (net.Conn, error) + DoPostCall func(ctx context.Context, servicePath, serviceMethod string, args interface{}, ...) error + DoPreCall func(ctx context.Context, servicePath, serviceMethod string, args interface{}) error + DoWrapSelect func(SelectFunc) SelectFunc + Remove func(plugin Plugin) + func NewPluginContainer() PluginContainer + type PostCallPlugin interface + PostCall func(ctx context.Context, servicePath, serviceMethod string, args interface{}, ...) error + type PreCallPlugin interface + PreCall func(ctx context.Context, servicePath, serviceMethod string, args interface{}) error + type RPCClient interface + Call func(ctx context.Context, servicePath, serviceMethod string, args interface{}, ...) error + Close func() error + Connect func(network, address string) error + GetConn func() net.Conn + Go func(ctx context.Context, servicePath, serviceMethod string, args interface{}, ...) *Call + IsClosing func() bool + IsShutdown func() bool + RegisterServerMessageChan func(ch chan<- *protocol.Message) + RemoteAddr func() string + SendRaw func(ctx context.Context, r *protocol.Message) (map[string]string, []byte, error) + UnregisterServerMessageChan func() + type RedisDiscovery struct + RetriesAfterWatchFailed int + func (d *RedisDiscovery) Clone(servicePath string) (ServiceDiscovery, error) + func (d *RedisDiscovery) Close() + func (d *RedisDiscovery) GetServices() []*KVPair + func (d *RedisDiscovery) RemoveWatcher(ch chan []*KVPair) + func (d *RedisDiscovery) SetFilter(filter ServiceDiscoveryFilter) + func (d *RedisDiscovery) WatchService() chan []*KVPair + type SelectFunc func(ctx context.Context, servicePath, serviceMethod string, args interface{}) string + type SelectMode int + const Closest + const ConsistentHash + const RandomSelect + const RoundRobin + const SelectByUser + const WeightedICMP + const WeightedRoundRobin + func SelectModeString(s string) (SelectMode, error) + func SelectModeValues() []SelectMode + func (i SelectMode) IsASelectMode() bool + func (i SelectMode) String() string + type SelectNodePlugin interface + WrapSelect func(SelectFunc) SelectFunc + type Selector interface + Select func(ctx context.Context, servicePath, serviceMethod string, args interface{}) string + UpdateServer func(servers map[string]string) + type ServiceDiscovery interface + Clone func(servicePath string) (ServiceDiscovery, error) + Close func() + GetServices func() []*KVPair + RemoveWatcher func(ch chan []*KVPair) + SetFilter func(ServiceDiscoveryFilter) + WatchService func() chan []*KVPair + func NewConsulDiscovery(basePath, servicePath string, consulAddr []string, options *store.Config) (ServiceDiscovery, error) + func NewConsulDiscoveryStore(basePath string, kv store.Store) (ServiceDiscovery, error) + func NewConsulDiscoveryTemplate(basePath string, consulAddr []string, options *store.Config) (ServiceDiscovery, error) + func NewDNSDiscovery(domain string, network string, port int, d time.Duration) (ServiceDiscovery, error) + func NewMDNSDiscovery(service string, timeout time.Duration, watchInterval time.Duration, ...) (ServiceDiscovery, error) + func NewMultipleServersDiscovery(pairs []*KVPair) (ServiceDiscovery, error) + func NewPeer2PeerDiscovery(server, metadata string) (ServiceDiscovery, error) + func NewRedisDiscovery(basePath string, servicePath string, redisAddr []string, options *store.Config) (ServiceDiscovery, error) + func NewRedisDiscoveryStore(basePath string, kv store.Store) (ServiceDiscovery, error) + func NewRedisDiscoveryTemplate(basePath string, etcdAddr []string, options *store.Config) (ServiceDiscovery, error) + func NewZookeeperDiscovery(basePath string, servicePath string, zkAddr []string, options *store.Config) (ServiceDiscovery, error) + func NewZookeeperDiscoveryTemplate(basePath string, zkAddr []string, options *store.Config) (ServiceDiscovery, error) + func NewZookeeperDiscoveryWithStore(basePath string, kv store.Store) (ServiceDiscovery, error) + type ServiceDiscoveryFilter func(kvp *KVPair) bool + type ServiceError string + func (e ServiceError) Error() string + type Weighted struct + CurrentWeight int + EffectiveWeight int + Server string + Weight int + type XClient interface + Auth func(auth string) + Broadcast func(ctx context.Context, serviceMethod string, args interface{}, reply interface{}) error + Call func(ctx context.Context, serviceMethod string, args interface{}, reply interface{}) error + Close func() error + ConfigGeoSelector func(latitude, longitude float64) + DownloadFile func(ctx context.Context, requestFileName string, saveTo io.Writer, ...) error + Fork func(ctx context.Context, serviceMethod string, args interface{}, reply interface{}) error + GetPlugins func() PluginContainer + Go func(ctx context.Context, serviceMethod string, args interface{}, reply interface{}, ...) (*Call, error) + SendFile func(ctx context.Context, fileName string, rateInBytesPerSecond int64, ...) error + SendRaw func(ctx context.Context, r *protocol.Message) (map[string]string, []byte, error) + SetPlugins func(plugins PluginContainer) + SetSelector func(s Selector) + Stream func(ctx context.Context, meta map[string]string) (net.Conn, error) + func NewBidirectionalXClient(servicePath string, failMode FailMode, selectMode SelectMode, ...) XClient + func NewXClient(servicePath string, failMode FailMode, selectMode SelectMode, ...) XClient + type XClientPool struct + func NewBidirectionalXClientPool(count int, servicePath string, failMode FailMode, selectMode SelectMode, ...) *XClientPool + func NewXClientPool(count int, servicePath string, failMode FailMode, selectMode SelectMode, ...) *XClientPool + func (c *XClientPool) Auth(auth string) + func (p *XClientPool) Close() + func (p *XClientPool) Get() XClient + type ZookeeperDiscovery struct + RetriesAfterWatchFailed int + func (d *ZookeeperDiscovery) Clone(servicePath string) (ServiceDiscovery, error) + func (d *ZookeeperDiscovery) Close() + func (d *ZookeeperDiscovery) GetServices() []*KVPair + func (d *ZookeeperDiscovery) RemoveWatcher(ch chan []*KVPair) + func (d *ZookeeperDiscovery) SetFilter(filter ServiceDiscoveryFilter) + func (d *ZookeeperDiscovery) WatchService() chan []*KVPair