sevenhill/QSfera
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bump reva deps (#8412)

Browse Source 
* bump dependencies

Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de>

* bump reva and add config options

Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de>

---------

Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de>
This commit is contained in:
Jörn Friedrich Dreyer
2024-02-21 10:20:36 +01:00
committed by GitHub
parent c92ebf4b46
commit 5ed57cc09a
490 changed files with 19128 additions and 11161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vendor/go-micro.dev/v4/client/cache.go
Vendored
+11 -7
View File
@@ -8,36 +8,39 @@ import (
"time"
cache "github.com/patrickmn/go-cache"
"go-micro.dev/v4/metadata"
"go-micro.dev/v4/transport/headers"
)
// NewCache returns an initialised cache.
// NewCache returns an initialized cache.
func NewCache() *Cache {
return &Cache{
cache: cache.New(cache.NoExpiration, 30*time.Second),
}
}
// Cache for responses
// Cache for responses.
type Cache struct {
cache *cache.Cache
}
// Get a response from the cache
// Get a response from the cache.
func (c *Cache) Get(ctx context.Context, req *Request) (interface{}, bool) {
return c.cache.Get(key(ctx, req))
}
// Set a response in the cache
// Set a response in the cache.
func (c *Cache) Set(ctx context.Context, req *Request, rsp interface{}, expiry time.Duration) {
c.cache.Set(key(ctx, req), rsp, expiry)
}
// List the key value pairs in the cache
// List the key value pairs in the cache.
func (c *Cache) List() map[string]string {
items := c.cache.Items()
rsp := make(map[string]string, len(items))
for k, v := range items {
bytes, _ := json.Marshal(v.Object)
rsp[k] = string(bytes)
@@ -46,9 +49,9 @@ func (c *Cache) List() map[string]string {
return rsp
}
// key returns a hash for the context and request
// key returns a hash for the context and request.
func key(ctx context.Context, req *Request) string {
ns, _ := metadata.Get(ctx, "Micro-Namespace")
ns, _ := metadata.Get(ctx, headers.Namespace)
bytes, _ := json.Marshal(map[string]interface{}{
"namespace": ns,
@@ -62,5 +65,6 @@ func key(ctx context.Context, req *Request) string {
h := fnv.New64()
h.Write(bytes)
return fmt.Sprintf("%x", h.Sum(nil))
}
vendor/go-micro.dev/v4/client/client.go
Vendored
+21 -35
View File
@@ -3,11 +3,17 @@ package client
import (
"context"
"time"
"go-micro.dev/v4/codec"
)
var (
// NewClient returns a new client.
NewClient func(...Option) Client = newRPCClient
// DefaultClient is a default client to use out of the box.
DefaultClient Client = newRPCClient()
)
// 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.
@@ -22,19 +28,19 @@ type Client interface {
String() string
}
// Router manages request routing
// Router manages request routing.
type Router interface {
SendRequest(context.Context, Request) (Response, error)
}
// Message is the interface for publishing asynchronously
// Message is the interface for publishing asynchronously.
type Message interface {
Topic() string
Payload() interface{}
ContentType() string
}
// Request is the interface for a synchronous request used by Call or Stream
// Request is the interface for a synchronous request used by Call or Stream.
type Request interface {
// The service to call
Service() string
@@ -52,7 +58,7 @@ type Request interface {
Stream() bool
}
// Response is the response received from a service
// Response is the response received from a service.
type Response interface {
// Read the response
Codec() codec.Reader
@@ -62,7 +68,7 @@ type Response interface {
Read() ([]byte, error)
}
// Stream is the inteface for a bidirectional synchronous stream
// Stream is the inteface for a bidirectional synchronous stream.
type Stream interface {
Closer
// Context for the stream
@@ -81,48 +87,28 @@ type Stream interface {
Close() error
}
// Closer handle client close
// Closer handle client close.
type Closer interface {
// CloseSend closes the send direction of the stream.
CloseSend() error
}
// Option used by the Client
// Option used by the Client.
type Option func(*Options)
// CallOption used by Call or Stream
// CallOption used by Call or Stream.
type CallOption func(*CallOptions)
// PublishOption used by Publish
// PublishOption used by Publish.
type PublishOption func(*PublishOptions)
// MessageOption used by NewMessage
// MessageOption used by NewMessage.
type MessageOption func(*MessageOptions)
// RequestOption used by NewRequest
// RequestOption used by NewRequest.
type RequestOption func(*RequestOptions)
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
)
// Makes a synchronous call to a service using the default client
// Makes a synchronous call to a service using the default client.
func Call(ctx context.Context, request Request, response interface{}, opts ...CallOption) error {
return DefaultClient.Call(ctx, request, response, opts...)
}
@@ -133,13 +119,13 @@ func Publish(ctx context.Context, msg Message, opts ...PublishOption) error {
return DefaultClient.Publish(ctx, msg, opts...)
}
// Creates a new message using the default client
// Creates a new message using the default client.
func NewMessage(topic string, payload interface{}, opts ...MessageOption) Message {
return DefaultClient.NewMessage(topic, payload, opts...)
}
// Creates a new request using the default client. Content Type will
// be set to the default within options and use the appropriate codec
// be set to the default within options and use the appropriate codec.
func NewRequest(service, endpoint string, request interface{}, reqOpts ...RequestOption) Request {
return DefaultClient.NewRequest(service, endpoint, request, reqOpts...)
}
vendor/go-micro.dev/v4/client/options.go
Vendored
+115 -81
View File
@@ -12,32 +12,38 @@ import (
"go-micro.dev/v4/transport"
)
type Options struct {
// Used to select codec
ContentType string
var (
// 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 = 5
// DefaultRequestTimeout is the default request timeout.
DefaultRequestTimeout = time.Second * 30
// DefaultConnectionTimeout is the default connection timeout.
DefaultConnectionTimeout = time.Second * 5
// DefaultPoolSize sets the connection pool size.
DefaultPoolSize = 100
// DefaultPoolTTL sets the connection pool ttl.
DefaultPoolTTL = time.Minute
)
// Plugged interfaces
Broker broker.Broker
Codecs map[string]codec.NewCodec
Registry registry.Registry
Selector selector.Selector
Transport transport.Transport
// Options are the Client options.
type Options struct {
// Default Call Options
CallOptions CallOptions
// Router sets the router
Router Router
// Connection Pool
PoolSize int
PoolTTL time.Duration
Registry registry.Registry
Selector selector.Selector
Transport transport.Transport
// Response cache
Cache *Cache
// Middleware for client
Wrappers []Wrapper
// Default Call Options
CallOptions CallOptions
// Plugged interfaces
Broker broker.Broker
// Logger is the underline logger
Logger logger.Logger
@@ -45,44 +51,65 @@ type Options struct {
// Other options for implementations of the interface
// can be stored in a context
Context context.Context
Codecs map[string]codec.NewCodec
// Response cache
Cache *Cache
// Used to select codec
ContentType string
// Middleware for client
Wrappers []Wrapper
// Connection Pool
PoolSize int
PoolTTL time.Duration
}
// CallOptions are options used to make calls to a server.
type CallOptions struct {
// Other options for implementations of the interface
// can be stored in a context
Context context.Context
// Backoff func
Backoff BackoffFunc
// Check if retriable func
Retry RetryFunc
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
// ConnectionTimeout of one request to the server.
// Set this lower than the RequestTimeout to enbale retries on connection timeout.
ConnectionTimeout time.Duration
// Request/Response timeout of entire srv.Call, for single request timeout set ConnectionTimeout.
RequestTimeout time.Duration
// Stream timeout for the stream
StreamTimeout time.Duration
// Duration to cache the response for
CacheExpiry time.Duration
// Transport Dial Timeout. Used for initial dial to establish a connection.
DialTimeout time.Duration
// Number of Call attempts
Retries int
// Use the services own auth token
ServiceToken bool
// ConnClose sets the Connection: close header.
ConnClose bool
}
type PublishOptions struct {
// Exchange is the routing exchange for the message
Exchange string
// Other options for implementations of the interface
// can be stored in a context
Context context.Context
// Exchange is the routing exchange for the message
Exchange string
}
type MessageOptions struct {
@@ -90,14 +117,15 @@ type MessageOptions struct {
}
type RequestOptions struct {
ContentType string
Stream bool
// Other options for implementations of the interface
// can be stored in a context
Context context.Context
Context context.Context
ContentType string
Stream bool
}
// NewOptions creates new Client options.
func NewOptions(options ...Option) Options {
opts := Options{
Cache: NewCache(),
@@ -105,11 +133,12 @@ func NewOptions(options ...Option) Options {
ContentType: DefaultContentType,
Codecs: make(map[string]codec.NewCodec),
CallOptions: CallOptions{
Backoff: DefaultBackoff,
Retry: DefaultRetry,
Retries: DefaultRetries,
RequestTimeout: DefaultRequestTimeout,
DialTimeout: transport.DefaultDialTimeout,
Backoff: DefaultBackoff,
Retry: DefaultRetry,
Retries: DefaultRetries,
RequestTimeout: DefaultRequestTimeout,
ConnectionTimeout: DefaultConnectionTimeout,
DialTimeout: transport.DefaultDialTimeout,
},
PoolSize: DefaultPoolSize,
PoolTTL: DefaultPoolTTL,
@@ -127,42 +156,42 @@ func NewOptions(options ...Option) Options {
return opts
}
// Broker to be used for pub/sub
// Broker to be used for pub/sub.
func Broker(b broker.Broker) Option {
return func(o *Options) {
o.Broker = b
}
}
// Codec to be used to encode/decode requests for a given content type
// Codec to be used to encode/decode requests for a given content type.
func Codec(contentType string, c codec.NewCodec) Option {
return func(o *Options) {
o.Codecs[contentType] = c
}
}
// Default content type of the client
// ContentType sets the default content type of the client.
func ContentType(ct string) Option {
return func(o *Options) {
o.ContentType = ct
}
}
// PoolSize sets the connection pool size
// PoolSize sets the connection pool size.
func PoolSize(d int) Option {
return func(o *Options) {
o.PoolSize = d
}
}
// PoolTTL sets the connection pool ttl
// PoolTTL sets the connection pool ttl.
func PoolTTL(d time.Duration) Option {
return func(o *Options) {
o.PoolTTL = d
}
}
// Registry to find nodes for a given service
// Registry to find nodes for a given service.
func Registry(r registry.Registry) Option {
return func(o *Options) {
o.Registry = r
@@ -171,28 +200,28 @@ func Registry(r registry.Registry) Option {
}
}
// Transport to use for communication e.g http, rabbitmq, etc
// Transport to use for communication e.g http, rabbitmq, etc.
func Transport(t transport.Transport) Option {
return func(o *Options) {
o.Transport = t
}
}
// Select is used to select a node to route a request to
// Select is used to select a node to route a request to.
func Selector(s selector.Selector) Option {
return func(o *Options) {
o.Selector = s
}
}
// Adds a Wrapper to a list of options passed into the client
// Adds a Wrapper to a list of options passed into the client.
func Wrap(w Wrapper) Option {
return func(o *Options) {
o.Wrappers = append(o.Wrappers, w)
}
}
// Adds a Wrapper to the list of CallFunc wrappers
// Adds a Wrapper to the list of CallFunc wrappers.
func WrapCall(cw ...CallWrapper) Option {
return func(o *Options) {
o.CallOptions.CallWrappers = append(o.CallOptions.CallWrappers, cw...)
@@ -200,15 +229,14 @@ func WrapCall(cw ...CallWrapper) Option {
}
// Backoff is used to set the backoff function used
// when retrying Calls
// when retrying Calls.
func Backoff(fn BackoffFunc) Option {
return func(o *Options) {
o.CallOptions.Backoff = fn
}
}
// Number of retries when making the request.
// Should this be a Call Option?
// Retries set the number of retries when making the request.
func Retries(i int) Option {
return func(o *Options) {
o.CallOptions.Retries = i
@@ -222,22 +250,21 @@ func Retry(fn RetryFunc) Option {
}
}
// The request timeout.
// Should this be a Call Option?
// RequestTimeout set the request timeout.
func RequestTimeout(d time.Duration) Option {
return func(o *Options) {
o.CallOptions.RequestTimeout = d
}
}
// StreamTimeout sets the stream timeout
// StreamTimeout sets the stream timeout.
func StreamTimeout(d time.Duration) Option {
return func(o *Options) {
o.CallOptions.StreamTimeout = d
}
}
// Transport dial timeout
// DialTimeout sets the transport dial timeout.
func DialTimeout(d time.Duration) Option {
return func(o *Options) {
o.CallOptions.DialTimeout = d
@@ -246,21 +273,21 @@ func DialTimeout(d time.Duration) Option {
// Call Options
// WithExchange sets the exchange to route a message through
// WithExchange sets the exchange to route a message through.
func WithExchange(e string) PublishOption {
return func(o *PublishOptions) {
o.Exchange = e
}
}
// PublishContext sets the context in publish options
// PublishContext sets the context in publish options.
func PublishContext(ctx context.Context) PublishOption {
return func(o *PublishOptions) {
o.Context = ctx
}
}
// WithAddress sets the remote addresses to use rather than using service discovery
// WithAddress sets the remote addresses to use rather than using service discovery.
func WithAddress(a ...string) CallOption {
return func(o *CallOptions) {
o.Address = a
@@ -273,7 +300,7 @@ func WithSelectOption(so ...selector.SelectOption) CallOption {
}
}
// WithCallWrapper is a CallOption which adds to the existing CallFunc wrappers
// WithCallWrapper is a CallOption which adds to the existing CallFunc wrappers.
func WithCallWrapper(cw ...CallWrapper) CallOption {
return func(o *CallOptions) {
o.CallWrappers = append(o.CallWrappers, cw...)
@@ -281,7 +308,7 @@ func WithCallWrapper(cw ...CallWrapper) CallOption {
}
// WithBackoff is a CallOption which overrides that which
// set in Options.CallOptions
// set in Options.CallOptions.
func WithBackoff(fn BackoffFunc) CallOption {
return func(o *CallOptions) {
o.Backoff = fn
@@ -289,15 +316,15 @@ func WithBackoff(fn BackoffFunc) CallOption {
}
// WithRetry is a CallOption which overrides that which
// set in Options.CallOptions
// set in Options.CallOptions.
func WithRetry(fn RetryFunc) CallOption {
return func(o *CallOptions) {
o.Retry = fn
}
}
// WithRetries is a CallOption which overrides that which
// set in Options.CallOptions
// WithRetries sets the number of tries for a call.
// This CallOption overrides Options.CallOptions.
func WithRetries(i int) CallOption {
return func(o *CallOptions) {
o.Retries = i
@@ -305,14 +332,21 @@ func WithRetries(i int) CallOption {
}
// WithRequestTimeout is a CallOption which overrides that which
// set in Options.CallOptions
// set in Options.CallOptions.
func WithRequestTimeout(d time.Duration) CallOption {
return func(o *CallOptions) {
o.RequestTimeout = d
}
}
// WithStreamTimeout sets the stream timeout
// WithConnClose sets the Connection header to close.
func WithConnClose() CallOption {
return func(o *CallOptions) {
o.ConnClose = true
}
}
// WithStreamTimeout sets the stream timeout.
func WithStreamTimeout(d time.Duration) CallOption {
return func(o *CallOptions) {
o.StreamTimeout = d
@@ -320,7 +354,7 @@ func WithStreamTimeout(d time.Duration) CallOption {
}
// WithDialTimeout is a CallOption which overrides that which
// set in Options.CallOptions
// set in Options.CallOptions.
func WithDialTimeout(d time.Duration) CallOption {
return func(o *CallOptions) {
o.DialTimeout = d
@@ -328,7 +362,7 @@ func WithDialTimeout(d time.Duration) CallOption {
}
// WithServiceToken is a CallOption which overrides the
// authorization header with the services own auth token
// authorization header with the services own auth token.
func WithServiceToken() CallOption {
return func(o *CallOptions) {
o.ServiceToken = true
@@ -336,7 +370,7 @@ func WithServiceToken() CallOption {
}
// WithCache is a CallOption which sets the duration the response
// shoull be cached for
// shoull be cached for.
func WithCache(c time.Duration) CallOption {
return func(o *CallOptions) {
o.CacheExpiry = c
@@ -363,14 +397,14 @@ func StreamingRequest() RequestOption {
}
}
// WithRouter sets the client router
// WithRouter sets the client router.
func WithRouter(r Router) Option {
return func(o *Options) {
o.Router = r
}
}
// WithLogger sets the underline logger
// WithLogger sets the underline logger.
func WithLogger(l logger.Logger) Option {
return func(o *Options) {
o.Logger = l
vendor/go-micro.dev/v4/client/retry.go
Vendored
+6 -5
View File
@@ -6,15 +6,15 @@ import (
"go-micro.dev/v4/errors"
)
// note that returning either false or a non-nil error will result in the call not being retried
// note that returning either false or a non-nil error will result in the call not being retried.
type RetryFunc func(ctx context.Context, req Request, retryCount int, err error) (bool, error)
// RetryAlways always retry on error
// RetryAlways always retry on error.
func RetryAlways(ctx context.Context, req Request, retryCount int, err error) (bool, error) {
return true, nil
}
// RetryOnError retries a request on a 500 or timeout error
// RetryOnError retries a request on a 500 or timeout error.
func RetryOnError(ctx context.Context, req Request, retryCount int, err error) (bool, error) {
if err == nil {
return false, nil
@@ -26,8 +26,9 @@ func RetryOnError(ctx context.Context, req Request, retryCount int, err error) (
}
switch e.Code {
// retry on timeout or internal server error
case 408, 500:
// Retry on timeout, not on 500 internal server error, as that is a business
// logic error that should be handled by the user.
case 408:
return true, nil
default:
return false, nil
vendor/go-micro.dev/v4/client/rpc_client.go
Vendored
+152 -64
View File
@@ -3,31 +3,43 @@ package client
import (
"context"
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/google/uuid"
"github.com/pkg/errors"
"go-micro.dev/v4/broker"
"go-micro.dev/v4/codec"
raw "go-micro.dev/v4/codec/bytes"
"go-micro.dev/v4/errors"
merrors "go-micro.dev/v4/errors"
log "go-micro.dev/v4/logger"
"go-micro.dev/v4/metadata"
"go-micro.dev/v4/registry"
"go-micro.dev/v4/selector"
"go-micro.dev/v4/transport"
"go-micro.dev/v4/transport/headers"
"go-micro.dev/v4/util/buf"
"go-micro.dev/v4/util/net"
"go-micro.dev/v4/util/pool"
)
const (
packageID = "go.micro.client"
)
type rpcClient struct {
seq uint64
once atomic.Value
opts Options
once atomic.Value
pool pool.Pool
seq uint64
mu sync.RWMutex
}
func newRpcClient(opt ...Option) Client {
func newRPCClient(opt ...Option) Client {
opts := NewOptions(opt...)
p := pool.NewPool(
@@ -57,14 +69,17 @@ func (r *rpcClient) newCodec(contentType string) (codec.NewCodec, error) {
if c, ok := r.opts.Codecs[contentType]; ok {
return c, nil
}
if cf, ok := DefaultCodecs[contentType]; ok {
return cf, nil
}
return nil, fmt.Errorf("unsupported Content-Type: %s", contentType)
}
func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request, resp interface{}, opts CallOptions) error {
address := node.Address
logger := r.Options().Logger
msg := &transport.Message{
Header: make(map[string]string),
@@ -73,31 +88,43 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
md, ok := metadata.FromContext(ctx)
if ok {
for k, v := range md {
// don't copy Micro-Topic header, that used for pub/sub
// this fix case then client uses the same context that received in subscriber
if k == "Micro-Topic" {
// Don't copy Micro-Topic header, that is used for pub/sub
// this is fixes the case when the client uses the same context that
// is received in the subscriber.
if k == headers.Message {
continue
}
msg.Header[k] = v
}
}
// Set connection timeout for single requests to the server. Should be > 0
// as otherwise requests can't be made.
cTimeout := opts.ConnectionTimeout
if cTimeout == 0 {
logger.Log(log.DebugLevel, "connection timeout was set to 0, overridng to default connection timeout")
cTimeout = DefaultConnectionTimeout
}
// set timeout in nanoseconds
msg.Header["Timeout"] = fmt.Sprintf("%d", opts.RequestTimeout)
msg.Header["Timeout"] = fmt.Sprintf("%d", cTimeout)
// set the content type for the request
msg.Header["Content-Type"] = req.ContentType()
// set the accept header
msg.Header["Accept"] = req.ContentType()
// setup old protocol
cf := setupProtocol(msg, node)
reqCodec := setupProtocol(msg, node)
// no codec specified
if cf == nil {
if reqCodec == nil {
var err error
cf, err = r.newCodec(req.ContentType())
reqCodec, err = r.newCodec(req.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return merrors.InternalServerError("go.micro.client", err.Error())
}
}
@@ -109,19 +136,29 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
dOpts = append(dOpts, transport.WithTimeout(opts.DialTimeout))
}
if opts.ConnClose {
dOpts = append(dOpts, transport.WithConnClose())
}
c, err := r.pool.Get(address, dOpts...)
if err != nil {
return errors.InternalServerError("go.micro.client", "connection error: %v", err)
return merrors.InternalServerError("go.micro.client", "connection error: %v", err)
}
seq := atomic.AddUint64(&r.seq, 1) - 1
codec := newRpcCodec(msg, c, cf, "")
codec := newRPCCodec(msg, c, reqCodec, "")
rsp := &rpcResponse{
socket: c,
codec: codec,
}
releaseFunc := func(err error) {
if err = r.pool.Release(c, err); err != nil {
logger.Log(log.ErrorLevel, "failed to release pool", err)
}
}
stream := &rpcStream{
id: fmt.Sprintf("%v", seq),
context: ctx,
@@ -129,11 +166,17 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
response: rsp,
codec: codec,
closed: make(chan bool),
release: func(err error) { r.pool.Release(c, err) },
close: opts.ConnClose,
release: releaseFunc,
sendEOS: false,
}
// close the stream on exiting this function
defer stream.Close()
defer func() {
if err := stream.Close(); err != nil {
logger.Log(log.ErrorLevel, "failed to close stream", err)
}
}()
// wait for error response
ch := make(chan error, 1)
@@ -141,7 +184,7 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
go func() {
defer func() {
if r := recover(); r != nil {
ch <- errors.InternalServerError("go.micro.client", "panic recovered: %v", r)
ch <- merrors.InternalServerError("go.micro.client", "panic recovered: %v", r)
}
}()
@@ -166,8 +209,8 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
select {
case err := <-ch:
return err
case <-ctx.Done():
grr = errors.Timeout("go.micro.client", fmt.Sprintf("%v", ctx.Err()))
case <-time.After(cTimeout):
grr = merrors.Timeout("go.micro.client", fmt.Sprintf("%v", ctx.Err()))
}
// set the stream error
@@ -184,6 +227,7 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request, opts CallOptions) (Stream, error) {
address := node.Address
logger := r.Options().Logger
msg := &transport.Message{
Header: make(map[string]string),
@@ -206,14 +250,15 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
msg.Header["Accept"] = req.ContentType()
// set old codecs
cf := setupProtocol(msg, node)
nCodec := setupProtocol(msg, node)
// no codec specified
if cf == nil {
if nCodec == nil {
var err error
cf, err = r.newCodec(req.ContentType())
nCodec, err = r.newCodec(req.ContentType())
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
return nil, merrors.InternalServerError("go.micro.client", err.Error())
}
}
@@ -227,7 +272,7 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
c, err := r.opts.Transport.Dial(address, dOpts...)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", "connection error: %v", err)
return nil, merrors.InternalServerError("go.micro.client", "connection error: %v", err)
}
// increment the sequence number
@@ -235,7 +280,7 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
id := fmt.Sprintf("%v", seq)
// create codec with stream id
codec := newRpcCodec(msg, c, cf, id)
codec := newRPCCodec(msg, c, nCodec, id)
rsp := &rpcResponse{
socket: c,
@@ -247,6 +292,12 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
r.codec = codec
}
releaseFunc := func(_ error) {
if err = c.Close(); err != nil {
logger.Log(log.ErrorLevel, err)
}
}
stream := &rpcStream{
id: id,
context: ctx,
@@ -257,8 +308,7 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
closed: make(chan bool),
// signal the end of stream,
sendEOS: true,
// release func
release: func(err error) { c.Close() },
release: releaseFunc,
}
// wait for error response
@@ -275,7 +325,7 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
case err := <-ch:
grr = err
case <-ctx.Done():
grr = errors.Timeout("go.micro.client", fmt.Sprintf("%v", ctx.Err()))
grr = merrors.Timeout("go.micro.client", fmt.Sprintf("%v", ctx.Err()))
}
if grr != nil {
@@ -285,7 +335,10 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
stream.Unlock()
// close the stream
stream.Close()
if err := stream.Close(); err != nil {
logger.Logf(log.ErrorLevel, "failed to close stream: %v", err)
}
return nil, grr
}
@@ -293,6 +346,9 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
}
func (r *rpcClient) Init(opts ...Option) error {
r.mu.Lock()
defer r.mu.Unlock()
size := r.opts.PoolSize
ttl := r.opts.PoolTTL
tr := r.opts.Transport
@@ -304,7 +360,10 @@ func (r *rpcClient) Init(opts ...Option) error {
// update pool configuration if the options changed
if size != r.opts.PoolSize || ttl != r.opts.PoolTTL || tr != r.opts.Transport {
// close existing pool
r.pool.Close()
if err := r.pool.Close(); err != nil {
return errors.Wrap(err, "failed to close pool")
}
// create new pool
r.pool = pool.NewPool(
pool.Size(r.opts.PoolSize),
@@ -316,11 +375,15 @@ func (r *rpcClient) Init(opts ...Option) error {
return nil
}
// Options retrives the options.
func (r *rpcClient) Options() Options {
r.mu.RLock()
defer r.mu.RUnlock()
return r.opts
}
// next returns an iterator for the next nodes to call
// next returns an iterator for the next nodes to call.
func (r *rpcClient) next(request Request, opts CallOptions) (selector.Next, error) {
// try get the proxy
service, address, _ := net.Proxy(request.Service(), opts.Address)
@@ -348,16 +411,22 @@ func (r *rpcClient) next(request Request, opts CallOptions) (selector.Next, erro
// get next nodes from the selector
next, err := r.opts.Selector.Select(service, opts.SelectOptions...)
if err != nil {
if err == selector.ErrNotFound {
return nil, errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
if errors.Is(err, selector.ErrNotFound) {
return nil, merrors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
}
return nil, errors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error())
return nil, merrors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error())
}
return next, nil
}
func (r *rpcClient) Call(ctx context.Context, request Request, response interface{}, opts ...CallOption) error {
// TODO: further validate these mutex locks. full lock would prevent
// parallel calls. Maybe we can set individual locks for secctions.
r.mu.RLock()
defer r.mu.RUnlock()
// make a copy of call opts
callOpts := r.opts.CallOptions
for _, opt := range opts {
@@ -375,6 +444,7 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
// no deadline so we create a new one
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, callOpts.RequestTimeout)
defer cancel()
} else {
// got a deadline so no need to setup context
@@ -386,7 +456,7 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
// should we noop right here?
select {
case <-ctx.Done():
return errors.Timeout("go.micro.client", fmt.Sprintf("%v", ctx.Err()))
return merrors.Timeout("go.micro.client", fmt.Sprintf("%v", ctx.Err()))
default:
}
@@ -403,7 +473,7 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
// call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, request, i)
if err != nil {
return errors.InternalServerError("go.micro.client", "backoff error: %v", err.Error())
return merrors.InternalServerError("go.micro.client", "backoff error: %v", err.Error())
}
// only sleep if greater than 0
@@ -414,16 +484,19 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
// select next node
node, err := next()
service := request.Service()
if err != nil {
if err == selector.ErrNotFound {
return errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
if errors.Is(err, selector.ErrNotFound) {
return merrors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
}
return errors.InternalServerError("go.micro.client", "error getting next %s node: %s", service, err.Error())
return merrors.InternalServerError("go.micro.client", "error getting next %s node: %s", service, err.Error())
}
// make the call
err = rcall(ctx, node, request, response, callOpts)
r.opts.Selector.Mark(service, node, err)
return err
}
@@ -431,11 +504,13 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
retries := callOpts.Retries
// disable retries when using a proxy
if _, _, ok := net.Proxy(request.Service(), callOpts.Address); ok {
retries = 0
}
// Note: I don't see why we should disable retries for proxies, so commenting out.
// if _, _, ok := net.Proxy(request.Service(), callOpts.Address); ok {
// retries = 0
// }
ch := make(chan error, retries+1)
var gerr error
for i := 0; i <= retries; i++ {
@@ -445,7 +520,7 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
select {
case <-ctx.Done():
return errors.Timeout("go.micro.client", fmt.Sprintf("call timeout: %v", ctx.Err()))
return merrors.Timeout("go.micro.client", fmt.Sprintf("call timeout: %v", ctx.Err()))
case err := <-ch:
// if the call succeeded lets bail early
if err == nil {
@@ -461,6 +536,8 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
return err
}
r.opts.Logger.Logf(log.DebugLevel, "Retrying request. Previous attempt failed with: %v", err)
gerr = err
}
}
@@ -469,6 +546,9 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
}
func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOption) (Stream, error) {
r.mu.RLock()
defer r.mu.RUnlock()
// make a copy of call opts
callOpts := r.opts.CallOptions
for _, opt := range opts {
@@ -480,10 +560,9 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
return nil, err
}
// should we noop right here?
select {
case <-ctx.Done():
return nil, errors.Timeout("go.micro.client", fmt.Sprintf("%v", ctx.Err()))
return nil, merrors.Timeout("go.micro.client", fmt.Sprintf("%v", ctx.Err()))
default:
}
@@ -491,7 +570,7 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
// call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, request, i)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", "backoff error: %v", err.Error())
return nil, merrors.InternalServerError("go.micro.client", "backoff error: %v", err.Error())
}
// only sleep if greater than 0
@@ -501,15 +580,18 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
node, err := next()
service := request.Service()
if err != nil {
if err == selector.ErrNotFound {
return nil, errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
if errors.Is(err, selector.ErrNotFound) {
return nil, merrors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
}
return nil, errors.InternalServerError("go.micro.client", "error getting next %s node: %s", service, err.Error())
return nil, merrors.InternalServerError("go.micro.client", "error getting next %s node: %s", service, err.Error())
}
stream, err := r.stream(ctx, node, request, callOpts)
r.opts.Selector.Mark(service, node, err)
return stream, err
}
@@ -527,6 +609,7 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
}
ch := make(chan response, retries+1)
var grr error
for i := 0; i <= retries; i++ {
@@ -537,7 +620,7 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
select {
case <-ctx.Done():
return nil, errors.Timeout("go.micro.client", fmt.Sprintf("call timeout: %v", ctx.Err()))
return nil, merrors.Timeout("go.micro.client", fmt.Sprintf("call timeout: %v", ctx.Err()))
case rsp := <-ch:
// if the call succeeded lets bail early
if rsp.err == nil {
@@ -568,15 +651,15 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt
o(&options)
}
md, ok := metadata.FromContext(ctx)
metadata, ok := metadata.FromContext(ctx)
if !ok {
md = make(map[string]string)
metadata = make(map[string]string)
}
id := uuid.New().String()
md["Content-Type"] = msg.ContentType()
md["Micro-Topic"] = msg.Topic()
md["Micro-Id"] = id
metadata["Content-Type"] = msg.ContentType()
metadata[headers.Message] = msg.Topic()
metadata[headers.ID] = id
// set the topic
topic := msg.Topic()
@@ -589,7 +672,7 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt
// encode message body
cf, err := r.newCodec(msg.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return merrors.InternalServerError(packageID, err.Error())
}
var body []byte
@@ -598,33 +681,38 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt
if d, ok := msg.Payload().(*raw.Frame); ok {
body = d.Data
} else {
// new buffer
b := buf.New(nil)
if err := cf(b).Write(&codec.Message{
if err = cf(b).Write(&codec.Message{
Target: topic,
Type: codec.Event,
Header: map[string]string{
"Micro-Id": id,
"Micro-Topic": msg.Topic(),
headers.ID: id,
headers.Message: msg.Topic(),
},
}, msg.Payload()); err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return merrors.InternalServerError(packageID, err.Error())
}
// set the body
body = b.Bytes()
}
if !r.once.Load().(bool) {
l, ok := r.once.Load().(bool)
if !ok {
return fmt.Errorf("failed to cast to bool")
}
if !l {
if err = r.opts.Broker.Connect(); err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
return merrors.InternalServerError(packageID, err.Error())
}
r.once.Store(true)
}
return r.opts.Broker.Publish(topic, &broker.Message{
Header: md,
Header: metadata,
Body: body,
}, broker.PublishContext(options.Context))
}
vendor/go-micro.dev/v4/client/rpc_codec.go
Vendored
+47 -34
View File
@@ -14,6 +14,7 @@ import (
"go-micro.dev/v4/errors"
"go-micro.dev/v4/registry"
"go-micro.dev/v4/transport"
"go-micro.dev/v4/transport/headers"
)
const (
@@ -28,7 +29,7 @@ func (e serverError) Error() string {
return string(e)
}
// errShutdown holds the specific error for closing/closed connections
// errShutdown holds the specific error for closing/closed connections.
var (
errShutdown = errs.New("connection is shut down")
)
@@ -50,8 +51,10 @@ type readWriteCloser struct {
}
var (
// DefaultContentType header.
DefaultContentType = "application/json"
// DefaultCodecs map.
DefaultCodecs = map[string]codec.NewCodec{
"application/grpc": grpc.NewCodec,
"application/grpc+json": grpc.NewCodec,
@@ -63,7 +66,7 @@ var (
"application/octet-stream": raw.NewCodec,
}
// TODO: remove legacy codec list
// TODO: remove legacy codec list.
defaultCodecs = map[string]codec.NewCodec{
"application/json": jsonrpc.NewCodec,
"application/json-rpc": jsonrpc.NewCodec,
@@ -84,6 +87,7 @@ func (rwc *readWriteCloser) Write(p []byte) (n int, err error) {
func (rwc *readWriteCloser) Close() error {
rwc.rbuf.Reset()
rwc.wbuf.Reset()
return nil
}
@@ -92,20 +96,21 @@ func getHeaders(m *codec.Message) {
if len(v) > 0 {
return v
}
return m.Header[hdr]
}
// check error in header
m.Error = set(m.Error, "Micro-Error")
m.Error = set(m.Error, headers.Error)
// check endpoint in header
m.Endpoint = set(m.Endpoint, "Micro-Endpoint")
m.Endpoint = set(m.Endpoint, headers.Endpoint)
// check method in header
m.Method = set(m.Method, "Micro-Method")
m.Method = set(m.Method, headers.Method)
// set the request id
m.Id = set(m.Id, "Micro-Id")
m.Id = set(m.Id, headers.ID)
}
func setHeaders(m *codec.Message, stream string) {
@@ -113,21 +118,22 @@ func setHeaders(m *codec.Message, stream string) {
if len(v) == 0 {
return
}
m.Header[hdr] = v
}
set("Micro-Id", m.Id)
set("Micro-Service", m.Target)
set("Micro-Method", m.Method)
set("Micro-Endpoint", m.Endpoint)
set("Micro-Error", m.Error)
set(headers.ID, m.Id)
set(headers.Request, m.Target)
set(headers.Method, m.Method)
set(headers.Endpoint, m.Endpoint)
set(headers.Error, m.Error)
if len(stream) > 0 {
set("Micro-Stream", stream)
set(headers.Stream, stream)
}
}
// setupProtocol sets up the old protocol
// setupProtocol sets up the old protocol.
func setupProtocol(msg *transport.Message, node *registry.Node) codec.NewCodec {
protocol := node.Metadata["protocol"]
@@ -137,7 +143,7 @@ func setupProtocol(msg *transport.Message, node *registry.Node) codec.NewCodec {
}
// processing topic publishing
if len(msg.Header["Micro-Topic"]) > 0 {
if len(msg.Header[headers.Message]) > 0 {
return nil
}
@@ -149,60 +155,59 @@ func setupProtocol(msg *transport.Message, node *registry.Node) codec.NewCodec {
msg.Header["Content-Type"] = "application/proto-rpc"
}
// now return codec
return defaultCodecs[msg.Header["Content-Type"]]
}
func newRpcCodec(req *transport.Message, client transport.Client, c codec.NewCodec, stream string) codec.Codec {
func newRPCCodec(req *transport.Message, client transport.Client, c codec.NewCodec, stream string) codec.Codec {
rwc := &readWriteCloser{
wbuf: bytes.NewBuffer(nil),
rbuf: bytes.NewBuffer(nil),
}
r := &rpcCodec{
return &rpcCodec{
buf: rwc,
client: client,
codec: c(rwc),
req: req,
stream: stream,
}
return r
}
func (c *rpcCodec) Write(m *codec.Message, body interface{}) error {
func (c *rpcCodec) Write(message *codec.Message, body interface{}) error {
c.buf.wbuf.Reset()
// create header
if m.Header == nil {
m.Header = map[string]string{}
if message.Header == nil {
message.Header = map[string]string{}
}
// copy original header
for k, v := range c.req.Header {
m.Header[k] = v
message.Header[k] = v
}
// set the mucp headers
setHeaders(m, c.stream)
setHeaders(message, c.stream)
// if body is bytes Frame don't encode
if body != nil {
if b, ok := body.(*raw.Frame); ok {
// set body
m.Body = b.Data
message.Body = b.Data
} else {
// write to codec
if err := c.codec.Write(m, body); err != nil {
if err := c.codec.Write(message, body); err != nil {
return errors.InternalServerError("go.micro.client.codec", err.Error())
}
// set body
m.Body = c.buf.wbuf.Bytes()
message.Body = c.buf.wbuf.Bytes()
}
}
// create new transport message
msg := transport.Message{
Header: m.Header,
Body: m.Body,
Header: message.Header,
Body: message.Body,
}
// send the request
@@ -213,7 +218,7 @@ func (c *rpcCodec) Write(m *codec.Message, body interface{}) error {
return nil
}
func (c *rpcCodec) ReadHeader(m *codec.Message, r codec.MessageType) error {
func (c *rpcCodec) ReadHeader(msg *codec.Message, r codec.MessageType) error {
var tm transport.Message
// read message from transport
@@ -225,13 +230,13 @@ func (c *rpcCodec) ReadHeader(m *codec.Message, r codec.MessageType) error {
c.buf.rbuf.Write(tm.Body)
// set headers from transport
m.Header = tm.Header
msg.Header = tm.Header
// read header
err := c.codec.ReadHeader(m, r)
err := c.codec.ReadHeader(msg, r)
// get headers
getHeaders(m)
getHeaders(msg)
// return header error
if err != nil {
@@ -252,15 +257,23 @@ func (c *rpcCodec) ReadBody(b interface{}) error {
if err := c.codec.ReadBody(b); err != nil {
return errors.InternalServerError("go.micro.client.codec", err.Error())
}
return nil
}
func (c *rpcCodec) Close() error {
c.buf.Close()
c.codec.Close()
if err := c.buf.Close(); err != nil {
return err
}
if err := c.codec.Close(); err != nil {
return err
}
if err := c.client.Close(); err != nil {
return errors.InternalServerError("go.micro.client.transport", err.Error())
}
return nil
}
vendor/go-micro.dev/v4/client/rpc_message.go
Vendored
+1 -1
View File
@@ -1,9 +1,9 @@
package client
type message struct {
payload interface{}
topic string
contentType string
payload interface{}
}
func newMessage(topic string, payload interface{}, contentType string, opts ...MessageOption) Message {
vendor/go-micro.dev/v4/client/rpc_request.go
Vendored
+3 -3
View File
@@ -5,13 +5,13 @@ import (
)
type rpcRequest struct {
opts RequestOptions
codec codec.Codec
body interface{}
service string
method string
endpoint string
contentType string
codec codec.Codec
body interface{}
opts RequestOptions
}
func newRequest(service, endpoint string, request interface{}, contentType string, reqOpts ...RequestOption) Request {
vendor/go-micro.dev/v4/client/rpc_response.go
Vendored
+2 -2
View File
@@ -6,10 +6,10 @@ import (
)
type rpcResponse struct {
header map[string]string
body []byte
socket transport.Socket
codec codec.Codec
header map[string]string
body []byte
}
func (r *rpcResponse) Codec() codec.Reader {
vendor/go-micro.dev/v4/client/rpc_stream.go
Vendored
+24 -10
View File
@@ -9,22 +9,25 @@ import (
"go-micro.dev/v4/codec"
)
// Implements the streamer interface
// Implements the streamer interface.
type rpcStream struct {
sync.RWMutex
id string
closed chan bool
err error
request Request
response Response
codec codec.Codec
context context.Context
// signal whether we should send EOS
sendEOS bool
closed chan bool
// release releases the connection back to the pool
release func(err error)
id string
sync.RWMutex
// Indicates whether connection should be closed directly.
close bool
// signal whether we should send EOS
sendEOS bool
}
func (r *rpcStream) isClosed() bool {
@@ -79,6 +82,7 @@ func (r *rpcStream) Recv(msg interface{}) error {
if r.isClosed() {
r.err = errShutdown
r.Unlock()
return errShutdown
}
@@ -87,15 +91,19 @@ func (r *rpcStream) Recv(msg interface{}) error {
r.Unlock()
err := r.codec.ReadHeader(&resp, codec.Response)
r.Lock()
if err != nil {
if err == io.EOF && !r.isClosed() {
if errors.Is(err, io.EOF) && !r.isClosed() {
r.err = io.ErrUnexpectedEOF
r.Unlock()
return io.ErrUnexpectedEOF
}
r.err = err
r.Unlock()
return err
}
@@ -124,13 +132,15 @@ func (r *rpcStream) Recv(msg interface{}) error {
}
}
r.Unlock()
defer r.Unlock()
return r.err
}
func (r *rpcStream) Error() error {
r.RLock()
defer r.RUnlock()
return r.err
}
@@ -152,6 +162,7 @@ func (r *rpcStream) Close() error {
// send the end of stream message
if r.sendEOS {
// no need to check for error
//nolint:errcheck,gosec
r.codec.Write(&codec.Message{
Id: r.id,
Target: r.request.Service(),
@@ -164,10 +175,13 @@ func (r *rpcStream) Close() error {
err := r.codec.Close()
rerr := r.Error()
if r.close && rerr == nil {
rerr = errors.New("connection header set to close")
}
// release the connection
r.release(r.Error())
r.release(rerr)
// return the codec error
return err
}
}
vendor/go-micro.dev/v4/client/wrapper.go
Vendored
+4 -4
View File
@@ -6,14 +6,14 @@ import (
"go-micro.dev/v4/registry"
)
// CallFunc represents the individual call func
// CallFunc represents the individual call func.
type CallFunc func(ctx context.Context, node *registry.Node, req Request, rsp interface{}, opts CallOptions) error
// CallWrapper is a low level wrapper for the CallFunc
// CallWrapper is a low level wrapper for the CallFunc.
type CallWrapper func(CallFunc) CallFunc
// Wrapper wraps a client and returns a client
// Wrapper wraps a client and returns a client.
type Wrapper func(Client) Client
// StreamWrapper wraps a Stream and returns the equivalent
// StreamWrapper wraps a Stream and returns the equivalent.
type StreamWrapper func(Stream) Stream