feat: enable retry for zrpc (#1237)

master
Kevin Wan 3 years ago committed by GitHub
parent 11f85d1b80
commit 09eb53f308
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -19,9 +19,7 @@ var (
// WithTimeout is an alias of internal.WithTimeout.
WithTimeout = internal.WithTimeout
// WithRetry is an alias of internal.WithRetry.
// TODO: enable it in v1.2.4
// WithRetry = internal.WithRetry
WithRetry = internal.WithRetry
// WithTransportCredentials return a func to make the gRPC calls secured with given credentials.
WithTransportCredentials = internal.WithTransportCredentials
// WithUnaryClientInterceptor is an alias of internal.WithUnaryClientInterceptor.
@ -65,10 +63,9 @@ func NewClient(c RpcClientConf, options ...ClientOption) (Client, error) {
if c.Timeout > 0 {
opts = append(opts, WithTimeout(time.Duration(c.Timeout)*time.Millisecond))
}
// TODO: enable it in v1.2.4
// if c.Retry {
// opts = append(opts, WithRetry())
// }
if c.Retry {
opts = append(opts, WithRetry())
}
opts = append(opts, options...)
var target string

@ -96,21 +96,20 @@ func TestDepositServer_Deposit(t *testing.T) {
return invoker(ctx, method, req, reply, cc, opts...)
}),
)
// TODO: enable it in v1.2.4
// retryClient := MustNewClient(
// RpcClientConf{
// Endpoints: []string{"foo"},
// App: "foo",
// Token: "bar",
// Timeout: 1000,
// Retry: true,
// },
// WithDialOption(grpc.WithContextDialer(dialer())),
// WithUnaryClientInterceptor(func(ctx context.Context, method string, req, reply interface{},
// cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
// return invoker(ctx, method, req, reply, cc, opts...)
// }),
// )
retryClient := MustNewClient(
RpcClientConf{
Endpoints: []string{"foo"},
App: "foo",
Token: "bar",
Timeout: 1000,
Retry: true,
},
WithDialOption(grpc.WithContextDialer(dialer())),
WithUnaryClientInterceptor(func(ctx context.Context, method string, req, reply interface{},
cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
return invoker(ctx, method, req, reply, cc, opts...)
}),
)
tarConfClient := MustNewClient(
RpcClientConf{
Target: "foo",
@ -135,8 +134,7 @@ func TestDepositServer_Deposit(t *testing.T) {
clients := []Client{
directClient,
nonBlockClient,
// TODO: enable it in v1.2.4
// retryClient,
retryClient,
tarConfClient,
targetClient,
}

@ -18,8 +18,7 @@ type (
// setting 0 means no timeout
Timeout int64 `json:",default=2000"`
CpuThreshold int64 `json:",default=900,range=[0:1000]"`
// TODO: enable it in v1.2.4
// MaxRetries int `json:",default=0,range=[0:]"`
MaxRetries int `json:",default=0,range=[0:]"`
}
// A RpcClientConf is a rpc client config.
@ -30,9 +29,8 @@ type (
App string `json:",optional"`
Token string `json:",optional"`
NonBlock bool `json:",optional"`
// TODO: enable it in v1.2.4
// Retry bool `json:",optional"` // grpc auto retry
Timeout int64 `json:",default=2000"`
Retry bool `json:",optional"` // grpc auto retry
Timeout int64 `json:",default=2000"`
}
)

@ -38,9 +38,10 @@ func NewServer(c RpcServerConf, register internal.RegisterFn) (*RpcServer, error
var server internal.Server
metrics := stat.NewMetrics(c.ListenOn)
// TODO: enable it in v1.2.4
// serverOptions := []internal.ServerOption{internal.WithMetrics(metrics), internal.WithMaxRetries(c.MaxRetries)}
serverOptions := []internal.ServerOption{internal.WithMetrics(metrics)}
serverOptions := []internal.ServerOption{
internal.WithMetrics(metrics),
internal.WithMaxRetries(c.MaxRetries),
}
if c.HasEtcd() {
server, err = internal.NewRpcPubServer(c.Etcd, c.ListenOn, serverOptions...)

Loading…
Cancel
Save