micro/client/rpc_client.go

448 lines
11 KiB
Go
Raw Normal View History

2015-01-13 23:31:27 +00:00
package client
import (
"bytes"
2015-01-13 23:31:27 +00:00
"fmt"
"sync"
2016-01-03 21:14:33 +00:00
"time"
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
"github.com/micro/go-micro/errors"
2016-01-28 17:55:28 +00:00
"github.com/micro/go-micro/metadata"
2015-12-09 19:23:16 +00:00
"github.com/micro/go-micro/selector"
2015-11-20 16:17:33 +00:00
"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 {
once sync.Once
opts Options
2016-05-13 15:58:53 +01:00
pool *pool
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 {
2016-01-02 23:16:15 +00:00
opts := newOptions(opt...)
2015-11-26 20:36:42 +00:00
rc := &rpcClient{
2016-05-13 15:58:53 +01:00
once: sync.Once{},
2015-05-23 17:40:53 +01:00
opts: opts,
2016-06-07 00:46:14 +01:00
pool: newPool(opts.PoolSize, opts.PoolTTL),
2015-05-23 17:40:53 +01:00
}
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)
2015-11-26 20:36:42 +00:00
}
return c
2015-05-23 17:40:53 +01:00
}
func (r *rpcClient) newCodec(contentType string) (codec.NewCodec, error) {
if c, ok := r.opts.Codecs[contentType]; ok {
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)
}
func (r *rpcClient) call(ctx context.Context, address string, req Request, resp interface{}, opts CallOptions) 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
2016-01-28 17:55:28 +00:00
md, ok := metadata.FromContext(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
}
2016-05-12 23:32:58 +01:00
// set timeout in nanoseconds
msg.Header["Timeout"] = fmt.Sprintf("%d", opts.RequestTimeout)
// set the content type for the request
2016-01-05 21:13:20 +00:00
msg.Header["Content-Type"] = req.ContentType()
2015-05-20 22:57:19 +01:00
2016-01-05 21:13:20 +00:00
cf, err := r.newCodec(req.ContentType())
2015-11-25 19:50:05 +00:00
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
2016-05-13 15:58:53 +01:00
var grr error
c, err := r.pool.getConn(address, r.opts.Transport, transport.WithTimeout(opts.DialTimeout))
2015-01-13 23:31:27 +00:00
if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
}
2016-05-13 15:58:53 +01:00
defer func() {
// defer execution of release
r.pool.release(address, c, grr)
}()
2015-01-13 23:31:27 +00:00
2016-01-05 21:13:20 +00:00
stream := &rpcStream{
context: ctx,
request: req,
closed: make(chan bool),
codec: newRpcPlusCodec(msg, c, cf),
}
2016-05-12 23:32:58 +01:00
defer stream.Close()
2016-01-03 21:14:33 +00:00
ch := make(chan error, 1)
go func() {
2016-01-05 21:13:20 +00:00
// send request
if err := stream.Send(req.Request()); err != nil {
ch <- err
return
2016-01-03 21:14:33 +00:00
}
2016-01-05 21:13:20 +00:00
// recv request
if err := stream.Recv(resp); err != nil {
ch <- err
return
}
// success
ch <- nil
2016-01-03 21:14:33 +00:00
}()
select {
2016-05-12 23:32:58 +01:00
case err := <-ch:
2016-05-13 15:58:53 +01:00
grr = err
2016-05-12 23:32:58 +01:00
return err
case <-ctx.Done():
2016-05-13 15:58:53 +01:00
grr = ctx.Err()
return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
2015-10-22 15:14:56 +01:00
}
}
2015-01-13 23:31:27 +00:00
func (r *rpcClient) stream(ctx context.Context, address string, req Request, opts CallOptions) (Streamer, error) {
msg := &transport.Message{
Header: make(map[string]string),
2015-01-13 23:31:27 +00:00
}
2016-01-28 17:55:28 +00:00
md, ok := metadata.FromContext(ctx)
if ok {
for k, v := range md {
msg.Header[k] = v
}
2015-01-13 23:31:27 +00:00
}
2016-05-12 23:32:58 +01:00
// set timeout in nanoseconds
msg.Header["Timeout"] = fmt.Sprintf("%d", opts.RequestTimeout)
// set the content type for the request
2015-12-17 20:37:35 +00:00
msg.Header["Content-Type"] = req.ContentType()
2015-01-13 23:31:27 +00:00
2015-12-17 20:37:35 +00:00
cf, err := r.newCodec(req.ContentType())
2015-11-25 19:50:05 +00:00
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
}
c, err := r.opts.Transport.Dial(address, transport.WithStream(), transport.WithTimeout(opts.DialTimeout))
2015-01-13 23:31:27 +00:00
if err != nil {
return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
2015-01-13 23:31:27 +00:00
}
2015-12-18 01:21:56 +00:00
stream := &rpcStream{
2015-12-17 20:37:35 +00:00
context: ctx,
request: req,
closed: make(chan bool),
2015-12-18 01:21:56 +00:00
codec: newRpcPlusCodec(msg, c, cf),
}
2016-01-03 21:14:33 +00:00
ch := make(chan error, 1)
go func() {
2016-01-05 21:13:20 +00:00
ch <- stream.Send(req.Request())
2016-01-03 21:14:33 +00:00
}()
2016-05-12 23:32:58 +01:00
var grr error
2016-01-03 21:14:33 +00:00
select {
2016-05-12 23:32:58 +01:00
case err := <-ch:
grr = err
case <-ctx.Done():
grr = errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
2016-01-03 21:14:33 +00:00
}
2016-05-12 23:32:58 +01:00
if grr != nil {
stream.Close()
return nil, grr
}
return stream, nil
2015-01-13 23:31:27 +00:00
}
2016-01-02 19:12:17 +00:00
func (r *rpcClient) Init(opts ...Option) error {
2016-06-07 00:46:14 +01:00
size := r.opts.PoolSize
ttl := r.opts.PoolTTL
2016-01-02 19:12:17 +00:00
for _, o := range opts {
o(&r.opts)
}
2016-06-07 00:46:14 +01:00
// recreate the pool if the options changed
if size != r.opts.PoolSize || ttl != r.opts.PoolTTL {
r.pool = newPool(r.opts.PoolSize, r.opts.PoolTTL)
}
2016-01-02 19:12:17 +00:00
return nil
}
func (r *rpcClient) Options() Options {
return r.opts
}
2015-12-08 19:25:42 +00:00
func (r *rpcClient) CallRemote(ctx context.Context, address string, request Request, response interface{}, opts ...CallOption) error {
// make a copy of call opts
callOpts := r.opts.CallOptions
for _, opt := range opts {
opt(&callOpts)
}
return r.call(ctx, address, request, response, callOpts)
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 {
// make a copy of call opts
callOpts := r.opts.CallOptions
2015-12-09 00:02:45 +00:00
for _, opt := range opts {
opt(&callOpts)
2015-12-09 00:02:45 +00:00
}
2016-05-12 23:32:58 +01:00
// get next nodes from the selector
next, err := r.opts.Selector.Select(request.Service(), callOpts.SelectOptions...)
2015-12-09 19:23:16 +00:00
if err != nil && err == selector.ErrNotFound {
2015-12-09 12:44:28 +00:00
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
}
2016-05-12 23:32:58 +01:00
// check if we already have a deadline
d, ok := ctx.Deadline()
if !ok {
// no deadline so we create a new one
ctx, _ = context.WithTimeout(ctx, callOpts.RequestTimeout)
} else {
// got a deadline so no need to setup context
// but we need to set the timeout we pass along
opt := WithRequestTimeout(d.Sub(time.Now()))
opt(&callOpts)
}
2016-01-03 21:14:33 +00:00
// should we noop right here?
select {
case <-ctx.Done():
return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
default:
}
2016-05-12 23:32:58 +01:00
// return errors.New("go.micro.client", "request timeout", 408)
call := func(i int) error {
2016-04-05 20:04:37 +01:00
// call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, request, 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)
}
2016-05-12 23:32:58 +01:00
// select next node
2016-01-02 23:16:15 +00:00
node, err := next()
if err != nil && err == selector.ErrNotFound {
return errors.NotFound("go.micro.client", err.Error())
} else if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
2016-05-12 23:32:58 +01:00
// set the address
2016-01-02 23:16:15 +00:00
address := node.Address
if node.Port > 0 {
address = fmt.Sprintf("%s:%d", address, node.Port)
}
2016-05-12 23:32:58 +01:00
// make the call
err = r.call(ctx, address, request, response, callOpts)
r.opts.Selector.Mark(request.Service(), node, err)
return err
}
ch := make(chan error, callOpts.Retries)
var gerr error
2015-01-13 23:31:27 +00:00
2016-05-12 23:32:58 +01:00
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)
2016-05-12 23:32:58 +01:00
case err := <-ch:
// if the call succeeded lets bail early
if err == nil {
return nil
}
gerr = err
2016-01-02 23:16:15 +00:00
}
2015-05-21 19:24:57 +01:00
}
2016-05-12 23:32:58 +01:00
return gerr
2015-01-13 23:31:27 +00:00
}
2015-12-17 20:37:35 +00:00
func (r *rpcClient) StreamRemote(ctx context.Context, address string, request Request, opts ...CallOption) (Streamer, error) {
// make a copy of call opts
callOpts := r.opts.CallOptions
for _, opt := range opts {
opt(&callOpts)
}
return r.stream(ctx, address, request, callOpts)
}
2015-12-17 20:37:35 +00:00
func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOption) (Streamer, error) {
// make a copy of call opts
callOpts := r.opts.CallOptions
2015-12-09 00:02:45 +00:00
for _, opt := range opts {
opt(&callOpts)
2015-12-09 00:02:45 +00:00
}
2016-05-12 23:32:58 +01:00
// get next nodes from the selector
next, err := r.opts.Selector.Select(request.Service(), callOpts.SelectOptions...)
2015-12-09 19:23:16 +00:00
if err != nil && err == selector.ErrNotFound {
2015-12-09 12:44:28 +00: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 00:02:45 +00:00
}
2016-05-12 23:32:58 +01:00
// check if we already have a deadline
d, ok := ctx.Deadline()
if !ok {
// no deadline so we create a new one
ctx, _ = context.WithTimeout(ctx, callOpts.RequestTimeout)
} else {
// got a deadline so no need to setup context
// but we need to set the timeout we pass along
opt := WithRequestTimeout(d.Sub(time.Now()))
opt(&callOpts)
}
2016-01-02 23:16:15 +00:00
// should we noop right here?
select {
case <-ctx.Done():
return nil, errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
default:
}
2016-05-12 23:32:58 +01:00
call := func(i int) (Streamer, error) {
2016-04-05 20:04:37 +01:00
// call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, request, i)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
}
// only sleep if greater than 0
if t.Seconds() > 0 {
time.Sleep(t)
}
2016-01-02 23:16:15 +00:00
node, err := next()
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())
}
2016-01-02 23:16:15 +00:00
address := node.Address
if node.Port > 0 {
address = fmt.Sprintf("%s:%d", address, node.Port)
}
2016-05-12 23:32:58 +01:00
stream, err := r.stream(ctx, address, request, callOpts)
r.opts.Selector.Mark(request.Service(), node, err)
return stream, err
}
type response struct {
stream Streamer
err error
}
2016-01-02 23:16:15 +00:00
2016-05-12 23:32:58 +01:00
ch := make(chan response, callOpts.Retries)
var grr error
for i := 0; i < callOpts.Retries; i++ {
go func() {
s, err := call(i)
ch <- response{s, err}
}()
select {
case <-ctx.Done():
return nil, errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
2016-05-12 23:32:58 +01:00
case rsp := <-ch:
// if the call succeeded lets bail early
if rsp.err == nil {
return rsp.stream, nil
}
grr = rsp.err
2016-01-02 23:16:15 +00:00
}
}
2016-05-12 23:32:58 +01:00
return nil, grr
}
2015-12-08 19:25:42 +00:00
func (r *rpcClient) Publish(ctx context.Context, p Publication, opts ...PublishOption) error {
2016-01-28 17:55:28 +00:00
md, ok := metadata.FromContext(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 {
return newRpcPublication(topic, message, r.opts.ContentType)
}
func (r *rpcClient) NewProtoPublication(topic string, message interface{}) Publication {
return newRpcPublication(topic, message, "application/octet-stream")
}
2015-12-17 20:37:35 +00:00
func (r *rpcClient) NewRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
return newRpcRequest(service, method, request, r.opts.ContentType, reqOpts...)
2015-01-13 23:31:27 +00:00
}
2015-12-17 20:37:35 +00:00
func (r *rpcClient) NewProtoRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
return newRpcRequest(service, method, request, "application/octet-stream", reqOpts...)
2015-01-13 23:31:27 +00:00
}
2015-12-17 20:37:35 +00:00
func (r *rpcClient) NewJsonRequest(service, method string, request interface{}, reqOpts ...RequestOption) Request {
return newRpcRequest(service, method, request, "application/json", reqOpts...)
2015-01-13 23:31:27 +00:00
}
2015-12-19 21:56:14 +00:00
func (r *rpcClient) String() string {
return "rpc"
}