rpcx

package module
v1.1.1 Latest Latest
Warning

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

 Go to latest Published: Oct 10, 2016 License: Apache-2.0 Imports: 14 Imported by: 0

README

rpcx

中文介绍 中文电子书

License GoDoc travis Go Report Card

rpcx is a distributed RPC framework like Alibaba Dubbo and Weibo Motan. It is developed based on Go net/rpc and provides extra governance features.

Throughput

very very good performance. Much better than gRPC, Dubbo and weibo Motan.

When we talk about RPC frameworks, Dubbo is first framework we should introduced, and there is also Dubbox mantained by dangdang. Dubbo has been widely used in e-commerce companies in China, for example, Alibaba, Jingdong and Dangdang.

Though Dubbo has still used Spring 2.5.6.SEC03 and seems has not been supported by Alibaba no longer, some other companies still use it and maintained their branches.

DUBBO is a distributed service framework , provides high performance and transparent RPC remote service call. It is the core framework of Alibaba SOA service governance programs. There are 3,000,000,000 calls for 2,000+ services per day, and it has been widely used in various member sites of Alibaba Group.

Motan is open source now by Weibo. As Zhang Lei said, he is current main developer of Motan:

Motan started in 2013. There are 100 billion calls for hundreds of service callsevery day.

Those two RPC frameworks are developed by Java. There are other famous RPC frameworks such as thrift、finagle。

Goal of rpcx is implemented a RPC framework like Dubbo in Go ecosphere. It is developed by Go, and for Go.

It is a distributed、plugable RPC framework with governance (service discovery、load balancer、fault tolerance、monitor, etc.).

As you know, there are some RPC frameworks, for example, net/rpc、grpc-go、gorilla-rpc, Then why re-invent a wheel?

Although those Go RPC frameworks work well, but their function is relatively simple and only implement end-to end communications. Some product features of service management functions are lack, such as service discovery, Load balancing, fault tolerance.

So I created rpcx and expect it could become a RPC framework like Dubbo.

The similar project is go-micro.

What's RPC

From wikiPedia:

In distributed computing, a remote procedure call (RPC) is when a computer program causes a procedure (subroutine) to execute in another address space (commonly on another computer on a shared network), which is coded as if it were a normal (local) procedure call, without the programmer explicitly coding the details for the remote interaction. That is, the programmer writes essentially the same code whether the subroutine is local to the executing program, or remote.[1] This is a form of client–server interaction (caller is client, executer is server), typically implemented via a request–response message-passing system. The object-oriented programming analog is remote method invocation (RMI). The RPC model implies a level of location transparency, namely that calling procedures is largely the same whether it is local or remote, but usually they are not identical, so local calls can be distinguished from remote calls. Remote calls are usually orders of magnitude slower and less reliable than local calls, so distinguishing them is useful.

RPCs are a form of inter-process communication (IPC), in that different processes have different address spaces: if on the same host machine, they have distinct virtual address spaces, even though the physical address space is the same; while if they are on different hosts, the physical address space is different. Many different (often incompatible) technologies have been used to implement the concept.

Sequence of events during an RPC

  1. The client calls the client stub. The call is a local procedure call, with parameters pushed on to the stack in the normal way.
  2. The client stub packs the parameters into a message and makes a system call to send the message. Packing the parameters is called marshalling.
  3. The client's local operating system sends the message from the client machine to the server machine.
  4. The local operating system on the server machine passes the incoming packets to the server stub.
  5. The server stub unpacks the parameters from the message. Unpacking the parameters is called unmarshalling.
  6. Finally, the server stub calls the server procedure. The reply traces the same steps in the reverse direction.

There are two ways to implement RPC frameworks. One focusses on cross-language calls and the other focusses on service governance.

Dubbo、DubboX、Motan are RPC framework of service governance . Thrift、gRPC、Hessian、Hprose are RPC framework of cross-language calls.

rpcx is a RPC framework of service governance.

Features

more features

  • bases on net/rpc. a Go net/prc project can be converted rpcx project whit few changes.
  • Plugable. Features are implemented by Plugins such as service discovery.
  • Communicate with TCP long connections.
  • Support GeoLocation
  • support many codec. for example, Gob、Json、MessagePack、gencode、ProtoBuf.
  • Service dicovery. support ZooKeeper、Etcd.
  • Fault tolerance：Failover、Failfast、Failtry.
  • Load banlancer：support randomSelecter, RoundRobin, consistent hash etc.
  • scalable.
  • Other: metrics、log、timeout.
  • Authorization.

rpcx-ui provides a web ui for services management. rpcx-ui

Architecture

rpcx contains three roles : RPC Server，RPC Client and Registry.

  • Server registers services on Registry
  • Client queries service list and select a server from server list returned from Registry.
  • When a Server is down, Registry can remove this server and then client can remove it too.

So far rpcx support zookeeper, etcd as Registry，Consul support is developing。

Examples

There is a client to call Posts service at tr.colobu.com. This service returns a hot tech post list of today:

package main

import (
	"encoding/json"
	"fmt"
	"time"

	"gopkg.in/mgo.v2/bson"

	"github.com/smallnest/rpcx"
)

type Args struct {
	PostType string `msg:"posttype"`
}

type Reply struct {
	Posts []Post `msg:"posts"`
}

type Post struct {
	PostID      bson.ObjectId `json:"id" xml:"id" bson:"_id,omitempty"`
	PostType    string        `json:"ptype" xml:"ptype" bson:"ptype,omitempty"`
	Title       string        `json:"title" xml:"title" bson:"title"`
	URL         string        `json:"url" xml:"url" bson:"url"`
	Domain      string        `json:"domain" xml:"domain" bson:"domain"`
	ShortURL    string        `json:"surl" xml:"surl" bson:"surl"`
	Description string        `json:"desc" xml:"desc" bson:"desc"`
	LikeCount   int           `json:"like" xml:"like" bson:"like"`
	ImageURL    string        `json:"imgurl" xml:"imgurl" bson:"imgurl"`
	RecommendBy string        `json:"-" xml:"-" bson:"-"`
	Tags        string        `json:"tags" xml:"tags" bson:"tags"`
	State       int           `json:"-" xml:"-" bson:"-"`
	Timestamp   time.Time     `json:"ts" xml:"timestamp" bson:"ts"`
}

func main() {
	s := &rpcx.DirectClientSelector{Network: "tcp", Address: "tr.colobu.com:8972", DialTimeout: 10 * time.Second}
	client := rpcx.NewClient(s)
	defer client.Close()

	args := &Args{"golang"}
	var reply Reply
	err := client.Call("Posts.Query", args, &reply)
	if err != nil {
		fmt.Printf("error for Posts: %s, %v \n", args.PostType, err)
		return
	}

	posts := reply.Posts
	data, _ := json.MarshalIndent(&posts, "", "\t")

	fmt.Printf("Posts: %s \n", string(data))
}

you can found more examples at _examples

Benchmark

Test Environment

  • CPU: Intel(R) Xeon(R) CPU E5-2620 v2 @ 2.10GHz, 24 cores
  • Memory: 16G
  • OS: Linux Server-3 2.6.32-358.el6.x86_64, CentOS 6.4
  • Go: 1.7

Test request is copied from protobuf project and encoded to a proto message. Its size is 581 bytes. The response update two fields of decoded requests so servers handle the decoding and encoding processing.

The test proto file is:

syntax = "proto2";

package main;

option optimize_for = SPEED;


message BenchmarkMessage {
  required string field1 = 1;
  optional string field9 = 9;
  optional string field18 = 18;
  optional bool field80 = 80 [default=false];
  optional bool field81 = 81 [default=true];
  required int32 field2 = 2;
  required int32 field3 = 3;
  optional int32 field280 = 280;
  optional int32 field6 = 6 [default=0];
  optional int64 field22 = 22;
  optional string field4 = 4;
  repeated fixed64 field5 = 5;
  optional bool field59 = 59 [default=false];
  optional string field7 = 7;
  optional int32 field16 = 16;
  optional int32 field130 = 130 [default=0];
  optional bool field12 = 12 [default=true];
  optional bool field17 = 17 [default=true];
  optional bool field13 = 13 [default=true];
  optional bool field14 = 14 [default=true];
  optional int32 field104 = 104 [default=0];
  optional int32 field100 = 100 [default=0];
  optional int32 field101 = 101 [default=0];
  optional string field102 = 102;
  optional string field103 = 103;
  optional int32 field29 = 29 [default=0];
  optional bool field30 = 30 [default=false];
  optional int32 field60 = 60 [default=-1];
  optional int32 field271 = 271 [default=-1];
  optional int32 field272 = 272 [default=-1];
  optional int32 field150 = 150;
  optional int32 field23 = 23 [default=0];
  optional bool field24 = 24 [default=false];
  optional int32 field25 = 25 [default=0];
  optional bool field78 = 78;
  optional int32 field67 = 67 [default=0];
  optional int32 field68 = 68;
  optional int32 field128 = 128 [default=0];
  optional string field129 = 129 [default="xxxxxxxxxxxxxxxxxxxxx"];
  optional int32 field131 = 131 [default=0];
}

The concurrent clients are 100, 1000,2000 and 5000. Count of the total requests for all clients are 1,000,000.

Test Result

rpcx: one client and one server in a same machine
concurrent clients mean(ms) median(ms) max(ms) min(ms) throughput(TPS)
100 0 0 17 0 164338
500 2 1 40 0 181126
1000 4 3 56 0 186219
2000 9 7 105 0 182815
5000 25 22 200 0 178858

you can use test code in _benchmark to test. server is used to start a server and client is used as clients via protobuf.

The above test is that client and server are running on the same mechine.

rpcx: one client and one server in separated machines

If I run them on separate servers, test results are:

concurrent clients mean(ms) median(ms) max(ms) min(ms) throughput(TPS)
100 1 1 20 0 127975
500 5 1 4350 0 136407
1000 10 2 3233 0 155255
2000 17 2 9735 0 159438
5000 44 2 12788 0 161917
rpcx: one client on a machine and two servers in two machines

If they are running on cluster mode, one is for the client and another two are for servers, test results are:

concurrent clients mean(ms) median(ms) max(ms) min(ms) throughput(TPS)
100 0 0 41 0 128932
500 3 2 273 0 150285
1000 5 5 621 0 150152
2000 10 7 288 0 159974
5000 23 12 629 0 155279
benchmarks of serialization libraries:
[root@localhost rpcx]# go test -bench . -test.benchmem
PASS
BenchmarkNetRPC_gob-16            100000             18742 ns/op             321 B/op          9 allocs/op
BenchmarkNetRPC_jsonrpc-16        100000             21360 ns/op            1170 B/op         31 allocs/op
BenchmarkNetRPC_msgp-16           100000             18617 ns/op             776 B/op         35 allocs/op
BenchmarkRPCX_gob-16              100000             18718 ns/op             320 B/op          9 allocs/op
BenchmarkRPCX_json-16             100000             21238 ns/op            1170 B/op         31 allocs/op
BenchmarkRPCX_msgp-16             100000             18635 ns/op             776 B/op         35 allocs/op
BenchmarkRPCX_gencodec-16         100000             18454 ns/op            4485 B/op         17 allocs/op
BenchmarkRPCX_protobuf-16         100000             17234 ns/op             733 B/op         13 allocs/op

Comparision with gRPC

gRPC is the RPC framework by Google. It support multiple programming lanaguage. I have compared three cases for prcx and gRPC. It shows rpcx is much better than gRPC.

current gRPC version is 1.0 .

Test results of rpcx has been listed on the above. Here is test results of gRPC.

gRPC: one client and one server in a same machine
concurrent clients mean(ms) median(ms) max(ms) min(ms) throughput(TPS)
100 1 0 21 0 68250
500 5 1 3059 0 78486
1000 10 1 6274 0 79980
2000 19 1 9736 0 58129
5000 43 2 14224 0 44724

gRPC: one client and one server in separated machines
concurrent clients mean(ms) median(ms) max(ms) min(ms) throughput(TPS)
100 1 0 21 0 68250
500 5 1 3059 0 78486
1000 10 1 6274 0 79980
2000 19 1 9736 0 58129
5000 43 2 14224 0 44724

gRPC: one client on a machine and two servers in two machines
concurrent clients mean(ms) median(ms) max(ms) min(ms) throughput(TPS)
100 1 0 19 0 88082
500 4 1 1461 0 90334
1000 9 1 6315 0 62305
2000 17 1 9736 0 44487
5000 38 1 25087 0 33198

Documentation

Index

Constants

View Source 
const (
	//DefaultRPCPath is the defaut HTTP RPC PATH
	DefaultRPCPath = "/_goRPC_"
)

Variables

View Source 
var (
	// ErrPluginAlreadyExists returns an error with message: 'Cannot activate the same plugin again, plugin '+plugin name[+plugin description]' is already exists'
	ErrPluginAlreadyExists = NewRPCError("Cannot use the same plugin again, '%s[%s]' is already exists")
	// ErrPluginActivate returns an error with message: 'While trying to activate plugin '+plugin name'. Trace: +specific error'
	ErrPluginActivate = NewRPCError("While trying to activate plugin '%s'. Trace: %s")
	// ErrPluginRemoveNoPlugins returns an error with message: 'No plugins are registed yet, you cannot remove a plugin from an empty list!'
	ErrPluginRemoveNoPlugins = NewRPCError("No plugins are registed yet, you cannot remove a plugin from an empty list!")
	// ErrPluginRemoveEmptyName returns an error with message: 'Plugin with an empty name cannot be removed'
	ErrPluginRemoveEmptyName = NewRPCError("Plugin with an empty name cannot be removed")
	// ErrPluginRemoveNotFound returns an error with message: 'Cannot remove a plugin which doesn't exists'
	ErrPluginRemoveNotFound = NewRPCError("Cannot remove a plugin which doesn't exists")
)

Functions

func Auth 

func Auth(fn AuthorizationFunc) error

Auth sets authorization handler

func Close 

func Close() error

Close closes RPC server.

func GetListenedAddress 

func GetListenedAddress() string

GetListenedAddress return the listening address.

func NewDirectHTTPRPCClient 

func NewDirectHTTPRPCClient(c *Client, clientCodecFunc ClientCodecFunc, network, address string, path string, timeout time.Duration) (*rpc.Client, error)

NewDirectHTTPRPCClient creates a rpc http client

func NewDirectRPCClient 

func NewDirectRPCClient(c *Client, clientCodecFunc ClientCodecFunc, network, address string, timeout time.Duration) (*rpc.Client, error)

NewDirectRPCClient creates a rpc client

func RegisterName 

func RegisterName(name string, service interface{}, metadata ...string)

RegisterName publishes in the server the set of methods .

func Serve 

func Serve(n, address string)

Serve starts and listens RCP requests. It is blocked until receiving connectings from clients.

func ServeByHTTP 

func ServeByHTTP(ln net.Listener, rpcPath, debugPath string)

ServeByHTTP implements RPC via HTTP

func ServeListener 

func ServeListener(ln net.Listener)

ServeListener serve with a listener

func ServeTLS 

func ServeTLS(n, address string, config *tls.Config)

ServeTLS starts and listens RCP requests. It is blocked until receiving connectings from clients.

func SetServerCodecFunc 

func SetServerCodecFunc(fn ServerCodecFunc)

SetServerCodecFunc sets a ServerCodecFunc

func Start 

func Start(n, address string)

Start starts and listens RCP requests without blocking.

func StartTLS 

func StartTLS(n, address string, config *tls.Config)

StartTLS starts and listens RCP requests without blocking.

Types

type AuthorizationAndServiceMethod 

type AuthorizationAndServiceMethod struct {
	Authorization string // Authorization
	ServiceMethod string // real ServiceMethod name
	Tag           string // extra tag for Authorization
}

AuthorizationAndServiceMethod represents Authorization header and ServiceMethod.

func (*AuthorizationAndServiceMethod) String added in v1.1.1 

func (aasm *AuthorizationAndServiceMethod) String() string

type AuthorizationClientPlugin 

type AuthorizationClientPlugin struct {
	AuthorizationAndServiceMethod *AuthorizationAndServiceMethod
}

AuthorizationClientPlugin is used to set Authorization info at client side.

func NewAuthorizationClientPlugin 

func NewAuthorizationClientPlugin(authorization, tag string) *AuthorizationClientPlugin

NewAuthorizationClientPlugin creates a AuthorizationClientPlugin with authorization header and tag

func (*AuthorizationClientPlugin) Description 

func (plugin *AuthorizationClientPlugin) Description() string

Description return description of this plugin.

func (*AuthorizationClientPlugin) Name 

func (plugin *AuthorizationClientPlugin) Name() string

Name return name of this plugin.

func (*AuthorizationClientPlugin) PreWriteRequest 

func (plugin *AuthorizationClientPlugin) PreWriteRequest(r *rpc.Request, body interface{}) error

PreWriteRequest adds Authorization info in requests

type AuthorizationFunc 

type AuthorizationFunc func(p *AuthorizationAndServiceMethod) error

AuthorizationFunc defines a method type which handles Authorization info

type AuthorizationServerPlugin 

type AuthorizationServerPlugin struct {
	AuthorizationFunc AuthorizationFunc
}

AuthorizationServerPlugin is used to authorize clients.

func (*AuthorizationServerPlugin) Description 

func (plugin *AuthorizationServerPlugin) Description() string

Description return description of this plugin.

func (*AuthorizationServerPlugin) Name 

func (plugin *AuthorizationServerPlugin) Name() string

Name return name of this plugin.

func (*AuthorizationServerPlugin) PostReadRequestHeader 

func (plugin *AuthorizationServerPlugin) PostReadRequestHeader(r *rpc.Request) (err error)

PostReadRequestHeader extracts Authorization header from ServiceMethod field.

type Client 

type Client struct {
	ClientSelector  ClientSelector
	ClientCodecFunc ClientCodecFunc
	PluginContainer IClientPluginContainer
	FailMode        FailMode
	TLSConfig       *tls.Config
	Retries         int
	//Timeout sets deadline for underlying net.Conns
	Timeout time.Duration
	//Timeout sets readdeadline for underlying net.Conns
	ReadTimeout time.Duration
	//Timeout sets writedeadline for underlying net.Conns
	WriteTimeout time.Duration
}

Client represents a RPC client.

func NewClient 

func NewClient(s ClientSelector) *Client

NewClient create a client.

func (*Client) Auth 

func (c *Client) Auth(authorization, tag string) error

Auth sets Authorization info

func (*Client) Call 

func (c *Client) Call(serviceMethod string, args interface{}, reply interface{}) (err error)

Call invokes the named function, waits for it to complete, and returns its error status.

func (*Client) Close 

func (c *Client) Close() error

Close closes the connection

func (*Client) Go 

func (c *Client) Go(serviceMethod string, args interface{}, reply interface{}, done chan *rpc.Call) *rpc.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.

type ClientCodecFunc 

type ClientCodecFunc func(conn io.ReadWriteCloser) rpc.ClientCodec

ClientCodecFunc is used to create a rpc.ClientCodecFunc from net.Conn.

type ClientPluginContainer 

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

ClientPluginContainer implements IPluginContainer interface.

func (*ClientPluginContainer) Add 

func (p *ClientPluginContainer) Add(plugin IPlugin) error

Add adds a plugin.

func (*ClientPluginContainer) DoPostReadResponseBody 

func (p *ClientPluginContainer) DoPostReadResponseBody(body interface{}) error

DoPostReadResponseBody invokes DoPostReadResponseBody plugin.

func (*ClientPluginContainer) DoPostReadResponseHeader 

func (p *ClientPluginContainer) DoPostReadResponseHeader(r *rpc.Response) error

DoPostReadResponseHeader invokes DoPostReadResponseHeader plugin.

func (*ClientPluginContainer) DoPostWriteRequest 

func (p *ClientPluginContainer) DoPostWriteRequest(r *rpc.Request, body interface{}) error

DoPostWriteRequest invokes DoPostWriteRequest plugin.

func (*ClientPluginContainer) DoPreReadResponseBody 

func (p *ClientPluginContainer) DoPreReadResponseBody(body interface{}) error

DoPreReadResponseBody invokes DoPreReadResponseBody plugin.

func (*ClientPluginContainer) DoPreReadResponseHeader 

func (p *ClientPluginContainer) DoPreReadResponseHeader(r *rpc.Response) error

DoPreReadResponseHeader invokes DoPreReadResponseHeader plugin.

func (*ClientPluginContainer) DoPreWriteRequest 

func (p *ClientPluginContainer) DoPreWriteRequest(r *rpc.Request, body interface{}) error

DoPreWriteRequest invokes DoPreWriteRequest plugin.

func (*ClientPluginContainer) GetAll 

func (p *ClientPluginContainer) GetAll() []IPlugin

GetAll returns all activated plugins

func (*ClientPluginContainer) GetByName 

func (p *ClientPluginContainer) GetByName(pluginName string) IPlugin

GetByName returns a plugin instance by it's name

func (*ClientPluginContainer) GetDescription 

func (p *ClientPluginContainer) GetDescription(plugin IPlugin) string

GetDescription returns the name of a plugin, if no GetDescription() implemented it returns an empty string ""

func (*ClientPluginContainer) GetName 

func (p *ClientPluginContainer) GetName(plugin IPlugin) string

GetName returns the name of a plugin, if no GetName() implemented it returns an empty string ""

func (*ClientPluginContainer) Remove 

func (p *ClientPluginContainer) Remove(pluginName string) error

Remove removes a plugin by it's name.

type ClientSelector 

type ClientSelector interface {
	//Select returns a new client and it also update current client
	Select(clientCodecFunc ClientCodecFunc, options ...interface{}) (*rpc.Client, error)
	//SetClient set current client
	SetClient(*Client)
	SetSelectMode(SelectMode)
	//AllClients returns all Clients
	AllClients(clientCodecFunc ClientCodecFunc) []*rpc.Client
}

ClientSelector defines an interface to create a rpc.Client from cluster or standalone.

type DirectClientSelector 

type DirectClientSelector struct {
	Network, Address string
	DialTimeout      time.Duration
	Client           *Client
	// contains filtered or unexported fields
}

DirectClientSelector is used to a direct rpc server. It don't select a node from service cluster but a specific rpc server.

func (*DirectClientSelector) AllClients 

func (s *DirectClientSelector) AllClients(clientCodecFunc ClientCodecFunc) []*rpc.Client

AllClients returns rpc.Clients to all servers

func (*DirectClientSelector) Select 

func (s *DirectClientSelector) Select(clientCodecFunc ClientCodecFunc, options ...interface{}) (*rpc.Client, error)

Select returns a rpc client.

func (*DirectClientSelector) SetClient 

func (s *DirectClientSelector) SetClient(c *Client)

SetClient sets the unique client.

func (*DirectClientSelector) SetSelectMode 

func (s *DirectClientSelector) SetSelectMode(sm SelectMode)

SetSelectMode is meaningless for DirectClientSelector because there is only one client.

type FailMode 

type FailMode int

FailMode is a feature to decide client actions 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
	//Broadcast sends requests to all servers and Success only when all servers return OK
	Broadcast
	//Forking sends requests to all servers and Success once one server returns OK
	Forking
)

type IClientPluginContainer 

type IClientPluginContainer interface {
	Add(plugin IPlugin) error
	Remove(pluginName string) error
	GetName(plugin IPlugin) string
	GetDescription(plugin IPlugin) string
	GetByName(pluginName string) IPlugin
	GetAll() []IPlugin

	DoPreReadResponseHeader(*rpc.Response) error
	DoPostReadResponseHeader(*rpc.Response) error
	DoPreReadResponseBody(interface{}) error
	DoPostReadResponseBody(interface{}) error

	DoPreWriteRequest(*rpc.Request, interface{}) error
	DoPostWriteRequest(*rpc.Request, interface{}) error
}

IClientPluginContainer represents a plugin container that defines all methods to manage plugins. And it also defines all extension points.

type IPlugin 

type IPlugin interface {
	Name() string
	Description() string
}

IPlugin represents a plugin.

type IPostConnAcceptPlugin 

type IPostConnAcceptPlugin interface {
	HandleConnAccept(net.Conn) bool
}

IPostConnAcceptPlugin represents connection accept plugin. if returns false, it means subsequent IPostConnAcceptPlugins should not contiune to handle this conn and this conn has been closed.

type IPostReadRequestBodyPlugin 

type IPostReadRequestBodyPlugin interface {
	PostReadRequestBody(body interface{}) error
}

IPostReadRequestBodyPlugin represents .

type IPostReadRequestHeaderPlugin 

type IPostReadRequestHeaderPlugin interface {
	PostReadRequestHeader(r *rpc.Request) error
}

IPostReadRequestHeaderPlugin represents .

type IPostReadResponseBodyPlugin 

type IPostReadResponseBodyPlugin interface {
	PostReadResponseBody(interface{}) error
}

IPostReadResponseBodyPlugin represents .

type IPostReadResponseHeaderPlugin 

type IPostReadResponseHeaderPlugin interface {
	PostReadResponseHeader(*rpc.Response) error
}

IPostReadResponseHeaderPlugin represents .

type IPostWriteRequestPlugin 

type IPostWriteRequestPlugin interface {
	PostWriteRequest(*rpc.Request, interface{}) error
}

IPostWriteRequestPlugin represents .

type IPostWriteResponsePlugin 

type IPostWriteResponsePlugin interface {
	PostWriteResponse(*rpc.Response, interface{}) error
}

IPostWriteResponsePlugin represents .

type IPreReadRequestBodyPlugin 

type IPreReadRequestBodyPlugin interface {
	PreReadRequestBody(body interface{}) error
}

IPreReadRequestBodyPlugin represents .

type IPreReadRequestHeaderPlugin 

type IPreReadRequestHeaderPlugin interface {
	PreReadRequestHeader(r *rpc.Request) error
}

IPreReadRequestHeaderPlugin represents .

type IPreReadResponseBodyPlugin 

type IPreReadResponseBodyPlugin interface {
	PreReadResponseBody(interface{}) error
}

IPreReadResponseBodyPlugin represents .

type IPreReadResponseHeaderPlugin 

type IPreReadResponseHeaderPlugin interface {
	PreReadResponseHeader(*rpc.Response) error
}

IPreReadResponseHeaderPlugin represents .

type IPreWriteRequestPlugin 

type IPreWriteRequestPlugin interface {
	PreWriteRequest(*rpc.Request, interface{}) error
}

IPreWriteRequestPlugin represents .

type IPreWriteResponsePlugin 

type IPreWriteResponsePlugin interface {
	PreWriteResponse(*rpc.Response, interface{}) error
}

IPreWriteResponsePlugin represents .

type IRegisterPlugin 

type IRegisterPlugin interface {
	Register(name string, rcvr interface{}, metadata ...string) error
}

IRegisterPlugin represents register plugin.

type IServerCodecPlugin 

type IServerCodecPlugin interface {
	IPreReadRequestHeaderPlugin
	IPostReadRequestHeaderPlugin
	IPreReadRequestBodyPlugin
	IPostReadRequestBodyPlugin
	IPreWriteResponsePlugin
	IPostWriteResponsePlugin
}

IServerCodecPlugin represents .

type IServerPluginContainer 

type IServerPluginContainer interface {
	Add(plugin IPlugin) error
	Remove(pluginName string) error
	GetName(plugin IPlugin) string
	GetDescription(plugin IPlugin) string
	GetByName(pluginName string) IPlugin
	GetAll() []IPlugin

	DoRegister(name string, rcvr interface{}, metadata ...string) error

	DoPostConnAccept(net.Conn) bool

	DoPreReadRequestHeader(r *rpc.Request) error
	DoPostReadRequestHeader(r *rpc.Request) error

	DoPreReadRequestBody(body interface{}) error
	DoPostReadRequestBody(body interface{}) error

	DoPreWriteResponse(*rpc.Response, interface{}) error
	DoPostWriteResponse(*rpc.Response, interface{}) error
}

IServerPluginContainer represents a plugin container that defines all methods to manage plugins. And it also defines all extension points.

func GetPluginContainer 

func GetPluginContainer() IServerPluginContainer

GetPluginContainer get PluginContainer of default server.

type MultiError 

type MultiError struct {
	Errors []error
}

MultiError holds multiple errors

func NewMultiError 

func NewMultiError(errors []error) *MultiError

NewMultiError creates and returns an Error with error splice

func (*MultiError) Error 

func (e *MultiError) Error() string

Error returns the message of the actual error

type RPCError 

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

RPCError holds the error

func NewRPCError 

func NewRPCError(errMsg string) *RPCError

NewRPCError creates and returns an Error with a message

func (*RPCError) Error 

func (e *RPCError) Error() string

Error returns the message of the actual error

func (*RPCError) Format 

func (e *RPCError) Format(args ...interface{}) error

Format returns a formatted new error based on the arguments

func (*RPCError) Panic 

func (e *RPCError) Panic()

Panic output the message and after panics

func (*RPCError) Panicf 

func (e *RPCError) Panicf(args ...interface{})

Panicf output the formatted message and after panics

func (*RPCError) Return 

func (e *RPCError) Return() error

Return returns the actual error as it is

func (*RPCError) With 

func (e *RPCError) With(err error) error

With does the same thing as Format but it receives an error type which if it's nil it returns a nil error

type SelectMode 

type SelectMode int

SelectMode defines the algorithm of selecting a services from cluster 

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
)

func (SelectMode) String 

func (s SelectMode) String() string

type Server 

type Server struct {
	ServerCodecFunc ServerCodecFunc
	//PluginContainer must be configured before starting and Register plugins must be configured before invoking RegisterName method
	PluginContainer IServerPluginContainer
	//Metadata describes extra info about this service, for example, weight, active status
	Metadata string

	Timeout      time.Duration
	ReadTimeout  time.Duration
	WriteTimeout time.Duration
	// contains filtered or unexported fields
}

Server represents a RPC Server.

func NewServer 

func NewServer() *Server

NewServer returns a new Server.

func (*Server) Address 

func (s *Server) Address() string

Address return the listening address.

func (*Server) Auth 

func (s *Server) Auth(fn AuthorizationFunc) error

Auth sets authorization function

func (*Server) Close 

func (s *Server) Close() error

Close closes RPC server.

func (*Server) RegisterName 

func (s *Server) RegisterName(name string, service interface{}, metadata ...string)

RegisterName publishes in the server the set of methods of the receiver value that satisfy the following conditions:

  • exported method of exported type
  • two arguments, both of exported type
  • the second argument is a pointer
  • one return value, of type error

It returns an error if the receiver is not an exported type or has no suitable methods. It also logs the error using package log. The client accesses each method using a string of the form "Type.Method", where Type is the receiver's concrete type.

func (*Server) Serve 

func (s *Server) Serve(network, address string)

Serve starts and listens RCP requests. It is blocked until receiving connectings from clients.

func (*Server) ServeByHTTP 

func (s *Server) ServeByHTTP(ln net.Listener, rpcPath string)

ServeByHTTP starts

func (*Server) ServeHTTP 

func (s *Server) ServeHTTP(w http.ResponseWriter, req *http.Request)

ServeHTTP implements net handler interface

func (*Server) ServeListener 

func (s *Server) ServeListener(ln net.Listener)

ServeListener starts

func (*Server) ServeTLS 

func (s *Server) ServeTLS(network, address string, config *tls.Config)

ServeTLS starts and listens RCP requests. It is blocked until receiving connectings from clients.

func (*Server) Start 

func (s *Server) Start(network, address string)

Start starts and listens RCP requests without blocking.

func (*Server) StartTLS 

func (s *Server) StartTLS(network, address string, config *tls.Config)

StartTLS starts and listens RCP requests without blocking.

type ServerCodecFunc 

type ServerCodecFunc func(conn io.ReadWriteCloser) rpc.ServerCodec

ServerCodecFunc is used to create a rpc.ServerCodec from net.Conn.

type ServerPluginContainer 

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

ServerPluginContainer implements IPluginContainer interface.

func (*ServerPluginContainer) Add 

func (p *ServerPluginContainer) Add(plugin IPlugin) error

Add adds a plugin.

func (*ServerPluginContainer) DoPostConnAccept 

func (p *ServerPluginContainer) DoPostConnAccept(conn net.Conn) bool

DoPostConnAccept handle accepted conn

func (*ServerPluginContainer) DoPostReadRequestBody 

func (p *ServerPluginContainer) DoPostReadRequestBody(body interface{}) error

DoPostReadRequestBody invokes DoPostReadRequestBody plugin.

func (*ServerPluginContainer) DoPostReadRequestHeader 

func (p *ServerPluginContainer) DoPostReadRequestHeader(r *rpc.Request) error

DoPostReadRequestHeader invokes DoPostReadRequestHeader plugin.

func (*ServerPluginContainer) DoPostWriteResponse 

func (p *ServerPluginContainer) DoPostWriteResponse(resp *rpc.Response, body interface{}) error

DoPostWriteResponse invokes DoPostWriteResponse plugin.

func (*ServerPluginContainer) DoPreReadRequestBody 

func (p *ServerPluginContainer) DoPreReadRequestBody(body interface{}) error

DoPreReadRequestBody invokes DoPreReadRequestBody plugin.

func (*ServerPluginContainer) DoPreReadRequestHeader 

func (p *ServerPluginContainer) DoPreReadRequestHeader(r *rpc.Request) error

DoPreReadRequestHeader invokes DoPreReadRequestHeader plugin.

func (*ServerPluginContainer) DoPreWriteResponse 

func (p *ServerPluginContainer) DoPreWriteResponse(resp *rpc.Response, body interface{}) error

DoPreWriteResponse invokes DoPreWriteResponse plugin.

func (*ServerPluginContainer) DoRegister 

func (p *ServerPluginContainer) DoRegister(name string, rcvr interface{}, metadata ...string) error

DoRegister invokes DoRegister plugin.

func (*ServerPluginContainer) GetAll 

func (p *ServerPluginContainer) GetAll() []IPlugin

GetAll returns all activated plugins

func (*ServerPluginContainer) GetByName 

func (p *ServerPluginContainer) GetByName(pluginName string) IPlugin

GetByName returns a plugin instance by it's name

func (*ServerPluginContainer) GetDescription 

func (p *ServerPluginContainer) GetDescription(plugin IPlugin) string

GetDescription returns the name of a plugin, if no GetDescription() implemented it returns an empty string ""

func (*ServerPluginContainer) GetName 

func (p *ServerPluginContainer) GetName(plugin IPlugin) string

GetName returns the name of a plugin, if no GetName() implemented it returns an empty string ""

func (*ServerPluginContainer) Remove 

func (p *ServerPluginContainer) Remove(pluginName string) error

Remove removes a plugin by it's name.

Source Files

View all Source files

Directories

Path Synopsis
Package main is a generated protocol buffer package.
 Package main is a generated protocol buffer package.
_examples
authorization
cache
docker
end2end
etcd_registry
geo
group
java
multi_server
realservice
rpcx-ui-demoserver
timeout
tls
zookeeper_registry

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.