rpcclient

README

CGRateS RPCClient

A Go RPC client library supporting multiple codecs and pooling strategies. Part of the CGRateS project. The library provides thread-safe operations, automatic failover with reconnection support, and optional TLS encryption.

Installation

go get github.com/cgrates/rpcclient

Features

• Multiple codec support:
  • Standard RPC: JSON (*json), GOB (*gob), HTTP JSON (*http_jsonrpc)
  • Bi-directional RPC: JSON (*birpc_json), GOB (*birpc_gob)
  • Internal direct calls (*internal, *birpc_internal) for objects implementing birpc.ClientConnector
• Connection pooling with various dispatch strategies
• Automatic failover for network errors, timeouts, and missing service errors

Pool Strategies

• *first: Uses first available connection, fails over on network/timeout/missing service errors
• *next: Round-robin between connections with same failover as *first
• *random: Random connection selection with same failover as *first
• *first_positive: Tries connections in order until getting any successful response (no error)
• *first_positive_async: Async version of *first_positive
• *broadcast: Sends to all connections, returns first successful response
• *broadcast_sync: Sends to all, waits for completion, logs errors that wouldn't trigger failover in *first
• *broadcast_async: Sends to all without waiting for responses or error handling
• *parallel: Connection pool that creates and reuses connections up to a maximum limit, either pre-initialized or on demand

Support

Join CGRateS on Google Groups here.

License

RpcClient is released under the MIT License.

Copyright (C) ITsysCOM GmbH. All Rights Reserved.

Documentation

Index

• Constants
• Variables
• func ShouldFailover(err error) bool
• type HTTPjsonRPCClient
  • func (client *HTTPjsonRPCClient) Call(ctx *context.Context, serviceMethod string, args interface{}, ...) (err error)
• type JSONrpcResponse
• type RPCClient
  • func NewRPCClient(ctx *context.Context, transport, addr string, tls bool, ...) (rpcClient *RPCClient, err error)
  • func (client *RPCClient) Call(ctx *context.Context, serviceMethod string, args interface{}, ...) (err error)
• type RPCCloner
• type RPCParallelClientPool
  • func NewRPCParallelClientPool(ctx *context.Context, transport, addr string, tls bool, ...) (rpcClient *RPCParallelClientPool, err error)
  • func (pool *RPCParallelClientPool) Call(ctx *context.Context, serviceMethod string, args interface{}, ...) (err error)
• type RPCPool
  • func NewRPCPool(transmissionType string, replyTimeout time.Duration) *RPCPool
  • func (pool *RPCPool) AddClient(rcc birpc.ClientConnector)
  • func (pool *RPCPool) Call(ctx *context.Context, serviceMethod string, args interface{}, ...) (err error)

Constants

const (
	JSONrpc     = "*json"
	HTTPjson    = "*http_jsonrpc"
	GOBrpc      = "*gob"
	InternalRPC = "*internal"

	BiRPCJSON     = "*birpc_json"
	BiRPCGOB      = "*birpc_gob"
	BiRPCInternal = "*birpc_internal"
)

Constants to define the codec for RpcClient

const (
	PoolFirst              = "*first"
	PoolAsync              = "*async"
	PoolRandom             = "*random"
	PoolNext               = "*next"
	PoolFirstPositive      = "*first_positive"
	PoolFirstPositiveAsync = "*first_positive_async"
	PoolParallel           = "*parallel"
	PoolBroadcast          = "*broadcast"
	PoolBroadcastSync      = "*broadcast_sync"
	PoolBroadcastAsync     = "*broadcast_async"
)

Constants to define the strategy for RpcClientPool

Variables

var (
	ErrReqUnsynchronized       = errors.New("REQ_UNSYNCHRONIZED")
	ErrUnsupporteServiceMethod = errors.New("UNSUPPORTED_SERVICE_METHOD")
	ErrDisconnected            = errors.New("DISCONNECTED")
	ErrFailedReconnect         = errors.New("FAILED_RECONNECT")
	ErrInternallyDisconnected  = errors.New("INTERNALLY_DISCONNECTED")
	ErrUnsupportedCodec        = errors.New("UNSUPPORTED_CODEC")
	ErrPartiallyExecuted       = errors.New("PARTIALLY_EXECUTED")
	ErrUnsupportedBiRPC        = errors.New("UNSUPPORTED_BIRPC")
)

Errors that library may return back

Functions

func ShouldFailover

func ShouldFailover(err error) bool

ShouldFailover decides whether to failover to the next connection in the pool.

Types

type HTTPjsonRPCClient

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

HTTPjsonRPCClient only for the rpc over http

func (*HTTPjsonRPCClient) Call

func (client *HTTPjsonRPCClient) Call(ctx *context.Context, serviceMethod string, args interface{}, reply interface{}) (err error)

Call the method needed to implement ClientConnector

type JSONrpcResponse

type JSONrpcResponse struct {
	ID     uint64
	Result *json.RawMessage
	Error  interface{}
}

JSONrpcResponse is the response received from JSON RPC

type RPCClient

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

RPCClient implements ClientConnector

func NewRPCClient

func NewRPCClient(ctx *context.Context, transport, addr string, tls bool, keyPath, certPath,
	caPath string, connectAttempts, reconnects int, maxReconnectInterval time.Duration,
	delayFunc func(time.Duration, time.Duration) func() time.Duration, connTimeout,
	replyTimeout time.Duration, codec string, internalChan chan birpc.ClientConnector,
	lazyConnect bool, biRPCClient interface{}) (rpcClient *RPCClient, err error)

NewRPCClient creates a client based on the config

func (*RPCClient) Call

func (client *RPCClient) Call(ctx *context.Context, serviceMethod string, args interface{}, reply interface{}) (err error)

Call the method needed to implement ClientConnector

type RPCCloner

type RPCCloner interface {
	RPCClone() (interface{}, error)
}

RPCCloner is an interface for objects to clone parts of themselves which are affected by concurrency at the time of RPC call

type RPCParallelClientPool

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

RPCParallelClientPool implements ClientConnector

func NewRPCParallelClientPool

func NewRPCParallelClientPool(ctx *context.Context, transport, addr string, tls bool, keyPath,
	certPath, caPath string, connectAttempts, reconnects int, maxReconnectInterval time.Duration,
	delayFunc func(time.Duration, time.Duration) func() time.Duration, connTimeout,
	replyTimeout time.Duration, codec string, internalChan chan birpc.ClientConnector, maxCounter int64,
	initConns bool, biRPCClient interface{}) (rpcClient *RPCParallelClientPool, err error)

NewRPCParallelClientPool returns a new RPCParallelClientPool

func (*RPCParallelClientPool) Call

func (pool *RPCParallelClientPool) Call(ctx *context.Context, serviceMethod string, args interface{}, reply interface{}) (err error)

Call the method needed to implement birpc.ClientConnector

type RPCPool

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

RPCPool is a pool of connections

func NewRPCPool

func NewRPCPool(transmissionType string, replyTimeout time.Duration) *RPCPool

NewRPCPool creates RPCPool

func (*RPCPool) AddClient

func (pool *RPCPool) AddClient(rcc birpc.ClientConnector)

AddClient adds a client connection in the pool

func (*RPCPool) Call

func (pool *RPCPool) Call(ctx *context.Context, serviceMethod string, args interface{}, reply interface{}) (err error)

Call the method needed to implement birpc.ClientConnector