update go-micro and fix client/selector usage

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2020-07-17 01:29:28 +03:00
parent d14ddecb6f
commit 280d3a64c9
4 changed files with 68 additions and 87 deletions

3
go.mod
View File

@ -4,8 +4,7 @@ go 1.13
require ( require (
github.com/golang/protobuf v1.4.0 github.com/golang/protobuf v1.4.0
github.com/micro/go-micro/v2 v2.9.1-0.20200716123506-3627e47f04eb github.com/micro/go-micro/v2 v2.9.1-0.20200716153311-f9bf56239306
github.com/teris-io/shortid v0.0.0-20171029131806-771a37caa5cf // indirect
) )
replace github.com/coreos/etcd => github.com/ozonru/etcd v3.3.20-grpc1.27-origmodule+incompatible replace github.com/coreos/etcd => github.com/ozonru/etcd v3.3.20-grpc1.27-origmodule+incompatible

10
go.sum
View File

@ -59,6 +59,7 @@ github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngE
github.com/blang/semver v3.1.0+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk= github.com/blang/semver v3.1.0+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY= github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4= github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
github.com/bradfitz/gomemcache v0.0.0-20190913173617-a41fca850d0b/go.mod h1:H0wQNHz2YrLsuXOZozoeDmnHXkNCRmMW0gwFWDfEZDA=
github.com/bwmarrin/discordgo v0.20.2 h1:nA7jiTtqUA9lT93WL2jPjUp8ZTEInRujBdx1C9gkr20= github.com/bwmarrin/discordgo v0.20.2 h1:nA7jiTtqUA9lT93WL2jPjUp8ZTEInRujBdx1C9gkr20=
github.com/bwmarrin/discordgo v0.20.2/go.mod h1:O9S4p+ofTFwB02em7jkpkV8M3R0/PUVOwN61zSZ0r4Q= github.com/bwmarrin/discordgo v0.20.2/go.mod h1:O9S4p+ofTFwB02em7jkpkV8M3R0/PUVOwN61zSZ0r4Q=
github.com/caddyserver/certmagic v0.10.6/go.mod h1:Y8jcUBctgk/IhpAzlHKfimZNyXCkfGgRTC0orl8gROQ= github.com/caddyserver/certmagic v0.10.6/go.mod h1:Y8jcUBctgk/IhpAzlHKfimZNyXCkfGgRTC0orl8gROQ=
@ -80,8 +81,6 @@ github.com/containerd/ttrpc v0.0.0-20190828154514-0e0f228740de/go.mod h1:PvCDdDG
github.com/containerd/typeurl v0.0.0-20180627222232-a93fcdb778cd/go.mod h1:Cm3kwCdlkCfMSHURc+r6fwoGH6/F1hH3S4sg0rLFWPc= github.com/containerd/typeurl v0.0.0-20180627222232-a93fcdb778cd/go.mod h1:Cm3kwCdlkCfMSHURc+r6fwoGH6/F1hH3S4sg0rLFWPc=
github.com/coreos/bbolt v1.3.3 h1:n6AiVyVRKQFNb6mJlwESEvvLoDyiTzXX7ORAUlkeBdY= github.com/coreos/bbolt v1.3.3 h1:n6AiVyVRKQFNb6mJlwESEvvLoDyiTzXX7ORAUlkeBdY=
github.com/coreos/bbolt v1.3.3/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk= github.com/coreos/bbolt v1.3.3/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
github.com/coreos/etcd v3.3.18+incompatible h1:Zz1aXgDrFFi1nadh58tA9ktt06cmPTwNNP3dXwIq1lE=
github.com/coreos/etcd v3.3.18+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM= github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM=
github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk= github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4= github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
@ -284,11 +283,8 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0j
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
github.com/micro/cli/v2 v2.1.2 h1:43J1lChg/rZCC1rvdqZNFSQDrGT7qfMrtp6/ztpIkEM= github.com/micro/cli/v2 v2.1.2 h1:43J1lChg/rZCC1rvdqZNFSQDrGT7qfMrtp6/ztpIkEM=
github.com/micro/cli/v2 v2.1.2/go.mod h1:EguNh6DAoWKm9nmk+k/Rg0H3lQnDxqzu5x5srOtGtYg= github.com/micro/cli/v2 v2.1.2/go.mod h1:EguNh6DAoWKm9nmk+k/Rg0H3lQnDxqzu5x5srOtGtYg=
github.com/micro/go-micro v1.18.0 h1:gP70EZVHpJuUIT0YWth192JmlIci+qMOEByHm83XE9E= github.com/micro/go-micro/v2 v2.9.1-0.20200716153311-f9bf56239306 h1:zm/cCJwRAySbM5DZdeqH4vf0F4Lvfe/XqC8AB9Vu5ow=
github.com/micro/go-micro/v2 v2.9.1-0.20200716123506-3627e47f04eb h1:+S2buLNVdGhMvp0NiGtfhmL8WKsa/fbsEpTL3GGavaI= github.com/micro/go-micro/v2 v2.9.1-0.20200716153311-f9bf56239306/go.mod h1:JgTt07BfD2x4mcTm/qZi8HZnxopoBYJxfpIETI8MKq8=
github.com/micro/go-micro/v2 v2.9.1-0.20200716123506-3627e47f04eb/go.mod h1:Szpx+Q9oZvNOoGc1cPweBt3PozVX4e/z3SC1hpxV4iw=
github.com/micro/go-micro/v2 v2.9.1 h1:+S9koIrNWARjpP6k2TZ7kt0uC9zUJtNXzIdZTZRms7Q=
github.com/micro/go-micro/v2 v2.9.1/go.mod h1:x55ZM3Puy0FyvvkR3e0ha0xsE9DFwfPSUMWAIbFY0SY=
github.com/miekg/dns v1.1.15/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg= github.com/miekg/dns v1.1.15/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
github.com/miekg/dns v1.1.27 h1:aEH/kqUzUxGJ/UHcEKdJY+ugH6WEzsEBBSPa8zuy1aM= github.com/miekg/dns v1.1.27 h1:aEH/kqUzUxGJ/UHcEKdJY+ugH6WEzsEBBSPa8zuy1aM=
github.com/miekg/dns v1.1.27/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM= github.com/miekg/dns v1.1.27/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=

131
http.go
View File

@ -16,16 +16,22 @@ import (
"github.com/micro/go-micro/v2/broker" "github.com/micro/go-micro/v2/broker"
"github.com/micro/go-micro/v2/client" "github.com/micro/go-micro/v2/client"
"github.com/micro/go-micro/v2/client/selector" "github.com/micro/go-micro/v2/cmd"
"github.com/micro/go-micro/v2/codec" "github.com/micro/go-micro/v2/codec"
raw "github.com/micro/go-micro/v2/codec/bytes" raw "github.com/micro/go-micro/v2/codec/bytes"
"github.com/micro/go-micro/v2/cmd"
errors "github.com/micro/go-micro/v2/errors" errors "github.com/micro/go-micro/v2/errors"
"github.com/micro/go-micro/v2/metadata" "github.com/micro/go-micro/v2/metadata"
"github.com/micro/go-micro/v2/registry" "github.com/micro/go-micro/v2/registry"
"github.com/micro/go-micro/v2/router"
"github.com/micro/go-micro/v2/selector"
"github.com/micro/go-micro/v2/transport" "github.com/micro/go-micro/v2/transport"
) )
func filterLabel(r []router.Route) []router.Route {
// selector.FilterLabel("protocol", "http")
return r
}
type httpClient struct { type httpClient struct {
once sync.Once once sync.Once
opts client.Options opts client.Options
@ -35,47 +41,6 @@ func init() {
cmd.DefaultClients["http"] = NewClient cmd.DefaultClients["http"] = NewClient
} }
func (h *httpClient) next(request client.Request, opts client.CallOptions) (selector.Next, error) {
service := request.Service()
// get proxy
if prx := os.Getenv("MICRO_PROXY"); len(prx) > 0 {
service = prx
}
// get proxy address
if prx := os.Getenv("MICRO_PROXY_ADDRESS"); len(prx) > 0 {
opts.Address = []string{prx}
}
// return remote address
if len(opts.Address) > 0 {
return func() (*registry.Node, error) {
return &registry.Node{
Address: opts.Address[0],
Metadata: map[string]string{
"protocol": "http",
},
}, nil
}, nil
}
// only get the things that are of mucp protocol
selectOptions := append(opts.SelectOptions, selector.WithFilter(
selector.FilterLabel("protocol", "http"),
))
// get next nodes from the selector
next, err := h.opts.Selector.Select(service, selectOptions...)
if err != nil && err == selector.ErrNotFound {
return nil, errors.NotFound("go.micro.client", err.Error())
} else if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
}
return next, nil
}
func (h *httpClient) call(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error { func (h *httpClient) call(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
// set the address // set the address
address := node.Address address := node.Address
@ -221,12 +186,6 @@ func (h *httpClient) Call(ctx context.Context, req client.Request, rsp interface
opt(&callOpts) opt(&callOpts)
} }
// get next nodes from the selector
next, err := h.next(req, callOpts)
if err != nil {
return err
}
// check if we already have a deadline // check if we already have a deadline
d, ok := ctx.Deadline() d, ok := ctx.Deadline()
if !ok { if !ok {
@ -267,17 +226,33 @@ func (h *httpClient) Call(ctx context.Context, req client.Request, rsp interface
time.Sleep(t) time.Sleep(t)
} }
// select next node // use the router passed as a call option, or fallback to the rpc clients router
node, err := next() if callOpts.Router == nil {
if err != nil && err == selector.ErrNotFound { callOpts.Router = h.opts.Router
return errors.NotFound("go.micro.client", err.Error())
} else if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
} }
// use the selector passed as a call option, or fallback to the rpc clients selector
if callOpts.Selector == nil {
callOpts.Selector = h.opts.Selector
}
callOpts.SelectOptions = append(callOpts.SelectOptions, selector.WithFilter(filterLabel))
// lookup the route to send the request via
route, err := client.LookupRoute(req, callOpts)
if err != nil {
return err
}
// pass a node to enable backwards compatability as changing the
// call func would be a breaking change.
// todo v3: change the call func to accept a route
node := &registry.Node{Address: route.Address, Metadata: route.Metadata}
node.Metadata["protocol"] = "http"
// make the call // make the call
err = hcall(ctx, node, req, rsp, callOpts) err = hcall(ctx, node, req, rsp, callOpts)
h.opts.Selector.Mark(req.Service(), node, err) h.opts.Selector.Record(*route, err)
return err return err
} }
@ -321,12 +296,6 @@ func (h *httpClient) Stream(ctx context.Context, req client.Request, opts ...cli
opt(&callOpts) opt(&callOpts)
} }
// get next nodes from the selector
next, err := h.next(req, callOpts)
if err != nil {
return nil, err
}
// check if we already have a deadline // check if we already have a deadline
d, ok := ctx.Deadline() d, ok := ctx.Deadline()
if !ok { if !ok {
@ -358,15 +327,32 @@ func (h *httpClient) Stream(ctx context.Context, req client.Request, opts ...cli
time.Sleep(t) time.Sleep(t)
} }
node, err := next() // use the router passed as a call option, or fallback to the rpc clients router
if err != nil && err == selector.ErrNotFound { if callOpts.Router == nil {
return nil, errors.NotFound("go.micro.client", err.Error()) callOpts.Router = h.opts.Router
} else if err != nil { }
return nil, errors.InternalServerError("go.micro.client", err.Error()) // use the selector passed as a call option, or fallback to the rpc clients selector
if callOpts.Selector == nil {
callOpts.Selector = h.opts.Selector
} }
callOpts.SelectOptions = append(callOpts.SelectOptions, selector.WithFilter(filterLabel))
// lookup the route to send the request via
route, err := client.LookupRoute(req, callOpts)
if err != nil {
return nil, err
}
// pass a node to enable backwards compatability as changing the
// call func would be a breaking change.
// todo v3: change the call func to accept a route
node := &registry.Node{Address: route.Address, Metadata: route.Metadata}
node.Metadata["protocol"] = "http"
stream, err := h.stream(ctx, node, req, callOpts) stream, err := h.stream(ctx, node, req, callOpts)
h.opts.Selector.Mark(req.Service(), node, err) h.opts.Selector.Record(*route, err)
return stream, err return stream, err
} }
@ -377,6 +363,7 @@ func (h *httpClient) Stream(ctx context.Context, req client.Request, opts ...cli
ch := make(chan response, callOpts.Retries) ch := make(chan response, callOpts.Retries)
var grr error var grr error
var err error
for i := 0; i < callOpts.Retries; i++ { for i := 0; i < callOpts.Retries; i++ {
go func() { go func() {
@ -491,14 +478,12 @@ func newClient(opts ...client.Option) client.Client {
options.Broker = broker.DefaultBroker options.Broker = broker.DefaultBroker
} }
if options.Registry == nil { if options.Router == nil {
options.Registry = registry.DefaultRegistry options.Router = router.DefaultRouter
} }
if options.Selector == nil { if options.Selector == nil {
options.Selector = selector.NewSelector( options.Selector = selector.DefaultSelector
selector.Registry(options.Registry),
)
} }
rc := &httpClient{ rc := &httpClient{

View File

@ -11,15 +11,16 @@ import (
"testing" "testing"
"github.com/micro/go-micro/v2/client" "github.com/micro/go-micro/v2/client"
"github.com/micro/go-micro/v2/client/selector"
"github.com/micro/go-micro/v2/registry" "github.com/micro/go-micro/v2/registry"
"github.com/micro/go-micro/v2/registry/memory" "github.com/micro/go-micro/v2/registry/memory"
"github.com/micro/go-micro/v2/router"
rrouter "github.com/micro/go-micro/v2/router/registry"
"github.com/micro/go-plugins/client/http/v2/test" "github.com/micro/go-plugins/client/http/v2/test"
) )
func TestHTTPClient(t *testing.T) { func TestHTTPClient(t *testing.T) {
r := memory.NewRegistry() r := memory.NewRegistry()
s := selector.NewSelector(selector.Registry(r)) s := rrouter.NewRouter(router.Registry(r))
l, err := net.Listen("tcp", "127.0.0.1:0") l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil { if err != nil {
@ -82,7 +83,7 @@ func TestHTTPClient(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
c := NewClient(client.Selector(s)) c := NewClient(client.Router(s))
for i := 0; i < 10; i++ { for i := 0; i < 10; i++ {
msg := &test.Message{ msg := &test.Message{
@ -103,7 +104,7 @@ func TestHTTPClient(t *testing.T) {
func TestHTTPClientStream(t *testing.T) { func TestHTTPClientStream(t *testing.T) {
r := memory.NewRegistry() r := memory.NewRegistry()
s := selector.NewSelector(selector.Registry(r)) s := rrouter.NewRouter(router.Registry(r))
l, err := net.Listen("tcp", "127.0.0.1:0") l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil { if err != nil {
@ -241,7 +242,7 @@ func TestHTTPClientStream(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
c := NewClient(client.Selector(s)) c := NewClient(client.Router(s))
req := c.NewRequest("test.service", "/foo/bar", new(test.Message)) req := c.NewRequest("test.service", "/foo/bar", new(test.Message))
stream, err := c.Stream(context.TODO(), req) stream, err := c.Stream(context.TODO(), req)
if err != nil { if err != nil {