rpc

README

cc rpc

相对 go rpc 优势

• 使用function接口而非反射, 提高调用效率
• client 支持服务发现
• client 支持连接池, 可以同时连接多个服务端
• client 支持断链重连, 而 go rpc 的client一旦连接断掉后不在重连, 调用Call会直接报错

Documentation

Index

• Constants
• Variables
• type BinaryWire
  • func NewBinaryWire(rwc io.ReadWriteCloser, compress string) (*BinaryWire, error)
  • func (w *BinaryWire) Close() error
  • func (w *BinaryWire) Read(msg *Message) error
  • func (w *BinaryWire) Write(msg *Message) error
• type Client
  • func Dial(connect string) (*Client, error)
  • func DialHTTPPath(network, address, path string) (*Client, error)
  • func NewClient(conn net.Conn, compress string) (*Client, error)
  • func (c *Client) Call(cmd string, input interface{}, result interface{}) error
  • func (c *Client) CallStream(cmd string, input interface{}) (*StreamMessage, error)
  • func (c *Client) Close() error
  • func (c *Client) Ping() error
  • func (c *Client) TargetID() string
• type ClientConfig
• type Codec
• type Compressor
  • func (c *Compressor) Flush() (err error)
  • func (c *Compressor) Read(p []byte) (n int, err error)
  • func (c *Compressor) Write(p []byte) (n int, err error)
• type HandlerFunc
• type HandlerStreamFunc
• type Message
  • func (msg *Message) Decode(value interface{}) error
  • func (msg *Message) Encode(value interface{}) error
• type MessageType
• type Request
• type Server
  • func NewServer() *Server
  • func (s *Server) Handle(name string, f HandlerFunc)
  • func (s *Server) HandleStream(name string, f HandlerStreamFunc)
  • func (s *Server) ServeHTTP(resp http.ResponseWriter, req *http.Request)
  • func (s *Server) SetCodec(codec Codec)
• type ServerSession
  • func NewServerSession(srv *Server, conn io.ReadWriteCloser, compress string) (*ServerSession, error)
  • func (s *ServerSession) Run() error
  • func (s *ServerSession) Stop()
• type ServerStream
• type StreamMessage
  • func NewStreamMessage(root *Message) *StreamMessage
  • func (m StreamMessage) Close() error
  • func (m StreamMessage) Recv(result interface{}) error
  • func (m StreamMessage) Send(data interface{}) error
• type Wire

Constants

const (
	// MagicVersion is the cc rpc protocal version
	MagicVersion = uint16(0x1b01) // cmdb01
)

Variables

var (
	//ErrRWTimeout r/w operation timeout
	ErrRWTimeout       = errors.New("r/w timeout")
	ErrPingTimeout     = errors.New("Ping timeout")
	ErrCommandNotFount = errors.New("Command not found")
	ErrStreamStoped    = errors.New("Stream stoped")
)

Errors define

Types

type BinaryWire

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

BinaryWire implements Wire interface

func NewBinaryWire

func NewBinaryWire(rwc io.ReadWriteCloser, compress string) (*BinaryWire, error)

NewBinaryWire returns a new BinaryWire

func (*BinaryWire) Close

func (w *BinaryWire) Close() error

Close close the wire

func (*BinaryWire) Read

func (w *BinaryWire) Read(msg *Message) error

func (*BinaryWire) Write

func (w *BinaryWire) Write(msg *Message) error

type Client

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

Client replica client

func Dial

func Dial(connect string) (*Client, error)

func DialHTTPPath

func DialHTTPPath(network, address, path string) (*Client, error)

DialHTTPPath connects to an HTTP RPC server at the specified network address and path.

func NewClient

func NewClient(conn net.Conn, compress string) (*Client, error)

NewClient replica client

func (*Client) Call

func (c *Client) Call(cmd string, input interface{}, result interface{}) error

Call replica client

func (*Client) CallStream

func (c *Client) CallStream(cmd string, input interface{}) (*StreamMessage, error)

CallStream replica client

func (*Client) Close

func (c *Client) Close() error

Close replica client

func (*Client) Ping

func (c *Client) Ping() error

Ping replica client

func (*Client) TargetID

func (c *Client) TargetID() string

TargetID operation target ID

type ClientConfig

type ClientConfig struct {
	Address string
}

type Codec

type Codec interface {
	Decode(data []byte, v interface{}) error
	Encode(v interface{}) ([]byte, error)
}

Codec define a codec

var JSONCodec Codec = new(jsonCodec)

JSONCodec implements Codec interface

type Compressor

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

func (*Compressor) Flush

func (c *Compressor) Flush() (err error)

func (*Compressor) Read

func (c *Compressor) Read(p []byte) (n int, err error)

func (*Compressor) Write

func (c *Compressor) Write(p []byte) (n int, err error)

type HandlerFunc

type HandlerFunc func(Request) (interface{}, error)

HandlerFunc define a HandlerFunc

type HandlerStreamFunc

type HandlerStreamFunc func(Request, ServerStream) error

HandlerStreamFunc define a HandlerStreamFunc

type Message

type Message struct {
	Data []byte
	// contains filtered or unexported fields
}

Message define a rpc message

func (*Message) Decode

func (msg *Message) Decode(value interface{}) error

Decode decode the message data

func (*Message) Encode

func (msg *Message) Encode(value interface{}) error

Encode encode the value to message data

type MessageType

type MessageType uint32

MessageType define

const (
	TypeRequest MessageType = iota
	TypeResponse
	TypeStream
	TypeError
	TypeClose
	TypePing
	TypeStreamClose
)

MessageType enumeration

type Request

type Request interface {
	Decode(value interface{}) error
}

Request define a request interface

type Server

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

Server define

func NewServer

func NewServer() *Server

NewServer returns new server

func (*Server) Handle

func (s *Server) Handle(name string, f HandlerFunc)

Handle regist new handler

func (*Server) HandleStream

func (s *Server) HandleStream(name string, f HandlerStreamFunc)

HandleStream regist new stream handler

func (*Server) ServeHTTP

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

ServeHTTP implements http.Handler interface

func (*Server) SetCodec

func (s *Server) SetCodec(codec Codec)

SetCodec set server codec

type ServerSession

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

ServerSession define a server session

func NewServerSession

func NewServerSession(srv *Server, conn io.ReadWriteCloser, compress string) (*ServerSession, error)

NewServerSession returns a new ServerSession

func (*ServerSession) Run

func (s *ServerSession) Run() error

Run run the Serssion

func (*ServerSession) Stop

func (s *ServerSession) Stop()

Stop stop the server session

type ServerStream

type ServerStream interface {
	Recv(result interface{}) error
	Send(data interface{}) error
}

ServerStream define interface

type StreamMessage

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

StreamMessage define

func NewStreamMessage

func NewStreamMessage(root *Message) *StreamMessage

NewStreamMessage returns a new StreamMessage

func (StreamMessage) Close

func (m StreamMessage) Close() error

Close should only call by client

func (StreamMessage) Recv

func (m StreamMessage) Recv(result interface{}) error

Recv receive message

func (StreamMessage) Send

func (m StreamMessage) Send(data interface{}) error

Send send message

type Wire

type Wire interface {
	Write(*Message) error
	Read(*Message) error
	Close() error
}

Wire define a wire