2015-01-13 23:31:27 +00:00
|
|
|
package client
|
|
|
|
|
|
|
|
import (
|
|
|
|
"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"
|
2015-05-23 11:53:40 +01:00
|
|
|
|
|
|
|
"golang.org/x/net/context"
|
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-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-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-06-01 18:55:27 +01:00
|
|
|
client := rpc.NewClientWithCodec(newRpcPlusCodec(msg, c))
|
|
|
|
return client.Call(ctx, request.Method(), request.Request(), response)
|
|
|
|
}
|
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-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-06-01 18:55:27 +01:00
|
|
|
client := rpc.NewClientWithCodec(newRpcPlusCodec(msg, c))
|
|
|
|
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-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())
|
|
|
|
}
|
|
|
|
|
2015-05-25 22:14:28 +01:00
|
|
|
if len(service.Nodes) == 0 {
|
2015-01-13 23:31:27 +00:00
|
|
|
return errors.NotFound("go.micro.client", "Service not found")
|
|
|
|
}
|
|
|
|
|
2015-05-25 22:14:28 +01:00
|
|
|
n := rand.Int() % len(service.Nodes)
|
|
|
|
node := service.Nodes[n]
|
2015-05-21 19:24:57 +01: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-05-23 11:53:40 +01:00
|
|
|
return r.call(ctx, address, request, response)
|
2015-01-13 23:31:27 +00:00
|
|
|
}
|
|
|
|
|
2015-06-01 18:55:27 +01:00
|
|
|
func (r *rpcClient) StreamRemote(ctx context.Context, address string, request Request, responseChan interface{}) (Streamer, error) {
|
|
|
|
return r.stream(ctx, address, request, responseChan)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (r *rpcClient) Stream(ctx context.Context, request Request, responseChan interface{}) (Streamer, error) {
|
|
|
|
service, err := registry.GetService(request.Service())
|
|
|
|
if err != nil {
|
|
|
|
return nil, errors.InternalServerError("go.micro.client", err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
if len(service.Nodes) == 0 {
|
|
|
|
return nil, errors.NotFound("go.micro.client", "Service not found")
|
|
|
|
}
|
|
|
|
|
|
|
|
n := rand.Int() % len(service.Nodes)
|
|
|
|
node := service.Nodes[n]
|
|
|
|
|
|
|
|
address := node.Address
|
|
|
|
if node.Port > 0 {
|
|
|
|
address = fmt.Sprintf("%s:%d", address, node.Port)
|
|
|
|
}
|
|
|
|
|
|
|
|
return r.stream(ctx, address, request, responseChan)
|
|
|
|
}
|
|
|
|
|
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")
|
|
|
|
}
|