server

type CMuxPlugin ¶ added in v1.6.3

type CMuxPlugin interface {
	MuxMatch(m cmux.CMux)
}

CMuxPlugin lets a plugin register protocol matchers when the server uses cmux to multiplex several protocols (for example HTTP and rpcx) on one listening port. MuxMatch is consulted once at startup.

type CORSOptions ¶ added in v1.4.1

type CORSOptions = cors.Options

func AllowAllCORSOptions() *CORSOptions

type Context ¶ added in v1.6.9

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

Context represents a rpcx FastCall context.

func NewContext(ctx *share.Context, conn net.Conn, req *protocol.Message, async bool) *Context

func (ctx *Context) Bind(v interface{}) error

func (ctx *Context) DeleteKey(key interface{})

func (ctx *Context) Get(key interface{}) interface{}

func (ctx *Context) Metadata() map[string]string

func (ctx *Context) Payload() []byte

func (ctx *Context) ServiceMethod() string

func (ctx *Context) ServicePath() string

func (ctx *Context) SetValue(key, val interface{})

func (ctx *Context) Write(v interface{}) error

func (ctx *Context) WriteError(err error) error

type DownloadFileHandler ¶ added in v1.6.2

type DownloadFileHandler func(conn net.Conn, args *share.DownloadFileArgs)

DownloadFileHandler handles downloading file. Must close the connection after it finished.

type FileTransfer ¶ added in v1.6.2

type FileTransfer struct {
	Addr          string
	AdvertiseAddr string
	// contains filtered or unexported fields
}

FileTransfer support transfer files from clients. It registers a file transfer service and listens a on the given port. Clients will invokes this service to get the token and send the token and the file to this port.

func NewFileTransfer(addr string, handler FileTransferHandler, downloadFileHandler DownloadFileHandler, waitNum int) *FileTransfer

func (s *FileTransfer) Start() error

func (s *FileTransfer) Stop() error

type FileTransferHandler ¶ added in v1.6.2

type FileTransferHandler func(conn net.Conn, args *share.FileTransferArgs)

FileTransferHandler handles uploading file. Must close the connection after it finished.

type FileTransferService ¶ added in v1.6.2

type FileTransferService struct {
	FileTransfer *FileTransfer
}

func (s *FileTransferService) DownloadFile(ctx context.Context, args *share.DownloadFileArgs, reply *share.FileTransferReply) error

func (s *FileTransferService) TransferFile(ctx context.Context, args *share.FileTransferArgs, reply *share.FileTransferReply) error

type Handler ¶ added in v1.6.9

type Handler func(ctx *Context) error

type HeartbeatPlugin ¶ added in v1.6.2

type HeartbeatPlugin interface {
	HeartbeatRequest(ctx context.Context, req *protocol.Message) error
}

HeartbeatPlugin is invoked when an incoming message is a heartbeat, instead of the normal request-handling path.

HeartbeatRequest receives the ctx and the heartbeat message req. Returning a non-nil error is propagated as the heartbeat handling error.

type ID ¶ added in v1.4.1

type ID struct {
	Name   string
	Number int64
}

ID is a jsonrpcRequest identifier. Only one of either the Name or Number members will be set, using the number form if the Name is the empty string.

func (id *ID) MarshalJSON() ([]byte, error)

func (id *ID) String() string

func (id *ID) UnmarshalJSON(data []byte) error

type JSONRPCError ¶ added in v1.4.1

type JSONRPCError struct {
	// Code is an error code indicating the type of failure.
	Code int64 `json:"code"`
	// Message is a short description of the error.
	Message string `json:"message"`
	// Data is optional structured data containing additional information about the error.
	Data *json.RawMessage `json:"data"`
}

JSONRPCError represents a structured error in a jsonrpcRespone.

func (err *JSONRPCError) JSONRPCError() string

type MakeListener ¶

type MakeListener func(s *Server, address string) (ln net.Listener, err error)

MakeListener defines a listener generator.

type OptionFn ¶

type OptionFn func(*Server)

OptionFn configures options of server.

func WithAsyncWrite() OptionFn

func WithCustomPool(pool WorkerPool) OptionFn

func WithPool(maxWorkers, maxCapacity int, options ...pond.Option) OptionFn

func WithReadTimeout(readTimeout time.Duration) OptionFn

func WithTCPKeepAlivePeriod(period time.Duration) OptionFn

func WithTLSConfig(cfg *tls.Config) OptionFn

func WithWriteTimeout(writeTimeout time.Duration) OptionFn

type Plugin ¶

type Plugin interface{}

Plugin is the server plugin interface.

A plugin is any value; rpcx discovers what a plugin can do by checking, at each extension point, whether the plugin also implements the corresponding XxxPlugin interface below (via a type assertion). So one plugin struct can implement several of these interfaces and hook into several points at once.

Plugins are invoked in the order they were added with PluginContainer.Add. Within a single extension point, every registered plugin is called in that order; for the request-path hooks, if any plugin returns a non-nil error the container stops and returns that error immediately (the remaining plugins for that point are skipped). The registration hooks (DoRegister / DoRegisterFunction / DoUnregister) are the exception: they collect errors from all plugins into a MultiError instead of short-circuiting.

Lifecycle order over a server's life and a single request:

On registering a service:
  RegisterPlugin.Register / RegisterFunctionPlugin.RegisterFunction

When a TCP-style connection is accepted:
  PostConnAcceptPlugin.HandleConnAccept

For each request on that connection:
  PreReadRequestPlugin.PreReadRequest
  (rpcx reads the request off the wire)
  PostReadRequestPlugin.PostReadRequest
  PreHandleRequestPlugin.PreHandleRequest
  PreCallPlugin.PreCall
  (the service method runs)
  PostCallPlugin.PostCall
  PreWriteResponsePlugin.PreWriteResponse
  (rpcx writes the response to the wire)
  PostWriteResponsePlugin.PostWriteResponse

When the connection closes:
  PostConnClosePlugin.HandleConnClose

PreWriteRequestPlugin / PostWriteRequestPlugin wrap the server writing a message it initiates (for example a server-push message), not the normal response path above. HeartbeatPlugin.HeartbeatRequest fires instead of the handle path when an incoming message is a heartbeat. PostHTTPRequestPlugin applies to the HTTP gateway, and CMuxPlugin.MuxMatch is consulted once at startup when a cmux is used to multiplex protocols on one port.

type PluginContainer ¶

type PluginContainer interface {
	Add(plugin Plugin)
	Remove(plugin Plugin)
	All() []Plugin

	DoRegister(name string, rcvr interface{}, metadata string) error
	DoRegisterFunction(serviceName, fname string, fn interface{}, metadata string) error
	DoUnregister(name string) error

	DoPostConnAccept(net.Conn) (net.Conn, bool)
	DoPostConnClose(net.Conn) bool

	DoPreReadRequest(ctx context.Context) error
	DoPostReadRequest(ctx context.Context, r *protocol.Message, e error) error
	DoPostHTTPRequest(ctx context.Context, r *http.Request, params httprouter.Params) error

	DoPreHandleRequest(ctx context.Context, req *protocol.Message) error
	DoPreCall(ctx context.Context, serviceName, methodName string, args interface{}) (interface{}, error)
	DoPostCall(ctx context.Context, serviceName, methodName string, args, reply interface{}, err error) (interface{}, error)

	DoPreWriteResponse(context.Context, *protocol.Message, *protocol.Message, error) error
	DoPostWriteResponse(context.Context, *protocol.Message, *protocol.Message, error) error

	DoPreWriteRequest(ctx context.Context) error
	DoPostWriteRequest(ctx context.Context, r *protocol.Message, e error) error

	DoHeartbeatRequest(ctx context.Context, req *protocol.Message) error

	MuxMatch(m cmux.CMux)
}

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

type PostCallPlugin ¶ added in v1.6.2

type PostCallPlugin interface {
	PostCall(ctx context.Context, serviceName, methodName string, args, reply interface{}, err error) (interface{}, error)
}

PostCallPlugin is invoked immediately after the service method returns. It can inspect or replace the reply, or observe the method's error.

PostCall receives the ctx, the service and method names, the args, the reply produced by the method, and the method's err. It returns the reply to actually send back (return the same value to leave it unchanged) and an error; a non-nil error replaces the outcome for this call.

type PostConnAcceptPlugin ¶

type PostConnAcceptPlugin interface {
	HandleConnAccept(net.Conn) (net.Conn, bool)
}

PostConnAcceptPlugin is invoked right after a new connection is accepted, before any request is read. It is the place to wrap or inspect the raw connection (TLS bookkeeping, rate limiting, connection counting).

HandleConnAccept receives the accepted net.Conn and returns the conn to use going forward (possibly wrapped) and a bool. If the bool is false, rpcx treats the connection as rejected: the container closes it and stops calling subsequent PostConnAcceptPlugins for this conn.

type PostConnClosePlugin ¶ added in v1.4.1

type PostConnClosePlugin interface {
	HandleConnClose(net.Conn) bool
}

PostConnClosePlugin is invoked after a connection has been closed, for cleanup or metrics.

HandleConnClose receives the closed net.Conn. Returning false stops the container from calling subsequent PostConnClosePlugins for this conn.

type PostHTTPRequestPlugin ¶ added in v1.8.8

type PostHTTPRequestPlugin interface {
	PostHTTPRequest(ctx context.Context, r *http.Request, params httprouter.Params) error
}

PostHTTPRequestPlugin is invoked for requests arriving through the HTTP gateway, after the HTTP request is read.

PostHTTPRequest receives the ctx, the *http.Request, and the matched router params. Returning a non-nil error aborts handling.

type PostReadRequestPlugin ¶

type PostReadRequestPlugin interface {
	PostReadRequest(ctx context.Context, r *protocol.Message, e error) error
}

PostReadRequestPlugin is invoked right after a request message has been read (or a read error occurred).

PostReadRequest receives the ctx, the decoded request message r, and the read error e (nil on success). Returning a non-nil error aborts further processing of this request.

type PostWriteRequestPlugin ¶

type PostWriteRequestPlugin interface {
	PostWriteRequest(ctx context.Context, r *protocol.Message, e error) error
}

PostWriteRequestPlugin is invoked after such a server-initiated message has been written.

PostWriteRequest receives the ctx, the message r that was written, and the write error e (nil on success).

type PostWriteResponsePlugin ¶

type PostWriteResponsePlugin interface {
	PostWriteResponse(context.Context, *protocol.Message, *protocol.Message, error) error
}

PostWriteResponsePlugin is invoked right after the response has been written, for metrics or cleanup.

PostWriteResponse receives the ctx, the request message, the response message that was written, and the write error (nil on success).

type PreCallPlugin ¶ added in v1.6.2

type PreCallPlugin interface {
	PreCall(ctx context.Context, serviceName, methodName string, args interface{}) (interface{}, error)
}

PreCallPlugin is invoked immediately before the resolved service method runs. It can inspect or replace the decoded arguments.

PreCall receives the ctx, the service and method names, and the decoded args. It returns the args to actually pass to the method (return the same value to leave them unchanged) and an error; a non-nil error skips the call and propagates the error.

type PreHandleRequestPlugin ¶ added in v1.4.1

type PreHandleRequestPlugin interface {
	PreHandleRequest(ctx context.Context, r *protocol.Message) error
}

PreHandleRequestPlugin is invoked after the request is read but before the service method is located and called.

PreHandleRequest receives the ctx and the request message r. Returning a non-nil error aborts handling of this request.

type PreReadRequestPlugin ¶

type PreReadRequestPlugin interface {
	PreReadRequest(ctx context.Context) error
}

PreReadRequestPlugin is invoked before rpcx reads the next request message from the connection.

PreReadRequest receives the request ctx. Returning a non-nil error aborts the read for this request.

type PreWriteRequestPlugin ¶

type PreWriteRequestPlugin interface {
	PreWriteRequest(ctx context.Context) error
}

PreWriteRequestPlugin is invoked before the server writes a message it initiates itself (such as a server-to-client push), not the normal response path.

PreWriteRequest receives the ctx. Returning a non-nil error aborts the write.

type PreWriteResponsePlugin ¶

type PreWriteResponsePlugin interface {
	PreWriteResponse(context.Context, *protocol.Message, *protocol.Message, error) error
}

PreWriteResponsePlugin is invoked just before the response is written to the connection.

PreWriteResponse receives the ctx, the request message, the response message about to be written, and the handler error so far. Returning a non-nil error aborts writing the response.

type RegisterFunctionPlugin ¶

type RegisterFunctionPlugin interface {
	RegisterFunction(serviceName, fname string, fn interface{}, metadata string) error
}

RegisterFunctionPlugin is invoked when a bare function (not a method on a service struct) is registered via Server.RegisterFunction / RegisterFunctionName.

RegisterFunction receives the service (path) name, the function name, the function value, and its metadata. Errors are collected into a MultiError.

type RegisterPlugin ¶

type RegisterPlugin interface {
	Register(name string, rcvr interface{}, metadata string) error
	Unregister(name string) error
}

RegisterPlugin is invoked when a service is registered or unregistered on the server (Server.Register / RegisterName / UnregisterAll).

Register receives the service name, the receiver value being registered, and its metadata string; a typical implementation publishes the service to a registry such as etcd/consul/zookeeper. Unregister receives the service name and removes it. Returning an error is collected into a MultiError alongside other plugins' errors rather than aborting the rest.

type Reset ¶

type Reset interface {
	Reset()
}

Reset defines Reset method for pooled object.

type RpcServiceInternalError ¶ added in v1.8.32

type RpcServiceInternalError struct {
	Err    string
	Method string
	Argv   interface{}
	// contains filtered or unexported fields
}

RpcServiceError represents an error that is case by service implementation.

func (e RpcServiceInternalError) Error() string

func (e RpcServiceInternalError) String() string

type Server ¶

type Server struct {
	DisableHTTPGateway bool // disable http invoke or not.
	DisableJSONRPC     bool // disable json rpc or not.
	AsyncWrite         bool // set true if your server only serves few clients

	Plugins PluginContainer

	// AuthFunc can be used to auth.
	AuthFunc func(ctx context.Context, req *protocol.Message, token string) error

	// HandleServiceError is used to get all service errors. You can use it write logs or others.
	HandleServiceError func(error)

	// ServerErrorFunc is a customized error handlers and you can use it to return customized error strings to clients.
	// If not set, it use err.Error()
	ServerErrorFunc func(res *protocol.Message, err error) string

	// The server is started.
	Started chan struct{}
	// contains filtered or unexported fields
}

Server is rpcx server that use TCP or UDP.

func NewServer(options ...OptionFn) *Server

func (s *Server) ActiveClientConn() []net.Conn

func (s *Server) AddHandler(servicePath, serviceMethod string, handler func(*Context) error)

func (s *Server) Address() net.Addr

func (s *Server) Close() error

func (s *Server) EnableFileTransfer(serviceName string, fileTransfer *FileTransfer)

func (s *Server) EnableStreamService(serviceName string, streamService *StreamService)

func (s *Server) ListServices() []string

func (s *Server) Register(rcvr interface{}, metadata string) error

func (s *Server) RegisterFunction(servicePath string, fn interface{}, metadata string) error

func (s *Server) RegisterFunctionName(servicePath string, name string, fn interface{}, metadata string) error

func (s *Server) RegisterName(name string, rcvr interface{}, metadata string) error

func (s *Server) RegisterNameWithMethods(name string, rcvr interface{}, methods []string, metadata string) error

func (s *Server) RegisterOnRestart(f func(s *Server))

func (s *Server) RegisterOnShutdown(f func(s *Server))

func (s *Server) RegisterWithMethods(rcvr interface{}, methods []string, metadata string) error

func (s *Server) Restart(ctx context.Context) error

func (s *Server) SendMessage(conn net.Conn, servicePath, serviceMethod string, metadata map[string]string, data []byte) error

ctx.Value(RemoteConnContextKey)

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

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

func (s *Server) ServeListener(network string, ln net.Listener) (err error)

func (s *Server) ServeWS(conn *websocket.Conn)

func (s *Server) SetCORS(options *CORSOptions)

cors.Options{
	AllowedOrigins:   []string{"foo.com"},
	AllowedMethods:   []string{http.MethodGet, http.MethodPost, http.MethodDelete},
	AllowCredentials: true,
}

func (s *Server) Shutdown(ctx context.Context) error

func (s *Server) UnregisterAll() error

func (s *Server) UpdateHandler(router map[string]Handler)

type StreamAcceptor ¶ added in v1.6.2

type StreamAcceptor func(ctx context.Context, args *share.StreamServiceArgs) bool

StreamAcceptor accepts connection from clients or not. You can use it to validate clients and determine if accept or drop the connection.

type StreamHandler ¶ added in v1.6.2

type StreamHandler func(conn net.Conn, args *share.StreamServiceArgs)

StreamHandler handles a streaming connection with client.

type StreamService ¶ added in v1.6.2

type StreamService struct {
	Addr          string
	AdvertiseAddr string
	// contains filtered or unexported fields
}

StreamService support streaming between clients and server. It registers a streaming service and listens on the given port. Clients will invokes this service to get the token and send the token and begin to stream.

func NewStreamService(addr string, streamHandler StreamHandler, acceptor StreamAcceptor, waitNum int) *StreamService

func (s *StreamService) Start() error

func (s *StreamService) Stop() error

func (s *StreamService) Stream(ctx context.Context, args *share.StreamServiceArgs, reply *share.StreamServiceReply) error

type VersionTag ¶ added in v1.4.1

type VersionTag struct{}

VersionTag is a special 0 sized struct that encodes as the jsonrpc version tag. It will fail during decode if it is not the correct version tag in the stream.

func (VersionTag) MarshalJSON() ([]byte, error)

func (VersionTag) UnmarshalJSON(data []byte) error

type WorkerPool ¶ added in v1.7.12

type WorkerPool interface {
	Submit(task func())
	StopAndWaitFor(deadline time.Duration)
	Stop() context.Context
	StopAndWait()
}