2015-01-13 23:31:27 +00:00
|
|
|
package client
|
|
|
|
|
|
|
|
import (
|
2015-11-28 11:22:29 +00:00
|
|
|
"bytes"
|
2015-01-13 23:31:27 +00:00
|
|
|
"fmt"
|
2015-06-12 19:52:27 +01:00
|
|
|
"sync"
|
2015-01-13 23:31:27 +00:00
|
|
|
|
2015-11-20 16:17:33 +00:00
|
|
|
"github.com/micro/go-micro/broker"
|
2015-11-27 00:17:36 +00:00
|
|
|
"github.com/micro/go-micro/codec"
|
2015-11-20 16:17:33 +00:00
|
|
|
c "github.com/micro/go-micro/context"
|
|
|
|
"github.com/micro/go-micro/errors"
|
|
|
|
"github.com/micro/go-micro/registry"
|
|
|
|
"github.com/micro/go-micro/transport"
|
2015-05-23 11:53:40 +01:00
|
|
|
|
|
|
|
"golang.org/x/net/context"
|
2015-01-13 23:31:27 +00:00
|
|
|
)
|
|
|
|
|
2015-05-23 17:40:53 +01:00
|
|
|
type rpcClient struct {
|
2015-06-12 19:52:27 +01:00
|
|
|
once sync.Once
|
2015-05-21 19:24:57 +01:00
|
|
|
opts options
|
2015-05-20 22:57:19 +01:00
|
|
|
}
|
2015-01-13 23:31:27 +00:00
|
|
|
|
2015-05-23 17:40:53 +01:00
|
|
|
func newRpcClient(opt ...Option) Client {
|
2015-11-26 20:36:42 +00:00
|
|
|
var once sync.Once
|
|
|
|
|
2015-11-25 19:50:05 +00:00
|
|
|
opts := options{
|
2015-11-28 11:22:29 +00:00
|
|
|
codecs: make(map[string]codec.NewCodec),
|
2015-11-25 19:50:05 +00:00
|
|
|
}
|
2015-05-23 17:40:53 +01:00
|
|
|
|
|
|
|
for _, o := range opt {
|
|
|
|
o(&opts)
|
|
|
|
}
|
|
|
|
|
2015-11-25 19:50:05 +00:00
|
|
|
if len(opts.contentType) == 0 {
|
|
|
|
opts.contentType = defaultContentType
|
|
|
|
}
|
|
|
|
|
2015-12-09 00:02:45 +00:00
|
|
|
if opts.broker == nil {
|
|
|
|
opts.broker = broker.DefaultBroker
|
2015-05-23 17:40:53 +01:00
|
|
|
}
|
|
|
|
|
2015-12-07 23:56:17 +00:00
|
|
|
if opts.registry == nil {
|
|
|
|
opts.registry = registry.DefaultRegistry
|
|
|
|
}
|
|
|
|
|
2015-12-09 00:02:45 +00:00
|
|
|
if opts.selector == nil {
|
|
|
|
opts.selector = registry.NewRandomSelector(
|
|
|
|
registry.SelectorRegistry(opts.registry),
|
|
|
|
)
|
2015-06-12 19:52:27 +01:00
|
|
|
}
|
|
|
|
|
2015-12-09 00:02:45 +00:00
|
|
|
if opts.transport == nil {
|
|
|
|
opts.transport = transport.DefaultTransport
|
2015-12-07 21:09:10 +00:00
|
|
|
}
|
|
|
|
|
2015-11-26 20:36:42 +00:00
|
|
|
rc := &rpcClient{
|
2015-06-12 19:52:27 +01:00
|
|
|
once: once,
|
2015-05-23 17:40:53 +01:00
|
|
|
opts: opts,
|
|
|
|
}
|
2015-11-26 20:36:42 +00:00
|
|
|
|
|
|
|
c := Client(rc)
|
|
|
|
|
|
|
|
// wrap in reverse
|
|
|
|
for i := len(opts.wrappers); i > 0; i-- {
|
|
|
|
c = opts.wrappers[i-1](c)
|
|
|
|
}
|
|
|
|
|
|
|
|
return c
|
2015-05-23 17:40:53 +01:00
|
|
|
}
|
|
|
|
|
2015-11-28 11:22:29 +00:00
|
|
|
func (r *rpcClient) newCodec(contentType string) (codec.NewCodec, error) {
|
2015-11-27 00:17:36 +00:00
|
|
|
if c, ok := r.opts.codecs[contentType]; ok {
|
2015-11-28 11:22:29 +00:00
|
|
|
return c, nil
|
2015-11-25 19:50:05 +00:00
|
|
|
}
|
|
|
|
if cf, ok := defaultCodecs[contentType]; ok {
|
|
|
|
return cf, nil
|
|
|
|
}
|
|
|
|
return nil, fmt.Errorf("Unsupported Content-Type: %s", contentType)
|
|
|
|
}
|
|
|
|
|
2015-05-23 17:40:53 +01:00
|
|
|
func (r *rpcClient) call(ctx context.Context, address string, request Request, response interface{}) error {
|
2015-05-20 22:57:19 +01:00
|
|
|
msg := &transport.Message{
|
|
|
|
Header: make(map[string]string),
|
|
|
|
}
|
2015-01-13 23:31:27 +00:00
|
|
|
|
2015-05-26 22:39:48 +01:00
|
|
|
md, ok := c.GetMetadata(ctx)
|
2015-05-23 11:53:40 +01:00
|
|
|
if ok {
|
|
|
|
for k, v := range md {
|
|
|
|
msg.Header[k] = v
|
2015-05-20 22:57:19 +01:00
|
|
|
}
|
2015-01-13 23:31:27 +00:00
|
|
|
}
|
|
|
|
|
2015-05-20 22:57:19 +01:00
|
|
|
msg.Header["Content-Type"] = request.ContentType()
|
|
|
|
|
2015-11-28 11:22:29 +00:00
|
|
|
cf, err := r.newCodec(request.ContentType())
|
2015-11-25 19:50:05 +00:00
|
|
|
if err != nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", err.Error())
|
|
|
|
}
|
|
|
|
|
2015-05-21 21:08:19 +01:00
|
|
|
c, err := r.opts.transport.Dial(address)
|
2015-01-13 23:31:27 +00:00
|
|
|
if err != nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
|
|
|
|
}
|
2015-10-22 10:33:33 +01:00
|
|
|
defer c.Close()
|
2015-01-13 23:31:27 +00:00
|
|
|
|
2015-12-01 23:32:23 +00:00
|
|
|
client := newClientWithCodec(newRpcPlusCodec(msg, c, cf))
|
2015-10-22 15:14:56 +01:00
|
|
|
err = client.Call(ctx, request.Method(), request.Request(), response)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
return client.Close()
|
2015-06-01 18:55:27 +01:00
|
|
|
}
|
2015-01-13 23:31:27 +00:00
|
|
|
|
2015-06-01 18:55:27 +01:00
|
|
|
func (r *rpcClient) stream(ctx context.Context, address string, request Request, responseChan interface{}) (Streamer, error) {
|
|
|
|
msg := &transport.Message{
|
|
|
|
Header: make(map[string]string),
|
2015-01-13 23:31:27 +00:00
|
|
|
}
|
|
|
|
|
2015-06-01 18:55:27 +01:00
|
|
|
md, ok := c.GetMetadata(ctx)
|
|
|
|
if ok {
|
|
|
|
for k, v := range md {
|
|
|
|
msg.Header[k] = v
|
|
|
|
}
|
2015-01-13 23:31:27 +00:00
|
|
|
}
|
|
|
|
|
2015-06-01 18:55:27 +01:00
|
|
|
msg.Header["Content-Type"] = request.ContentType()
|
2015-01-13 23:31:27 +00:00
|
|
|
|
2015-11-28 11:22:29 +00:00
|
|
|
cf, err := r.newCodec(request.ContentType())
|
2015-11-25 19:50:05 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, errors.InternalServerError("go.micro.client", err.Error())
|
|
|
|
}
|
|
|
|
|
2015-06-01 18:55:27 +01:00
|
|
|
c, err := r.opts.transport.Dial(address, transport.WithStream())
|
2015-01-13 23:31:27 +00:00
|
|
|
if err != nil {
|
2015-06-01 18:55:27 +01:00
|
|
|
return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
|
2015-01-13 23:31:27 +00:00
|
|
|
}
|
|
|
|
|
2015-12-01 23:32:23 +00:00
|
|
|
client := newClientWithCodec(newRpcPlusCodec(msg, c, cf))
|
2015-06-01 18:55:27 +01:00
|
|
|
call := client.StreamGo(request.Method(), request.Request(), responseChan)
|
|
|
|
|
|
|
|
return &rpcStream{
|
|
|
|
request: request,
|
|
|
|
call: call,
|
|
|
|
client: client,
|
|
|
|
}, nil
|
2015-01-13 23:31:27 +00:00
|
|
|
}
|
|
|
|
|
2015-12-08 19:25:42 +00:00
|
|
|
func (r *rpcClient) CallRemote(ctx context.Context, address string, request Request, response interface{}, opts ...CallOption) error {
|
2015-05-23 11:53:40 +01:00
|
|
|
return r.call(ctx, address, request, response)
|
2015-01-13 23:31:27 +00:00
|
|
|
}
|
|
|
|
|
2015-12-08 19:25:42 +00:00
|
|
|
func (r *rpcClient) Call(ctx context.Context, request Request, response interface{}, opts ...CallOption) error {
|
2015-12-09 00:02:45 +00:00
|
|
|
var copts callOptions
|
|
|
|
for _, opt := range opts {
|
|
|
|
opt(&copts)
|
|
|
|
}
|
|
|
|
|
|
|
|
next, err := r.opts.selector.Select(request.Service(), copts.selectOptions...)
|
2015-12-09 12:44:28 +00:00
|
|
|
if err != nil && err == registry.ErrNotFound {
|
|
|
|
return errors.NotFound("go.micro.client", err.Error())
|
|
|
|
} else if err != nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", err.Error())
|
2015-12-09 00:02:45 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
node, err := next()
|
2015-12-09 12:44:28 +00:00
|
|
|
if err != nil && err == registry.ErrNotFound {
|
|
|
|
return errors.NotFound("go.micro.client", err.Error())
|
|
|
|
} else if err != nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", err.Error())
|
2015-01-13 23:31:27 +00:00
|
|
|
}
|
|
|
|
|
2015-05-25 22:14:28 +01:00
|
|
|
address := node.Address
|
|
|
|
if node.Port > 0 {
|
|
|
|
address = fmt.Sprintf("%s:%d", address, node.Port)
|
2015-05-21 19:24:57 +01:00
|
|
|
}
|
|
|
|
|
2015-12-07 23:56:17 +00:00
|
|
|
err = r.call(ctx, address, request, response)
|
2015-12-09 00:02:45 +00:00
|
|
|
r.opts.selector.Mark(request.Service(), node, err)
|
2015-12-07 23:56:17 +00:00
|
|
|
return err
|
2015-01-13 23:31:27 +00:00
|
|
|
}
|
|
|
|
|
2015-12-08 19:25:42 +00:00
|
|
|
func (r *rpcClient) StreamRemote(ctx context.Context, address string, request Request, responseChan interface{}, opts ...CallOption) (Streamer, error) {
|
2015-06-01 18:55:27 +01:00
|
|
|
return r.stream(ctx, address, request, responseChan)
|
|
|
|
}
|
|
|
|
|
2015-12-08 19:25:42 +00:00
|
|
|
func (r *rpcClient) Stream(ctx context.Context, request Request, responseChan interface{}, opts ...CallOption) (Streamer, error) {
|
2015-12-09 00:02:45 +00:00
|
|
|
var copts callOptions
|
|
|
|
for _, opt := range opts {
|
|
|
|
opt(&copts)
|
|
|
|
}
|
|
|
|
|
|
|
|
next, err := r.opts.selector.Select(request.Service(), copts.selectOptions...)
|
2015-12-09 12:44:28 +00:00
|
|
|
if err != nil && err == registry.ErrNotFound {
|
|
|
|
return nil, errors.NotFound("go.micro.client", err.Error())
|
|
|
|
} else if err != nil {
|
|
|
|
return nil, errors.InternalServerError("go.micro.client", err.Error())
|
2015-12-09 00:02:45 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
node, err := next()
|
2015-12-09 12:44:28 +00:00
|
|
|
if err != nil && err == registry.ErrNotFound {
|
|
|
|
return nil, errors.NotFound("go.micro.client", err.Error())
|
|
|
|
} else if err != nil {
|
|
|
|
return nil, errors.InternalServerError("go.micro.client", err.Error())
|
2015-06-01 18:55:27 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
address := node.Address
|
|
|
|
if node.Port > 0 {
|
|
|
|
address = fmt.Sprintf("%s:%d", address, node.Port)
|
|
|
|
}
|
|
|
|
|
2015-12-07 23:56:17 +00:00
|
|
|
stream, err := r.stream(ctx, address, request, responseChan)
|
2015-12-09 00:02:45 +00:00
|
|
|
r.opts.selector.Mark(request.Service(), node, err)
|
2015-12-07 23:56:17 +00:00
|
|
|
return stream, err
|
2015-06-01 18:55:27 +01:00
|
|
|
}
|
|
|
|
|
2015-12-08 19:25:42 +00:00
|
|
|
func (r *rpcClient) Publish(ctx context.Context, p Publication, opts ...PublishOption) error {
|
2015-06-12 19:52:27 +01:00
|
|
|
md, ok := c.GetMetadata(ctx)
|
|
|
|
if !ok {
|
|
|
|
md = make(map[string]string)
|
|
|
|
}
|
|
|
|
md["Content-Type"] = p.ContentType()
|
|
|
|
|
|
|
|
// encode message body
|
2015-11-28 11:22:29 +00:00
|
|
|
cf, err := r.newCodec(p.ContentType())
|
|
|
|
if err != nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", err.Error())
|
|
|
|
}
|
|
|
|
b := &buffer{bytes.NewBuffer(nil)}
|
|
|
|
if err := cf(b).Write(&codec.Message{Type: codec.Publication}, p.Message()); err != nil {
|
|
|
|
return errors.InternalServerError("go.micro.client", err.Error())
|
2015-06-12 19:52:27 +01:00
|
|
|
}
|
|
|
|
r.once.Do(func() {
|
|
|
|
r.opts.broker.Connect()
|
|
|
|
})
|
|
|
|
|
|
|
|
return r.opts.broker.Publish(p.Topic(), &broker.Message{
|
|
|
|
Header: md,
|
2015-11-28 11:22:29 +00:00
|
|
|
Body: b.Bytes(),
|
2015-06-12 19:52:27 +01:00
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
func (r *rpcClient) NewPublication(topic string, message interface{}) Publication {
|
2015-11-25 19:50:05 +00:00
|
|
|
return newRpcPublication(topic, message, r.opts.contentType)
|
2015-06-12 19:52:27 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
func (r *rpcClient) NewProtoPublication(topic string, message interface{}) Publication {
|
|
|
|
return newRpcPublication(topic, message, "application/octet-stream")
|
|
|
|
}
|
2015-05-23 17:40:53 +01:00
|
|
|
func (r *rpcClient) NewRequest(service, method string, request interface{}) Request {
|
2015-11-25 19:50:05 +00:00
|
|
|
return newRpcRequest(service, method, request, r.opts.contentType)
|
2015-01-13 23:31:27 +00:00
|
|
|
}
|
|
|
|
|
2015-05-23 17:40:53 +01:00
|
|
|
func (r *rpcClient) NewProtoRequest(service, method string, request interface{}) Request {
|
2015-01-13 23:31:27 +00:00
|
|
|
return newRpcRequest(service, method, request, "application/octet-stream")
|
|
|
|
}
|
|
|
|
|
2015-05-23 17:40:53 +01:00
|
|
|
func (r *rpcClient) NewJsonRequest(service, method string, request interface{}) Request {
|
2015-01-13 23:31:27 +00:00
|
|
|
return newRpcRequest(service, method, request, "application/json")
|
|
|
|
}
|