Add transport dial timeout

This commit is contained in:
Asim 2016-01-03 21:25:03 +00:00
parent 332a229169
commit d43d3e8efa
4 changed files with 32 additions and 8 deletions

View File

@ -20,6 +20,7 @@ type Options struct {
Wrappers []Wrapper Wrappers []Wrapper
Retries int Retries int
RequestTimeout time.Duration RequestTimeout time.Duration
DialTimeout time.Duration
// Other options to be used by client implementations // Other options to be used by client implementations
Options map[string]string Options map[string]string
@ -49,6 +50,7 @@ func newOptions(options ...Option) Options {
Codecs: make(map[string]codec.NewCodec), Codecs: make(map[string]codec.NewCodec),
Retries: DefaultRetries, Retries: DefaultRetries,
RequestTimeout: DefaultRequestTimeout, RequestTimeout: DefaultRequestTimeout,
DialTimeout: transport.DefaultDialTimeout,
} }
for _, o := range options { for _, o := range options {
@ -145,6 +147,13 @@ func RequestTimeout(d time.Duration) Option {
} }
} }
// Transport dial timeout
func DialTimeout(d time.Duration) Option {
return func(o *Options) {
o.DialTimeout = d
}
}
// Call Options // Call Options
func WithSelectOption(so selector.SelectOption) CallOption { func WithSelectOption(so selector.SelectOption) CallOption {

View File

@ -70,7 +70,7 @@ func (r *rpcClient) call(ctx context.Context, address string, request Request, r
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", err.Error())
} }
c, err := r.opts.Transport.Dial(address) c, err := r.opts.Transport.Dial(address, transport.WithTimeout(r.opts.DialTimeout))
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err)) return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
} }
@ -116,7 +116,7 @@ func (r *rpcClient) stream(ctx context.Context, address string, req Request) (St
return nil, errors.InternalServerError("go.micro.client", err.Error()) return nil, errors.InternalServerError("go.micro.client", err.Error())
} }
c, err := r.opts.Transport.Dial(address, transport.WithStream()) c, err := r.opts.Transport.Dial(address, transport.WithStream(), transport.WithTimeout(r.opts.DialTimeout))
if err != nil { if err != nil {
return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err)) return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
} }

View File

@ -276,17 +276,19 @@ func (h *httpTransportListener) Accept(fn func(Socket)) error {
} }
func (h *httpTransport) Dial(addr string, opts ...DialOption) (Client, error) { func (h *httpTransport) Dial(addr string, opts ...DialOption) (Client, error) {
conn, err := net.Dial("tcp", addr) dopts := DialOptions{
if err != nil { Timeout: DefaultDialTimeout,
return nil, err
} }
var dopts DialOptions
for _, opt := range opts { for _, opt := range opts {
opt(&dopts) opt(&dopts)
} }
conn, err := net.DialTimeout("tcp", addr, dopts.Timeout)
if err != nil {
return nil, err
}
return &httpTransportClient{ return &httpTransportClient{
ht: h, ht: h,
addr: addr, addr: addr,

View File

@ -1,5 +1,9 @@
package transport package transport
import (
"time"
)
type Message struct { type Message struct {
Header map[string]string Header map[string]string
Body []byte Body []byte
@ -35,7 +39,8 @@ type Options struct {
} }
type DialOptions struct { type DialOptions struct {
Stream bool Stream bool
Timeout time.Duration
// Other options to be used by broker implementations // Other options to be used by broker implementations
Options map[string]string Options map[string]string
@ -47,6 +52,8 @@ type DialOption func(*DialOptions)
var ( var (
DefaultTransport Transport = newHttpTransport([]string{}) DefaultTransport Transport = newHttpTransport([]string{})
DefaultDialTimeout = time.Second * 5
) )
func WithStream() DialOption { func WithStream() DialOption {
@ -55,6 +62,12 @@ func WithStream() DialOption {
} }
} }
func WithTimeout(d time.Duration) DialOption {
return func(o *DialOptions) {
o.Timeout = d
}
}
func NewTransport(addrs []string, opt ...Option) Transport { func NewTransport(addrs []string, opt ...Option) Transport {
return newHttpTransport(addrs, opt...) return newHttpTransport(addrs, opt...)
} }