micro/client/rpc_client.go

188 lines
4.6 KiB
Go
Raw Normal View History

2015-01-13 23:31:27 +00:00
package client
import (
"bytes"
"fmt"
"math/rand"
"net/http"
"time"
2015-05-23 11:53:40 +01:00
c "github.com/myodc/go-micro/context"
2015-05-05 19:05:06 +01:00
"github.com/myodc/go-micro/errors"
"github.com/myodc/go-micro/registry"
2015-05-20 22:57:19 +01:00
"github.com/myodc/go-micro/transport"
2015-05-23 11:53:40 +01:00
2015-01-13 23:31:27 +00:00
rpc "github.com/youtube/vitess/go/rpcplus"
js "github.com/youtube/vitess/go/rpcplus/jsonrpc"
pb "github.com/youtube/vitess/go/rpcplus/pbrpc"
2015-05-23 11:53:40 +01:00
"golang.org/x/net/context"
2015-05-16 00:33:43 +01:00
"google.golang.org/grpc"
2015-01-13 23:31:27 +00:00
)
type headerRoundTripper struct {
r http.RoundTripper
}
2015-05-23 17:40:53 +01:00
type rpcClient struct {
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
func init() {
rand.Seed(time.Now().UnixNano())
}
2015-05-23 17:40:53 +01:00
func newRpcClient(opt ...Option) Client {
var opts options
for _, o := range opt {
o(&opts)
}
if opts.transport == nil {
opts.transport = transport.DefaultTransport
}
return &rpcClient{
opts: opts,
}
}
2015-01-13 23:31:27 +00:00
func (t *headerRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
r.Header.Set("X-Client-Version", "1.0")
return t.r.RoundTrip(r)
}
2015-05-23 17:40:53 +01:00
func (r *rpcClient) call(ctx context.Context, address string, request Request, response interface{}) error {
2015-05-16 00:33:43 +01:00
switch request.ContentType() {
case "application/grpc":
cc, err := grpc.Dial(address)
if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error connecting to server: %v", err))
}
2015-05-23 11:53:40 +01:00
if err := grpc.Invoke(ctx, request.Method(), request.Request(), response, cc); err != nil {
2015-05-16 00:33:43 +01:00
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
}
return nil
}
2015-01-13 23:31:27 +00:00
pReq := &rpc.Request{
ServiceMethod: request.Method(),
}
reqB := bytes.NewBuffer(nil)
defer reqB.Reset()
buf := &buffer{
reqB,
}
var cc rpc.ClientCodec
switch request.ContentType() {
case "application/octet-stream":
cc = pb.NewClientCodec(buf)
case "application/json":
cc = js.NewClientCodec(buf)
default:
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Unsupported request type: %s", request.ContentType()))
}
err := cc.WriteRequest(pReq, request.Request())
if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error writing request: %v", err))
}
2015-05-20 22:57:19 +01:00
msg := &transport.Message{
Header: make(map[string]string),
Body: reqB.Bytes(),
}
2015-01-13 23:31:27 +00:00
2015-05-23 11:53:40 +01:00
md, ok := c.GetMetaData(ctx)
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-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-05-20 22:57:19 +01:00
rsp, err := c.Send(msg)
2015-01-13 23:31:27 +00:00
if err != nil {
2015-05-20 22:57:19 +01:00
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
2015-01-13 23:31:27 +00:00
}
2015-05-20 22:57:19 +01:00
rspB := bytes.NewBuffer(rsp.Body)
2015-01-13 23:31:27 +00:00
defer rspB.Reset()
rBuf := &buffer{
rspB,
}
2015-05-20 22:57:19 +01:00
switch rsp.Header["Content-Type"] {
2015-01-13 23:31:27 +00:00
case "application/octet-stream":
cc = pb.NewClientCodec(rBuf)
case "application/json":
cc = js.NewClientCodec(rBuf)
default:
2015-05-20 22:57:19 +01:00
return errors.InternalServerError("go.micro.client", string(rsp.Body))
2015-01-13 23:31:27 +00:00
}
pRsp := &rpc.Response{}
err = cc.ReadResponseHeader(pRsp)
if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error reading response headers: %v", err))
}
if len(pRsp.Error) > 0 {
return errors.Parse(pRsp.Error)
}
err = cc.ReadResponseBody(response)
if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error reading response body: %v", err))
}
return nil
}
2015-05-23 17:40:53 +01:00
func (r *rpcClient) CallRemote(ctx context.Context, address string, request Request, response interface{}) error {
2015-05-23 11:53:40 +01:00
return r.call(ctx, address, request, response)
2015-01-13 23:31:27 +00:00
}
// TODO: Call(..., opts *Options) error {
2015-05-23 17:40:53 +01:00
func (r *rpcClient) Call(ctx context.Context, request Request, response interface{}) error {
2015-01-13 23:31:27 +00:00
service, err := registry.GetService(request.Service())
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
if len(service.Nodes()) == 0 {
return errors.NotFound("go.micro.client", "Service not found")
}
n := rand.Int() % len(service.Nodes())
node := service.Nodes()[n]
2015-05-21 19:24:57 +01:00
address := node.Address()
if node.Port() > 0 {
address = fmt.Sprintf("%s:%d", address, node.Port())
}
2015-05-23 11:53:40 +01:00
return r.call(ctx, address, request, response)
2015-01-13 23:31:27 +00:00
}
2015-05-23 17:40:53 +01:00
func (r *rpcClient) NewRequest(service, method string, request interface{}) Request {
2015-01-13 23:31:27 +00:00
return r.NewProtoRequest(service, method, request)
}
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")
}