micro/client/rpc_client.go

261 lines
6.8 KiB
Go
Raw Normal View History

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