unistack-org/micro
4
0
Fork 3
Code Issues 19 Pull Requests 1 Actions Packages Projects Releases 259 Wiki Activity

Improvements #130

Merged
vtolstov merged 14 commits from improvements into v3 2022-05-03 16:08:51 +03:00
Conversation 1 Commits 14 Files Changed 16 +53 -24
2 changed files with 53 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 80d342a72a - Show all commits

16
client/context_test.go
View File

@ -14,9 +14,25 @@ func TestFromContext(t *testing.T) {
}
}
func TestFromNilContext(t *testing.T) {
// nolint: staticcheck
c, ok := FromContext(nil)
if ok || c != nil {
t.Fatal("FromContext not works")
}
}
func TestNewContext(t *testing.T) {
ctx := NewContext(context.TODO(), NewClient())
c, ok := FromContext(ctx)
if c == nil || !ok {
t.Fatal("NewContext not works")
}
}
func TestNewNilContext(t *testing.T) {
// nolint: staticcheck
ctx := NewContext(nil, NewClient())
c, ok := FromContext(ctx)
if c == nil || !ok {
t.Fatal("NewContext not works")

61
client/noop.go
View File

@ -9,6 +9,7 @@ import (
"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
@ -233,18 +234,7 @@ func (n *noopClient) Call(ctx context.Context, req Request, rsp interface{}, opt
callOpts.Address = []string{n.opts.Proxy}
}
// 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
}
var next selector.Next
// return errors.New("go.micro.client", "request timeout", 408)
call := func(i int) error {
@ -259,6 +249,22 @@ func (n *noopClient) Call(ctx context.Context, req Request, rsp interface{}, opt
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
@ -323,6 +329,8 @@ func (n *noopClient) NewMessage(topic string, msg interface{}, opts ...MessageOp
}
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 {
@ -374,18 +382,7 @@ func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption
callOpts.Address = []string{n.opts.Proxy}
}
// 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
}
var next selector.Next
call := func(i int) (Stream, error) {
// call backoff first. Someone may want an initial start delay
@ -399,6 +396,22 @@ func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption
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)