retryrpc

Documentation

Overview

Package retryrpc provides a client and server RPC model which survives lost connections on either the client or the server.

Index

• Constants
• type Client
  • func NewClient(config *ClientConfig) (client *Client, err error)
  • func (client *Client) Close()
  • func (client *Client) GetStatsGroupName() (s string)
  • func (client *Client) Send(method string, request interface{}, reply interface{}) (err error)
• type ClientCallbacks
• type ClientConfig
• type MsgType
• type PayloadProtocols
• type Server
  • func NewServer(config *ServerConfig) *Server
  • func (server *Server) Close()
  • func (server *Server) CloseClientConn()
  • func (server *Server) CompletedCnt() (totalCnt int)
  • func (server *Server) Register(retrySvr interface{}) (err error)
  • func (server *Server) Run()
  • func (server *Server) SendCallback(clientID string, msg []byte)
  • func (server *Server) Start() (err error)
• type ServerConfig
• type ServerCreds

Constants

const (
	// INITIAL means the Client struct has just been created
	INITIAL clientState = iota + 1
	// DISCONNECTED means the Client has lost the connection to the server
	DISCONNECTED
	// CONNECTED means the Client is connected to the server
	CONNECTED
	// RETRANSMITTING means a goroutine is in the middle of recovering
	// from a loss of a connection with the server
	RETRANSMITTING
)

const (
	ConnectionRetryDelayMultiplier = 2
	ConnectionRetryInitialDelay    = 100 * time.Millisecond
	ConnectionRetryLimit           = 8
)

Types

type Client

type Client struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Client tracking structure

func NewClient

func NewClient(config *ClientConfig) (client *Client, err error)

NewClient returns a Client structure

If the server wants to send an async message to the client it uses the Interrupt method defined in cb

NOTE: It is assumed that if a client calls NewClient(), it will always use a unique myUniqueID. Otherwise, the server may have old entries.

TODO - purge cache of old entries on server and/or use different starting point for requestID.

func (*Client) Close

func (client *Client) Close()

Close gracefully shuts down the client

func (*Client) GetStatsGroupName

func (client *Client) GetStatsGroupName() (s string)

GetStatsGroupName returns the bucketstats GroupName for this client

func (*Client) Send

func (client *Client) Send(method string, request interface{}, reply interface{}) (err error)

Send the request and block until it has completed

type ClientCallbacks

type ClientCallbacks interface {
	Interrupt(payload []byte)
}

ClientCallbacks contains the methods required when supporting callbacks from the Server.

type ClientConfig

type ClientConfig struct {
	MyUniqueID               string
	IPAddr                   string        // IP Address of Server
	Port                     int           // Port of Server
	RootCAx509CertificatePEM []byte        // Root certificate
	Callbacks                interface{}   // Structure implementing ClientCallbacks
	DeadlineIO               time.Duration // How long I/Os on sockets wait even if idle
	KeepAlivePeriod          time.Duration // How frequently a KEEPALIVE is sent
}

ClientConfig is used to configure a retryrpc Client

type MsgType

type MsgType uint16

MsgType is the type of message being sent

const (
	// RPC represents an RPC from client to server
	RPC MsgType = iota + 1
	// Upcall represents an upcall from server to client
	Upcall
	// PassID is the message sent by the client to identify itself to server
	PassID
)

type PayloadProtocols

type PayloadProtocols int

PayloadProtocols defines the supported protocols for the payload

const (
	JSON PayloadProtocols = 1
)

Support payload protocols

type Server

type Server struct {
	sync.Mutex

	Creds *ServerCreds
	// contains filtered or unexported fields
}

Server tracks the state of the server

func NewServer

func NewServer(config *ServerConfig) *Server

NewServer creates the Server object

func (*Server) Close

func (server *Server) Close()

Close stops the server

func (*Server) CloseClientConn

func (server *Server) CloseClientConn()

CloseClientConn - This is debug code to cause some connections to be closed It is called from a stress test case to cause retransmits

func (*Server) CompletedCnt

func (server *Server) CompletedCnt() (totalCnt int)

CompletedCnt returns count of pendingRequests

This is only useful for testing.

func (*Server) Register

func (server *Server) Register(retrySvr interface{}) (err error)

Register creates the map of server methods

func (*Server) Run

func (server *Server) Run()

Run server loop, accept connections, read request, run RPC method and return the results.

func (*Server) SendCallback

func (server *Server) SendCallback(clientID string, msg []byte)

SendCallback sends a message to clientID so that clientID contacts the RPC server.

The assumption is that this callback only gets called when the server has an async message for the client

The message is "best effort" - if we fail to write on socket then the message is silently dropped on floor.

func (*Server) Start

func (server *Server) Start() (err error)

Start listener

type ServerConfig

type ServerConfig struct {
	LongTrim        time.Duration // How long the results of an RPC are stored on a Server before removed
	ShortTrim       time.Duration // How frequently completed and ACKed RPCs results are removed from Server
	IPAddr          string        // IP Address that Server uses to listen
	Port            int           // Port that Server uses to listen
	DeadlineIO      time.Duration // How long I/Os on sockets wait even if idle
	KeepAlivePeriod time.Duration // How frequently a KEEPALIVE is sent
	// contains filtered or unexported fields
}

ServerConfig is used to configure a retryrpc Server

type ServerCreds

type ServerCreds struct {
	RootCAx509CertificatePEM []byte
	// contains filtered or unexported fields
}

ServerCreds tracks the root CA and the server CA