trpc

README

English | 中文

tRPC-Go Framework

tRPC-Go, is the Go language implementation of tRPC, which is a pluggable, high-performance RPC framework.

For more information, please refer to the quick start guide and detailed documentation.

Overall Architecture

tRPC-Go has the following features:

• Multiple services can be started within a single process, listening on multiple addresses.
• All components are pluggable, with default implementations for various basic functionalities that can be replaced. Other components can be implemented by third parties and registered within the framework.
• All interfaces can be mock tested using gomock&mockgen to generate mock code, facilitating testing.
• The framework supports any third-party protocol by implementing the codec interfaces for the respective protocol. It defaults to supporting trpc and http protocols and can be switched at any time.
• It provides the trpc command-line tool for generating code templates.

Related Documentation

• quick start guide and detailed documentation
• readme documents in each directory
• trpc command-line tool
• helloworld development guide
• example documentation for various features

Ecosystem

• codec plugins
• filter plugins
• database plugins
• more...

How to Contribute

If you're interested in contributing, please take a look at the contribution guidelines and check the unassigned issues in the repository. Claim a task and let's contribute together to tRPC-Go.

Documentation

Overview

Package trpc is the Go implementation of tRPC, which is designed to be high-performance, everything-pluggable and easy for testing.

Index

• Constants
• Variables
• func BackgroundContext() context.Context
• func CloneContext(ctx context.Context) context.Context
• func GetAdminService(s *server.Server) (*admin.Server, error)
• func GetMetaData(ctx context.Context, key string) []byte
• func Go(ctx context.Context, timeout time.Duration, handler func(context.Context)) error
• func GoAndWait(handlers ...func() error) error
• func LoadGlobalConfig(configPath string) error
• func Message(ctx context.Context) codec.Msg
• func NewServer(opt ...server.Option) *server.Server
• func NewServerWithConfig(cfg *Config, opt ...server.Option) *server.Server
• func RepairConfig(cfg *Config) error
• func Request(ctx context.Context) *trpcpb.RequestProtocol
• func Response(ctx context.Context) *trpcpb.ResponseProtocol
• func SetGlobalConfig(cfg *Config)
• func SetMetaData(ctx context.Context, key string, val []byte)
• func Setup(cfg *Config) error
• func SetupClients(cfg *ClientConfig) error
• func SetupPlugins(cfg plugin.Config) (func() error, error)
• func Version() string
• type ClientCodec
  • func (c *ClientCodec) Decode(msg codec.Msg, rspBuf []byte) (rspBody []byte, err error)
  • func (c *ClientCodec) Encode(msg codec.Msg, reqBody []byte) (reqBuf []byte, err error)
• type ClientConfig
• type ClientStreamCodec
  • func NewClientStreamCodec() *ClientStreamCodec
  • func (c *ClientStreamCodec) Decode(msg codec.Msg, rspBuf []byte) ([]byte, error)
  • func (c *ClientStreamCodec) Encode(msg codec.Msg, reqBuf []byte) ([]byte, error)
• type Config
  • func GlobalConfig() *Config
  • func LoadConfig(configPath string) (*Config, error)
• type FrameHead
• type FramerBuilder
  • func (fb *FramerBuilder) New(reader io.Reader) codec.Framer
  • func (fb *FramerBuilder) Parse(rc io.Reader) (vid uint32, buf []byte, err error)
• type Goer
  • func NewAsyncGoer(workerPoolSize int, panicBufLen int, shouldRecover bool) Goer
  • func NewSyncGoer() Goer
• type RPCZConfig
• type RecordWhenConfig
  • func (c *RecordWhenConfig) UnmarshalYAML(node *yaml.Node) error
• type ServerCodec
  • func (s *ServerCodec) Decode(msg codec.Msg, reqBuf []byte) ([]byte, error)
  • func (s *ServerCodec) Encode(msg codec.Msg, rspBody []byte) ([]byte, error)
• type ServerStreamCodec
  • func NewServerStreamCodec() *ServerStreamCodec
  • func (s *ServerStreamCodec) Decode(msg codec.Msg, reqBuf []byte) ([]byte, error)
  • func (s *ServerStreamCodec) Encode(msg codec.Msg, reqBuf []byte) (rspbuf []byte, err error)
• type ServiceConfig

Constants

const (
	DyeingKey   = "trpc-dyeing-key" // dyeing key
	UserIP      = "trpc-user-ip"    // user ip
	EnvTransfer = "trpc-env"        // env info

	ProtocolName = "trpc" // protocol name
)

frequently used const variables

const (
	MajorVersion  = 0
	MinorVersion  = 14
	PatchVersion  = 0
	VersionSuffix = "-dev" // -alpha -alpha.1 -beta -rc -rc.1
)

rule of trpc version 1. MAJOR version when you make incompatible API changes； 2. MINOR version when you add functionality in a backwards-compatible manner； 3. PATCH version when you make backwards-compatible bug fixes； 4. Additional labels for pre-release and build metadata are available as extensions to the MAJOR.MINOR.PATCH format； alpha 0.1.0-alpha beta 0.1.0-beta release candidate 0.1.0-rc release 0.1.0

Variables

var (
	DefaultServerCodec = &ServerCodec{
		streamCodec: NewServerStreamCodec(),
	}
	DefaultClientCodec = &ClientCodec{
		streamCodec:   NewClientStreamCodec(),
		defaultCaller: fmt.Sprintf("trpc.client.%s.service", path.Base(os.Args[0])),
	}
	DefaultFramerBuilder = &FramerBuilder{}

	// DefaultMaxFrameSize is the default max size of frame including attachment,
	// which can be modified if size of the packet is bigger than this.
	DefaultMaxFrameSize = 10 * 1024 * 1024
)

default codec

var DefaultGoer = NewAsyncGoer(0, PanicBufLen, true)

DefaultGoer is an async goer without worker pool.

var PanicBufLen = 1024

PanicBufLen is len of buffer used for stack trace logging when the goroutine panics, 1024 by default.

var ServerConfigPath = defaultConfigPath

ServerConfigPath is the file path of trpc server config file. By default, it's ./trpc_go.yaml. It can be set by the flag -conf.

Functions

func BackgroundContext

func BackgroundContext() context.Context

BackgroundContext puts an initialized msg into background context and returns it.

func CloneContext

func CloneContext(ctx context.Context) context.Context

CloneContext copies the context to get a context that retains the value and doesn't cancel. This is used when the handler is processed asynchronously to detach the original timeout control and retains the original context information.

After the trpc handler function returns, ctx will be canceled, and put the ctx's Msg back into pool, and the associated Metrics and logger will be released.

Before starting a goroutine to run the handler function asynchronously, this method must be called to copy context, detach the original timeout control, and retain the information in Msg for Metrics.

Retain the logger context for printing the associated log, keep other value in context, such as tracing context, etc.

func GetAdminService

func GetAdminService(s *server.Server) (*admin.Server, error)

GetAdminService gets admin service from server.Server.

func GetMetaData

func GetMetaData(ctx context.Context, key string) []byte

GetMetaData returns metadata from ctx by key.

func Go

func Go(ctx context.Context, timeout time.Duration, handler func(context.Context)) error

Go launches a safer goroutine for async task inside rpc handler. it clones ctx and msg before the goroutine, and will recover and report metrics when the goroutine panics. you should set a suitable timeout to control the lifetime of the new goroutine to prevent goroutine leaks.

func GoAndWait

func GoAndWait(handlers ...func() error) error

GoAndWait provides safe concurrent handling. Per input handler, it starts a goroutine. Then it waits until all handlers are done and will recover if any handler panics. The returned error is the first non-nil error returned by one of the handlers. It can be set that non-nil error will be returned if the "key" handler fails while other handlers always return nil error.

func LoadGlobalConfig

func LoadGlobalConfig(configPath string) error

LoadGlobalConfig loads a Config from the config file path and sets it as the global Config.

func Message

func Message(ctx context.Context) codec.Msg

Message returns msg from ctx.

func NewServer

func NewServer(opt ...server.Option) *server.Server

NewServer parses the yaml config file to quickly start the server with multiple services. The config file is ./trpc_go.yaml by default and can be set by the flag -conf. This method should be called only once.

func NewServerWithConfig

func NewServerWithConfig(cfg *Config, opt ...server.Option) *server.Server

NewServerWithConfig initializes a server with a Config. If yaml config file not used, custom Config parsing is needed to pass the Config into this function. Plugins' setup is left to do if this method is called.

func RepairConfig

func RepairConfig(cfg *Config) error

RepairConfig repairs the Config by filling in some fields with default values.

func Request

func Request(ctx context.Context) *trpcpb.RequestProtocol

Request returns RequestProtocol from ctx. If the RequestProtocol not found, a new RequestProtocol will be created and returned.

func Response

func Response(ctx context.Context) *trpcpb.ResponseProtocol

Response returns ResponseProtocol from ctx. If the ResponseProtocol not found, a new ResponseProtocol will be created and returned.

func SetGlobalConfig

func SetGlobalConfig(cfg *Config)

SetGlobalConfig set the global Config.

func SetMetaData

func SetMetaData(ctx context.Context, key string, val []byte)

SetMetaData sets metadata which will be returned to upstream. This method is not thread-safe. Notice: SetMetaData can only be called in the server side rpc entry goroutine, not in goroutine that calls the client.

func Setup

func Setup(cfg *Config) error

Setup registers client config and setups plugins according to the Config.

func SetupClients

func SetupClients(cfg *ClientConfig) error

SetupClients sets up all backends and the wildcard.

func SetupPlugins

func SetupPlugins(cfg plugin.Config) (func() error, error)

SetupPlugins sets up all plugins and returns a function to close them.

func Version

func Version() string

Version returns the version of trpc.

Types

type ClientCodec

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

ClientCodec is an implementation of codec.Codec. Used for trpc clientside codec.

func (*ClientCodec) Decode

func (c *ClientCodec) Decode(msg codec.Msg, rspBuf []byte) (rspBody []byte, err error)

Decode implements codec.Codec. It decodes rspBuf into rspBody.

func (*ClientCodec) Encode

func (c *ClientCodec) Encode(msg codec.Msg, reqBody []byte) (reqBuf []byte, err error)

Encode implements codec.Codec. It encodes reqBody into binary data. New msg will be cloned by client stub.

type ClientConfig

type ClientConfig struct {
	Network        string                  `yaml:"network"`        // Network for all backends. Default is tcp.
	Protocol       string                  `yaml:"protocol"`       // Protocol for all backends. Default is trpc.
	Filter         []string                `yaml:"filter"`         // Filters for all backends.
	StreamFilter   []string                `yaml:"stream_filter"`  // Stream filters for all backends.
	Namespace      string                  `yaml:"namespace"`      // Namespace for all backends.
	Transport      string                  `yaml:"transport"`      // Transport type.
	Timeout        int                     `yaml:"timeout"`        // Timeout in milliseconds.
	Discovery      string                  `yaml:"discovery"`      // Discovery mechanism.
	ServiceRouter  string                  `yaml:"servicerouter"`  // Service router.
	Loadbalance    string                  `yaml:"loadbalance"`    // Load balancing algorithm.
	Circuitbreaker string                  `yaml:"circuitbreaker"` // Circuit breaker configuration.
	Service        []*client.BackendConfig `yaml:"service"`        // Configuration for each individual backend.
}

ClientConfig is the configuration for the client to request backends.

type ClientStreamCodec

type ClientStreamCodec struct {
}

ClientStreamCodec is an implementation of codec.Codec. Used for trpc client streaming codec.

func NewClientStreamCodec

func NewClientStreamCodec() *ClientStreamCodec

NewClientStreamCodec initializes and returns a ClientStreamCodec.

func (*ClientStreamCodec) Decode

func (c *ClientStreamCodec) Decode(msg codec.Msg, rspBuf []byte) ([]byte, error)

Decode implements codec.Codec.

func (*ClientStreamCodec) Encode

func (c *ClientStreamCodec) Encode(msg codec.Msg, reqBuf []byte) ([]byte, error)

Encode implements codec.Codec.

type Config

type Config struct {
	Global struct {
		Namespace     string `yaml:"namespace"`      // Namespace for the configuration.
		EnvName       string `yaml:"env_name"`       // Environment name.
		ContainerName string `yaml:"container_name"` // Container name.
		LocalIP       string `yaml:"local_ip"`       // Local IP address.
		EnableSet     string `yaml:"enable_set"`     // Y/N. Whether to enable Set. Default is N.
		// Full set name with the format: [set name].[set region].[set group name].
		FullSetName string `yaml:"full_set_name"`
		// Size of the read buffer in bytes. <=0 means read buffer disabled. Default value will be used if not set.
		ReadBufferSize *int `yaml:"read_buffer_size,omitempty"`
	}
	Server struct {
		App      string `yaml:"app"`       // Application name.
		Server   string `yaml:"server"`    // Server name.
		BinPath  string `yaml:"bin_path"`  // Binary file path.
		DataPath string `yaml:"data_path"` // Data file path.
		ConfPath string `yaml:"conf_path"` // Configuration file path.
		Admin    struct {
			IP           string      `yaml:"ip"`            // NIC IP to bind, e.g., 127.0.0.1.
			Nic          string      `yaml:"nic"`           // NIC to bind.
			Port         uint16      `yaml:"port"`          // Port to bind, e.g., 80. Default is 9028.
			ReadTimeout  int         `yaml:"read_timeout"`  // Read timeout in milliseconds for admin HTTP server.
			WriteTimeout int         `yaml:"write_timeout"` // Write timeout in milliseconds for admin HTTP server.
			EnableTLS    bool        `yaml:"enable_tls"`    // Whether to enable TLS.
			RPCZ         *RPCZConfig `yaml:"rpcz"`          // RPCZ configuration.
		}
		Transport    string           `yaml:"transport"`     // Transport type.
		Network      string           `yaml:"network"`       // Network type for all services. Default is tcp.
		Protocol     string           `yaml:"protocol"`      // Protocol type for all services. Default is trpc.
		Filter       []string         `yaml:"filter"`        // Filters for all services.
		StreamFilter []string         `yaml:"stream_filter"` // Stream filters for all services.
		Service      []*ServiceConfig `yaml:"service"`       // Configuration for each individual service.
		// Minimum waiting time in milliseconds when closing the server to wait for deregister finish.
		CloseWaitTime int `yaml:"close_wait_time"`
		// Maximum waiting time in milliseconds when closing the server to wait for requests to finish.
		MaxCloseWaitTime int `yaml:"max_close_wait_time"`
		Timeout          int `yaml:"timeout"` // Timeout in milliseconds.
	}
	Client  ClientConfig  `yaml:"client"`  // Client configuration.
	Plugins plugin.Config `yaml:"plugins"` // Plugins configuration.
}

Config is the configuration for trpc, which can be divided into 4 parts: 1. Global config. 2. Server config. 3. Client config. 4. Plugins config.

func GlobalConfig

func GlobalConfig() *Config

GlobalConfig returns the global Config.

func LoadConfig

func LoadConfig(configPath string) (*Config, error)

LoadConfig loads a Config from the config file path.

type FrameHead

type FrameHead struct {
	FrameType       uint8  // type of the frame
	StreamFrameType uint8  // type of the stream frame
	TotalLen        uint32 // total length
	HeaderLen       uint16 // header's length
	StreamID        uint32 // stream id for streaming rpc, request id for unary rpc
	ProtocolVersion uint8  // version of protocol
	FrameReserved   uint8  // reserved bits for further development
}

FrameHead is head of the trpc frame.

type FramerBuilder

type FramerBuilder struct{}

FramerBuilder is an implementation of codec.FramerBuilder. Used for trpc protocol.

func (*FramerBuilder) New

func (fb *FramerBuilder) New(reader io.Reader) codec.Framer

New implements codec.FramerBuilder.

func (*FramerBuilder) Parse

func (fb *FramerBuilder) Parse(rc io.Reader) (vid uint32, buf []byte, err error)

Parse implement multiplexed.FrameParser interface.

type Goer

type Goer interface {
	Go(ctx context.Context, timeout time.Duration, handler func(context.Context)) error
}

Goer is the interface that launches a testable and safe goroutine.

func NewAsyncGoer

func NewAsyncGoer(workerPoolSize int, panicBufLen int, shouldRecover bool) Goer

NewAsyncGoer creates a goer that executes handler asynchronously with a goroutine when Go() is called.

func NewSyncGoer

func NewSyncGoer() Goer

NewSyncGoer creates a goer that executes handler synchronously without cloning ctx when Go() is called. it's usually used for testing.

type RPCZConfig

type RPCZConfig struct {
	Fraction   float64           `yaml:"fraction"`
	Capacity   uint32            `yaml:"capacity"`
	RecordWhen *RecordWhenConfig `yaml:"record_when"`
}

RPCZConfig is the config for rpcz.GlobalRPCZ, and is a field of Config.Admin.

type RecordWhenConfig

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

RecordWhenConfig stores the RecordWhenConfig field of Config.

func (*RecordWhenConfig) UnmarshalYAML

func (c *RecordWhenConfig) UnmarshalYAML(node *yaml.Node) error

UnmarshalYAML customizes RecordWhenConfig's behavior when being unmarshalled from a YAML document.

type ServerCodec

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

ServerCodec is an implementation of codec.Codec. Used for trpc serverside codec.

func (*ServerCodec) Decode

func (s *ServerCodec) Decode(msg codec.Msg, reqBuf []byte) ([]byte, error)

Decode implements codec.Codec. It decodes the reqBuf and updates the msg that already initialized by service handler.

func (*ServerCodec) Encode

func (s *ServerCodec) Encode(msg codec.Msg, rspBody []byte) ([]byte, error)

Encode implements codec.Codec. It encodes the rspBody to binary data and returns it to client.

type ServerStreamCodec

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

ServerStreamCodec is an implementation of codec.Codec. Used for trpc server streaming codec.

func NewServerStreamCodec

func NewServerStreamCodec() *ServerStreamCodec

NewServerStreamCodec initializes and returns a ServerStreamCodec.

func (*ServerStreamCodec) Decode

func (s *ServerStreamCodec) Decode(msg codec.Msg, reqBuf []byte) ([]byte, error)

Decode implements codec.Codec. It decodes the head and the stream frame data.

func (*ServerStreamCodec) Encode

func (s *ServerStreamCodec) Encode(msg codec.Msg, reqBuf []byte) (rspbuf []byte, err error)

Encode implements codec.Codec.

type ServiceConfig

type ServiceConfig struct {
	// Disable request timeout inherited from upstream service.
	DisableRequestTimeout bool   `yaml:"disable_request_timeout"`
	IP                    string `yaml:"ip"` // IP address to listen to.
	// Service name in the format: trpc.app.server.service. Used for naming the service.
	Name string `yaml:"name"`
	Nic  string `yaml:"nic"`  // Network Interface Card (NIC) to listen to. No need to configure.
	Port uint16 `yaml:"port"` // Port to listen to.
	// Address to listen to. If set, ipport will be ignored. Otherwise, ipport will be used.
	Address  string `yaml:"address"`
	Network  string `yaml:"network"`  // Network type like tcp/udp.
	Protocol string `yaml:"protocol"` // Protocol type like trpc.
	// Longest time in milliseconds for a handler to handle a request.
	Timeout int `yaml:"timeout"`
	// Maximum idle time in milliseconds for a server connection. Default is 1 minute.
	Idletime          int      `yaml:"idletime"`
	DisableKeepAlives bool     `yaml:"disable_keep_alives"`    // Disables keep-alives.
	Registry          string   `yaml:"registry"`               // Registry to use, e.g., polaris.
	Filter            []string `yaml:"filter"`                 // Filters for the service.
	StreamFilter      []string `yaml:"stream_filter"`          // Stream filters for the service.
	TLSKey            string   `yaml:"tls_key"`                // Server TLS key.
	TLSCert           string   `yaml:"tls_cert"`               // Server TLS certificate.
	CACert            string   `yaml:"ca_cert"`                // CA certificate to validate client certificate.
	ServerAsync       *bool    `yaml:"server_async,omitempty"` // Whether to enable server asynchronous mode.
	// MaxRoutines is the maximum number of goroutines for server asynchronous mode.
	// Requests exceeding MaxRoutines will be queued. Prolonged overages may lead to OOM!
	// MaxRoutines is not the solution to alleviate server overloading.
	MaxRoutines int    `yaml:"max_routines"`
	Writev      *bool  `yaml:"writev,omitempty"` // Whether to enable writev.
	Transport   string `yaml:"transport"`        // Transport type.
}

ServiceConfig is a configuration for a single service. A server process might have multiple services.