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) GetMyUniqueID() uint64
  • 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 uint64, msg []byte)
  • func (server *Server) Start() (err error)
• type ServerConfig

Constants

const (
	// INITIAL means the Client struct has just been created
	INITIAL clientState = iota + 1
	// 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) GetMyUniqueID

func (client *Client) GetMyUniqueID() uint64

GetMyUniqueID returns the unique ID of 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.

NOTE: It is assumed that ALL Interrupt() routines will eventually return. Failure to do so will cause client.Close() to hang!

type ClientConfig

type ClientConfig struct {
	DNSOrIPAddr              string        // DNS name or IP Address of Server
	Port                     int           // Port of Server
	RootCAx509CertificatePEM []byte        // If TLS...Root certificate; If TCP... nil
	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
	Logger                   *log.Logger   // If nil, defaults to log.New()
}

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
	// AskMyUniqueID is the message sent by the client during it's initial connection
	// to get it's unique client ID.
	AskMyUniqueID
	// ReturnUniqueID is the message sent by the server to client after initial connection.
	// The server is returning the unique ID created for this client.
	ReturnUniqueID
	// PassID is the message sent by the client to identify itself to server.
	// This is used when we are retransmitting.
	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
	// 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 uint64, 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
	DNSOrIPAddr     string          // DNS or 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
	TLSCertificate  tls.Certificate // TLS Certificate to present to Clients (or tls.Certificate{} if using TCP)
	Logger          *log.Logger     // If nil, defaults to log.New()
	// contains filtered or unexported fields
}

ServerConfig is used to configure a retryrpc Server