Documentation ¶
Overview ¶
Package client is an interface for an RPC client
Index ¶
- Variables
- func Call(ctx context.Context, request Request, response interface{}, opts ...CallOption) error
- func NewContext(ctx context.Context, c Client) context.Context
- func Publish(ctx context.Context, msg Message, opts ...PublishOption) error
- func RetryAlways(ctx context.Context, req Request, retryCount int, err error) (bool, error)
- func RetryOnError(ctx context.Context, req Request, retryCount int, err error) (bool, error)
- func String() string
- type BackoffFunc
- type Cache
- type CallFunc
- type CallOption
- func WithAddress(a ...string) CallOption
- func WithBackoff(fn BackoffFunc) CallOption
- func WithCache(c time.Duration) CallOption
- func WithCallWrapper(cw ...CallWrapper) CallOption
- func WithDialTimeout(d time.Duration) CallOption
- func WithRequestTimeout(d time.Duration) CallOption
- func WithRetries(i int) CallOption
- func WithRetry(fn RetryFunc) CallOption
- func WithSelectOption(so ...selector.SelectOption) CallOption
- func WithServiceToken() CallOption
- func WithStreamTimeout(d time.Duration) CallOption
- type CallOptions
- type CallWrapper
- type Client
- type Message
- type MessageOption
- type MessageOptions
- type Option
- func Backoff(fn BackoffFunc) Option
- func Broker(b broker.Broker) Option
- func Codec(contentType string, c codec.NewCodec) Option
- func ContentType(ct string) Option
- func DialTimeout(d time.Duration) Option
- func PoolSize(d int) Option
- func PoolTTL(d time.Duration) Option
- func Registry(r registry.Registry) Option
- func RequestTimeout(d time.Duration) Option
- func Retries(i int) Option
- func Retry(fn RetryFunc) Option
- func Selector(s selector.Selector) Option
- func StreamTimeout(d time.Duration) Option
- func Transport(t transport.Transport) Option
- func WithRouter(r Router) Option
- func Wrap(w Wrapper) Option
- func WrapCall(cw ...CallWrapper) Option
- type Options
- type PublishOption
- type PublishOptions
- type Request
- type RequestOption
- type RequestOptions
- type Response
- type RetryFunc
- type Router
- type Stream
- type StreamWrapper
- type Wrapper
Constants ¶
This section is empty.
Variables ¶
var ( // DefaultClient is a default client to use out of the box DefaultClient Client = newRpcClient() // DefaultBackoff is the default backoff function for retries DefaultBackoff = exponentialBackoff // DefaultRetry is the default check-for-retry function for retries DefaultRetry = RetryOnError // DefaultRetries is the default number of times a request is tried DefaultRetries = 1 // DefaultRequestTimeout is the default request timeout DefaultRequestTimeout = time.Second * 5 // DefaultPoolSize sets the connection pool size DefaultPoolSize = 100 // DefaultPoolTTL sets the connection pool ttl DefaultPoolTTL = time.Minute // NewClient returns a new client NewClient func(...Option) Client = newRpcClient )
var ( DefaultContentType = "application/protobuf" DefaultCodecs = map[string]codec.NewCodec{ "application/grpc": grpc.NewCodec, "application/grpc+json": grpc.NewCodec, "application/grpc+proto": grpc.NewCodec, "application/protobuf": proto.NewCodec, "application/json": json.NewCodec, "application/json-rpc": jsonrpc.NewCodec, "application/proto-rpc": protorpc.NewCodec, "application/octet-stream": raw.NewCodec, } )
Functions ¶
func Call ¶
func Call(ctx context.Context, request Request, response interface{}, opts ...CallOption) error
Makes a synchronous call to a service using the default client
func Publish ¶
func Publish(ctx context.Context, msg Message, opts ...PublishOption) error
Publishes a publication using the default client. Using the underlying broker set within the options.
func RetryAlways ¶
RetryAlways always retry on error
func RetryOnError ¶
RetryOnError retries a request on a 500 or timeout error
Types ¶
type BackoffFunc ¶
type Cache ¶
type Cache struct {
// contains filtered or unexported fields
}
Cache for responses
type CallFunc ¶
type CallFunc func(ctx context.Context, node *registry.Node, req Request, rsp interface{}, opts CallOptions) error
CallFunc represents the individual call func
type CallOption ¶
type CallOption func(*CallOptions)
CallOption used by Call or Stream
func WithAddress ¶
func WithAddress(a ...string) CallOption
WithAddress sets the remote addresses to use rather than using service discovery
func WithBackoff ¶
func WithBackoff(fn BackoffFunc) CallOption
WithBackoff is a CallOption which overrides that which set in Options.CallOptions
func WithCache ¶
func WithCache(c time.Duration) CallOption
WithCache is a CallOption which sets the duration the response shoull be cached for
func WithCallWrapper ¶
func WithCallWrapper(cw ...CallWrapper) CallOption
WithCallWrapper is a CallOption which adds to the existing CallFunc wrappers
func WithDialTimeout ¶
func WithDialTimeout(d time.Duration) CallOption
WithDialTimeout is a CallOption which overrides that which set in Options.CallOptions
func WithRequestTimeout ¶
func WithRequestTimeout(d time.Duration) CallOption
WithRequestTimeout is a CallOption which overrides that which set in Options.CallOptions
func WithRetries ¶
func WithRetries(i int) CallOption
WithRetries is a CallOption which overrides that which set in Options.CallOptions
func WithRetry ¶
func WithRetry(fn RetryFunc) CallOption
WithRetry is a CallOption which overrides that which set in Options.CallOptions
func WithSelectOption ¶
func WithSelectOption(so ...selector.SelectOption) CallOption
func WithServiceToken ¶
func WithServiceToken() CallOption
WithServiceToken is a CallOption which overrides the authorization header with the services own auth token
func WithStreamTimeout ¶
func WithStreamTimeout(d time.Duration) CallOption
WithStreamTimeout sets the stream timeout
type CallOptions ¶
type CallOptions struct { SelectOptions []selector.SelectOption // Address of remote hosts Address []string // Backoff func Backoff BackoffFunc // Check if retriable func Retry RetryFunc // Transport Dial Timeout DialTimeout time.Duration // Number of Call attempts Retries int // Request/Response timeout RequestTimeout time.Duration // Stream timeout for the stream StreamTimeout time.Duration // Use the services own auth token ServiceToken bool // Duration to cache the response for CacheExpiry time.Duration // Middleware for low level call func CallWrappers []CallWrapper // Other options for implementations of the interface // can be stored in a context Context context.Context }
type CallWrapper ¶
CallWrapper is a low level wrapper for the CallFunc
type Client ¶
type Client interface { Init(...Option) error Options() Options NewMessage(topic string, msg interface{}, opts ...MessageOption) Message NewRequest(service, endpoint string, req interface{}, reqOpts ...RequestOption) Request Call(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error Stream(ctx context.Context, req Request, opts ...CallOption) (Stream, error) Publish(ctx context.Context, msg Message, opts ...PublishOption) error String() string }
Client is the interface used to make requests to services. It supports Request/Response via Transport and Publishing via the Broker. It also supports bidirectional streaming of requests.
type Message ¶
Message is the interface for publishing asynchronously
func NewMessage ¶
func NewMessage(topic string, payload interface{}, opts ...MessageOption) Message
Creates a new message using the default client
type MessageOption ¶
type MessageOption func(*MessageOptions)
MessageOption used by NewMessage
func WithMessageContentType ¶
func WithMessageContentType(ct string) MessageOption
type MessageOptions ¶
type MessageOptions struct {
ContentType string
}
type Option ¶
type Option func(*Options)
Option used by the Client
func Backoff ¶
func Backoff(fn BackoffFunc) Option
Backoff is used to set the backoff function used when retrying Calls
func RequestTimeout ¶
The request timeout. Should this be a Call Option?
func StreamTimeout ¶
StreamTimeout sets the stream timeout
func WrapCall ¶
func WrapCall(cw ...CallWrapper) Option
Adds a Wrapper to the list of CallFunc wrappers
type Options ¶
type Options struct { // Used to select codec ContentType string // Plugged interfaces Broker broker.Broker Codecs map[string]codec.NewCodec Registry registry.Registry Selector selector.Selector Transport transport.Transport // Router sets the router Router Router // Connection Pool PoolSize int PoolTTL time.Duration // Response cache Cache *Cache // Middleware for client Wrappers []Wrapper // Default Call Options CallOptions CallOptions // Other options for implementations of the interface // can be stored in a context Context context.Context }
func NewOptions ¶
type PublishOption ¶
type PublishOption func(*PublishOptions)
PublishOption used by Publish
func PublishContext ¶
func PublishContext(ctx context.Context) PublishOption
PublishContext sets the context in publish options
func WithExchange ¶
func WithExchange(e string) PublishOption
WithExchange sets the exchange to route a message through
type PublishOptions ¶
type Request ¶
type Request interface { // The service to call Service() string // The action to take Method() string // The endpoint to invoke Endpoint() string // The content type ContentType() string // The unencoded request body Body() interface{} // Write to the encoded request writer. This is nil before a call is made Codec() codec.Writer // indicates whether the request will be a streaming one rather than unary Stream() bool }
Request is the interface for a synchronous request used by Call or Stream
func NewRequest ¶
func NewRequest(service, endpoint string, request interface{}, reqOpts ...RequestOption) Request
Creates a new request using the default client. Content Type will be set to the default within options and use the appropriate codec
type RequestOption ¶
type RequestOption func(*RequestOptions)
RequestOption used by NewRequest
func StreamingRequest ¶
func StreamingRequest() RequestOption
func WithContentType ¶
func WithContentType(ct string) RequestOption
type RequestOptions ¶
type Response ¶
type Response interface { // Read the response Codec() codec.Reader // read the header Header() map[string]string // Read the undecoded response Read() ([]byte, error) }
Response is the response received from a service
type RetryFunc ¶
note that returning either false or a non-nil error will result in the call not being retried
type Stream ¶
type Stream interface { // Context for the stream Context() context.Context // The request made Request() Request // The response read Response() Response // Send will encode and send a request Send(interface{}) error // Recv will decode and read a response Recv(interface{}) error // Error returns the stream error Error() error // Close closes the stream Close() error }
Stream is the inteface for a bidirectional synchronous stream
type StreamWrapper ¶
StreamWrapper wraps a Stream and returns the equivalent