micro/client/noop.go
Vasiliy Tolstov 8863c10ef4
Some checks failed
lint / lint (pull_request) Failing after 1m29s
pr / test (pull_request) Failing after 2m36s
client/noop: fix metadata overwrite
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-12-21 00:06:56 +03:00

540 lines
11 KiB
Go

package client
import (
"context"
"fmt"
"time"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/errors"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/selector"
)
// 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 noopMessage struct {
topic string
payload interface{}
opts MessageOptions
}
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 ...Option) Client {
nc := &noopClient{opts: NewOptions(opts...)}
// wrap in reverse
c := Client(nc)
for i := len(nc.opts.Wrappers); i > 0; i-- {
c = nc.opts.Wrappers[i-1](c)
}
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 *noopMessage) Topic() string {
return n.topic
}
func (n *noopMessage) Payload() interface{} {
return n.payload
}
func (n *noopMessage) ContentType() string {
return n.opts.ContentType
}
func (n *noopMessage) Metadata() metadata.Metadata {
return n.opts.Metadata
}
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 ...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 ...CallOption) 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 := WithRequestTimeout(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
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
}
}
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 ...RequestOption) Request {
return &noopRequest{service: service, endpoint: endpoint}
}
func (n *noopClient) NewMessage(topic string, msg interface{}, opts ...MessageOption) Message {
options := NewMessageOptions(append([]MessageOption{MessageContentType(n.opts.ContentType)}, opts...)...)
return &noopMessage{topic: topic, payload: msg, opts: options}
}
func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption) (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 := WithStreamTimeout(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()
stream, cerr := n.stream(ctx, node, req, callOpts)
// 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) (Stream, error) {
return &noopStream{}, nil
}
func (n *noopClient) BatchPublish(ctx context.Context, ps []Message, opts ...PublishOption) error {
return n.publish(ctx, ps, opts...)
}
func (n *noopClient) Publish(ctx context.Context, p Message, opts ...PublishOption) error {
return n.publish(ctx, []Message{p}, opts...)
}
func (n *noopClient) publish(ctx context.Context, ps []Message, opts ...PublishOption) error {
options := NewPublishOptions(opts...)
msgs := make([]*broker.Message, 0, len(ps))
for _, p := range ps {
omd, ok := metadata.FromOutgoingContext(ctx)
if !ok {
omd = metadata.New(0)
}
md := metadata.Copy(omd)
iter := p.Metadata().Iterator()
var k, v string
for iter.Next(&k, &v) {
md.Set(k, v)
}
md[metadata.HeaderContentType] = p.ContentType()
topic := p.Topic()
// get the exchange
if len(options.Exchange) > 0 {
topic = options.Exchange
}
md[metadata.HeaderTopic] = topic
var body []byte
// passed in raw data
if d, ok := p.Payload().(*codec.Frame); ok {
body = d.Data
} else {
// use codec for payload
cf, err := n.newCodec(p.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
// set the body
b, err := cf.Marshal(p.Payload())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
body = b
}
msgs = append(msgs, &broker.Message{Header: md, Body: body})
}
return n.opts.Broker.BatchPublish(ctx, msgs,
broker.PublishContext(options.Context),
broker.PublishBodyOnly(options.BodyOnly),
)
}