package client import ( "context" "fmt" "time" "go.unistack.org/micro/v4/codec" "go.unistack.org/micro/v4/errors" "go.unistack.org/micro/v4/metadata" "go.unistack.org/micro/v4/options" "go.unistack.org/micro/v4/selector" "go.unistack.org/micro/v4/semconv" ) // DefaultCodecs will be used to encode/decode data var DefaultCodecs = map[string]codec.Codec{ "application/octet-stream": codec.NewCodec(), } type noopClient struct { opts Options } type noopRequest struct { body interface{} codec codec.Codec service string method string endpoint string contentType string stream bool } // NewClient returns new noop client func NewClient(opts ...options.Option) Client { nc := &noopClient{opts: NewOptions(opts...)} // wrap in reverse c := Client(nc) return c } func (n *noopClient) Name() string { return n.opts.Name } func (n *noopRequest) Service() string { return n.service } func (n *noopRequest) Method() string { return n.method } func (n *noopRequest) Endpoint() string { return n.endpoint } func (n *noopRequest) ContentType() string { return n.contentType } func (n *noopRequest) Body() interface{} { return n.body } func (n *noopRequest) Codec() codec.Codec { return n.codec } func (n *noopRequest) Stream() bool { return n.stream } type noopResponse struct { codec codec.Codec header metadata.Metadata } func (n *noopResponse) Codec() codec.Codec { return n.codec } func (n *noopResponse) Header() metadata.Metadata { return n.header } func (n *noopResponse) Read() ([]byte, error) { return nil, nil } type noopStream struct{} func (n *noopStream) Context() context.Context { return context.Background() } func (n *noopStream) Request() Request { return &noopRequest{} } func (n *noopStream) Response() Response { return &noopResponse{} } func (n *noopStream) Send(interface{}) error { return nil } func (n *noopStream) Recv(interface{}) error { return nil } func (n *noopStream) SendMsg(interface{}) error { return nil } func (n *noopStream) RecvMsg(interface{}) error { return nil } func (n *noopStream) Error() error { return nil } func (n *noopStream) Close() error { return nil } func (n *noopStream) CloseSend() error { return nil } func (n *noopClient) newCodec(contentType string) (codec.Codec, error) { if cf, ok := n.opts.Codecs[contentType]; ok { return cf, nil } if cf, ok := DefaultCodecs[contentType]; ok { return cf, nil } return nil, codec.ErrUnknownContentType } func (n *noopClient) Init(opts ...options.Option) error { for _, o := range opts { o(&n.opts) } return nil } func (n *noopClient) Options() Options { return n.opts } func (n *noopClient) String() string { return "noop" } func (n *noopClient) Call(ctx context.Context, req Request, rsp interface{}, opts ...options.Option) error { // make a copy of call opts callOpts := n.opts.CallOptions for _, opt := range opts { opt(&callOpts) } // check if we already have a deadline d, ok := ctx.Deadline() if !ok { var cancel context.CancelFunc // no deadline so we create a new one ctx, cancel = context.WithTimeout(ctx, callOpts.RequestTimeout) defer cancel() } else { // got a deadline so no need to setup context // but we need to set the timeout we pass along opt := RequestTimeout(time.Until(d)) opt(&callOpts) } // should we noop right here? select { case <-ctx.Done(): return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408) default: } // make copy of call method hcall := n.call // wrap the call in reverse for i := len(callOpts.CallWrappers); i > 0; i-- { hcall = callOpts.CallWrappers[i-1](hcall) } // use the router passed as a call option, or fallback to the rpc clients router if callOpts.Router == nil { callOpts.Router = n.opts.Router } if callOpts.Selector == nil { callOpts.Selector = n.opts.Selector } // inject proxy address // TODO: don't even bother using Lookup/Select in this case if len(n.opts.Proxy) > 0 { callOpts.Address = []string{n.opts.Proxy} } var next selector.Next // return errors.New("go.micro.client", "request timeout", 408) call := func(i int) error { // call backoff first. Someone may want an initial start delay t, err := callOpts.Backoff(ctx, req, i) if err != nil { return errors.InternalServerError("go.micro.client", err.Error()) } // only sleep if greater than 0 if t.Seconds() > 0 { time.Sleep(t) } if next == nil { var routes []string // lookup the route to send the reques to // TODO apply any filtering here routes, err = n.opts.Lookup(ctx, req, callOpts) if err != nil { return errors.InternalServerError("go.micro.client", err.Error()) } // balance the list of nodes next, err = callOpts.Selector.Select(routes) if err != nil { return err } } node := next() // make the call err = hcall(ctx, node, req, rsp, callOpts) // record the result of the call to inform future routing decisions if verr := n.opts.Selector.Record(node, err); verr != nil { return verr } // try and transform the error to a go-micro error if verr, ok := err.(*errors.Error); ok { return verr } return err } ch := make(chan error, callOpts.Retries) var gerr error ts := time.Now() endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint()) n.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", endpoint).Inc() for i := 0; i <= callOpts.Retries; i++ { go func() { ch <- call(i) }() select { case <-ctx.Done(): return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408) case err := <-ch: // if the call succeeded lets bail early if err == nil { return nil } retry, rerr := callOpts.Retry(ctx, req, i, err) if rerr != nil { return rerr } if !retry { return err } gerr = err } } if gerr != nil { n.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", endpoint, "status", "failure").Inc() } else { n.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", endpoint, "status", "success").Inc() } n.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", endpoint).Dec() te := time.Since(ts) n.opts.Meter.Summary(semconv.ClientRequestLatencyMicroseconds, "endpoint", endpoint).Update(te.Seconds()) n.opts.Meter.Histogram(semconv.ClientRequestDurationSeconds, "endpoint", endpoint).Update(te.Seconds()) return gerr } func (n *noopClient) call(ctx context.Context, addr string, req Request, rsp interface{}, opts CallOptions) error { return nil } func (n *noopClient) NewRequest(service, endpoint string, req interface{}, opts ...options.Option) Request { return &noopRequest{service: service, endpoint: endpoint} } func (n *noopClient) Stream(ctx context.Context, req Request, opts ...options.Option) (Stream, error) { var err error // make a copy of call opts callOpts := n.opts.CallOptions for _, o := range opts { o(&callOpts) } // check if we already have a deadline d, ok := ctx.Deadline() if !ok && callOpts.StreamTimeout > time.Duration(0) { var cancel context.CancelFunc // no deadline so we create a new one ctx, cancel = context.WithTimeout(ctx, callOpts.StreamTimeout) defer cancel() } else { // got a deadline so no need to setup context // but we need to set the timeout we pass along o := StreamTimeout(time.Until(d)) o(&callOpts) } // should we noop right here? select { case <-ctx.Done(): return nil, errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408) default: } /* // make copy of call method hstream := h.stream // wrap the call in reverse for i := len(callOpts.CallWrappers); i > 0; i-- { hstream = callOpts.CallWrappers[i-1](hstream) } */ // use the router passed as a call option, or fallback to the rpc clients router if callOpts.Router == nil { callOpts.Router = n.opts.Router } if callOpts.Selector == nil { callOpts.Selector = n.opts.Selector } // inject proxy address // TODO: don't even bother using Lookup/Select in this case if len(n.opts.Proxy) > 0 { callOpts.Address = []string{n.opts.Proxy} } var next selector.Next call := func(i int) (Stream, error) { // call backoff first. Someone may want an initial start delay t, cerr := callOpts.Backoff(ctx, req, i) if cerr != nil { return nil, errors.InternalServerError("go.micro.client", cerr.Error()) } // only sleep if greater than 0 if t.Seconds() > 0 { time.Sleep(t) } if next == nil { var routes []string // lookup the route to send the reques to // TODO apply any filtering here routes, err = n.opts.Lookup(ctx, req, callOpts) if err != nil { return nil, errors.InternalServerError("go.micro.client", err.Error()) } // balance the list of nodes next, err = callOpts.Selector.Select(routes) if err != nil { return nil, err } } node := next() // ts := time.Now() endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint()) n.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", endpoint).Inc() stream, cerr := n.stream(ctx, node, req, callOpts) if cerr != nil { n.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", endpoint, "status", "failure").Inc() } else { n.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", endpoint, "status", "success").Inc() } // record the result of the call to inform future routing decisions if verr := n.opts.Selector.Record(node, cerr); verr != nil { return nil, verr } // try and transform the error to a go-micro error if verr, ok := cerr.(*errors.Error); ok { return nil, verr } return stream, cerr } type response struct { stream Stream err error } ch := make(chan response, callOpts.Retries) var grr error for i := 0; i <= callOpts.Retries; i++ { go func() { s, cerr := call(i) ch <- response{s, cerr} }() select { case <-ctx.Done(): return nil, errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408) case rsp := <-ch: // if the call succeeded lets bail early if rsp.err == nil { return rsp.stream, nil } retry, rerr := callOpts.Retry(ctx, req, i, err) if rerr != nil { return nil, rerr } if !retry { return nil, rsp.err } grr = rsp.err } } return nil, grr } func (n *noopClient) stream(ctx context.Context, addr string, req Request, opts CallOptions) (*noopStream, error) { return &noopStream{}, nil }