Dont expose rpc client/server

This commit is contained in:
Asim 2015-05-23 17:40:53 +01:00
parent 3d8dd6e121
commit e192f335da
8 changed files with 73 additions and 83 deletions

View File

@ -20,7 +20,7 @@ type options struct {
type Option func(*options) type Option func(*options)
var ( var (
DefaultClient Client = NewRpcClient() DefaultClient Client = newRpcClient()
) )
func Transport(t transport.Transport) Option { func Transport(t transport.Transport) Option {
@ -37,6 +37,10 @@ func CallRemote(ctx context.Context, address string, request Request, response i
return DefaultClient.CallRemote(ctx, address, request, response) return DefaultClient.CallRemote(ctx, address, request, response)
} }
func New(opt ...Option) Client {
return newRpcClient(opt...)
}
func NewRequest(service, method string, request interface{}) Request { func NewRequest(service, method string, request interface{}) Request {
return DefaultClient.NewRequest(service, method, request) return DefaultClient.NewRequest(service, method, request)
} }

View File

@ -24,7 +24,7 @@ type headerRoundTripper struct {
r http.RoundTripper r http.RoundTripper
} }
type RpcClient struct { type rpcClient struct {
opts options opts options
} }
@ -32,12 +32,28 @@ func init() {
rand.Seed(time.Now().UnixNano()) rand.Seed(time.Now().UnixNano())
} }
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,
}
}
func (t *headerRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { func (t *headerRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
r.Header.Set("X-Client-Version", "1.0") r.Header.Set("X-Client-Version", "1.0")
return t.r.RoundTrip(r) return t.r.RoundTrip(r)
} }
func (r *RpcClient) call(ctx context.Context, address string, request Request, response interface{}) error { func (r *rpcClient) call(ctx context.Context, address string, request Request, response interface{}) error {
switch request.ContentType() { switch request.ContentType() {
case "application/grpc": case "application/grpc":
cc, err := grpc.Dial(address) cc, err := grpc.Dial(address)
@ -132,12 +148,12 @@ func (r *RpcClient) call(ctx context.Context, address string, request Request, r
return nil return nil
} }
func (r *RpcClient) CallRemote(ctx context.Context, address string, request Request, response interface{}) error { func (r *rpcClient) CallRemote(ctx context.Context, address string, request Request, response interface{}) error {
return r.call(ctx, address, request, response) return r.call(ctx, address, request, response)
} }
// TODO: Call(..., opts *Options) error { // TODO: Call(..., opts *Options) error {
func (r *RpcClient) Call(ctx context.Context, request Request, response interface{}) error { func (r *rpcClient) Call(ctx context.Context, request Request, response interface{}) error {
service, err := registry.GetService(request.Service()) service, err := registry.GetService(request.Service())
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", err.Error())
@ -158,30 +174,14 @@ func (r *RpcClient) Call(ctx context.Context, request Request, response interfac
return r.call(ctx, address, request, response) return r.call(ctx, address, request, response)
} }
func (r *RpcClient) NewRequest(service, method string, request interface{}) Request { func (r *rpcClient) NewRequest(service, method string, request interface{}) Request {
return r.NewProtoRequest(service, method, request) return r.NewProtoRequest(service, method, request)
} }
func (r *RpcClient) NewProtoRequest(service, method string, request interface{}) Request { func (r *rpcClient) NewProtoRequest(service, method string, request interface{}) Request {
return newRpcRequest(service, method, request, "application/octet-stream") return newRpcRequest(service, method, request, "application/octet-stream")
} }
func (r *RpcClient) NewJsonRequest(service, method string, request interface{}) Request { func (r *rpcClient) NewJsonRequest(service, method string, request interface{}) Request {
return newRpcRequest(service, method, request, "application/json") return newRpcRequest(service, method, request, "application/json")
} }
func NewRpcClient(opt ...Option) *RpcClient {
var opts options
for _, o := range opt {
o(&opts)
}
if opts.transport == nil {
opts.transport = transport.DefaultTransport
}
return &RpcClient{
opts: opts,
}
}

View File

@ -1,14 +1,14 @@
package client package client
type RpcRequest struct { type rpcRequest struct {
service string service string
method string method string
contentType string contentType string
request interface{} request interface{}
} }
func newRpcRequest(service, method string, request interface{}, contentType string) *RpcRequest { func newRpcRequest(service, method string, request interface{}, contentType string) Request {
return &RpcRequest{ return &rpcRequest{
service: service, service: service,
method: method, method: method,
request: request, request: request,
@ -16,22 +16,18 @@ func newRpcRequest(service, method string, request interface{}, contentType stri
} }
} }
func (r *RpcRequest) ContentType() string { func (r *rpcRequest) ContentType() string {
return r.contentType return r.contentType
} }
func (r *RpcRequest) Service() string { func (r *rpcRequest) Service() string {
return r.service return r.service
} }
func (r *RpcRequest) Method() string { func (r *rpcRequest) Method() string {
return r.method return r.method
} }
func (r *RpcRequest) Request() interface{} { func (r *rpcRequest) Request() interface{} {
return r.request return r.request
} }
func NewRpcRequest(service, method string, request interface{}, contentType string) *RpcRequest {
return newRpcRequest(service, method, request, contentType)
}

View File

@ -114,7 +114,7 @@ func Setup(c *cli.Context) error {
transport.DefaultTransport = transport.NewNatsTransport(tAddrs) transport.DefaultTransport = transport.NewNatsTransport(tAddrs)
} }
client.DefaultClient = client.NewRpcClient() client.DefaultClient = client.New()
return nil return nil
} }

View File

@ -12,7 +12,6 @@ import (
func main() { func main() {
cmd.Init() cmd.Init()
client.DefaultClient = client.NewRpcClient()
// Create new request to service go.micro.service.go-template, method Example.Call // Create new request to service go.micro.service.go-template, method Example.Call
req := client.NewRequest("go.micro.service.template", "Example.Call", &example.Request{ req := client.NewRequest("go.micro.service.template", "Example.Call", &example.Request{
Name: "John", Name: "John",

View File

@ -1,29 +1,21 @@
package server package server
type RpcReceiver struct { type rpcReceiver struct {
name string name string
handler interface{} handler interface{}
} }
func newRpcReceiver(name string, handler interface{}) *RpcReceiver { func newRpcReceiver(name string, handler interface{}) Receiver {
return &RpcReceiver{ return &rpcReceiver{
name: name, name: name,
handler: handler, handler: handler,
} }
} }
func (r *RpcReceiver) Name() string { func (r *rpcReceiver) Name() string {
return r.name return r.name
} }
func (r *RpcReceiver) Handler() interface{} { func (r *rpcReceiver) Handler() interface{} {
return r.handler return r.handler
} }
func NewRpcReceiver(handler interface{}) *RpcReceiver {
return newRpcReceiver("", handler)
}
func NewNamedRpcReceiver(name string, handler interface{}) *RpcReceiver {
return newRpcReceiver(name, handler)
}

View File

@ -15,7 +15,7 @@ import (
"golang.org/x/net/context" "golang.org/x/net/context"
) )
type RpcServer struct { type rpcServer struct {
mtx sync.RWMutex mtx sync.RWMutex
address string address string
opts options opts options
@ -23,12 +23,26 @@ type RpcServer struct {
exit chan chan error exit chan chan error
} }
var ( func newRpcServer(address string, opt ...Option) Server {
HealthPath = "/_status/health" var opts options
RpcPath = "/_rpc"
)
func (s *RpcServer) accept(sock transport.Socket) { for _, o := range opt {
o(&opts)
}
if opts.transport == nil {
opts.transport = transport.DefaultTransport
}
return &rpcServer{
opts: opts,
address: address,
rpc: rpc.NewServer(),
exit: make(chan chan error),
}
}
func (s *rpcServer) accept(sock transport.Socket) {
var msg transport.Message var msg transport.Message
if err := sock.Recv(&msg); err != nil { if err := sock.Recv(&msg); err != nil {
return return
@ -72,26 +86,26 @@ func (s *RpcServer) accept(sock transport.Socket) {
}) })
} }
func (s *RpcServer) Address() string { func (s *rpcServer) Address() string {
s.mtx.RLock() s.mtx.RLock()
address := s.address address := s.address
s.mtx.RUnlock() s.mtx.RUnlock()
return address return address
} }
func (s *RpcServer) Init() error { func (s *rpcServer) Init() error {
return nil return nil
} }
func (s *RpcServer) NewReceiver(handler interface{}) Receiver { func (s *rpcServer) NewReceiver(handler interface{}) Receiver {
return newRpcReceiver("", handler) return newRpcReceiver("", handler)
} }
func (s *RpcServer) NewNamedReceiver(name string, handler interface{}) Receiver { func (s *rpcServer) NewNamedReceiver(name string, handler interface{}) Receiver {
return newRpcReceiver(name, handler) return newRpcReceiver(name, handler)
} }
func (s *RpcServer) Register(r Receiver) error { func (s *rpcServer) Register(r Receiver) error {
if len(r.Name()) > 0 { if len(r.Name()) > 0 {
s.rpc.RegisterName(r.Name(), r.Handler()) s.rpc.RegisterName(r.Name(), r.Handler())
return nil return nil
@ -101,7 +115,7 @@ func (s *RpcServer) Register(r Receiver) error {
return nil return nil
} }
func (s *RpcServer) Start() error { func (s *rpcServer) Start() error {
registerHealthChecker(s) registerHealthChecker(s)
ts, err := s.opts.transport.Listen(s.address) ts, err := s.opts.transport.Listen(s.address)
@ -125,27 +139,8 @@ func (s *RpcServer) Start() error {
return nil return nil
} }
func (s *RpcServer) Stop() error { func (s *rpcServer) Stop() error {
ch := make(chan error) ch := make(chan error)
s.exit <- ch s.exit <- ch
return <-ch return <-ch
} }
func NewRpcServer(address string, opt ...Option) *RpcServer {
var opts options
for _, o := range opt {
o(&opts)
}
if opts.transport == nil {
opts.transport = transport.DefaultTransport
}
return &RpcServer{
opts: opts,
address: address,
rpc: rpc.NewServer(),
exit: make(chan chan error),
}
}

View File

@ -54,12 +54,16 @@ func Init() error {
} }
if DefaultServer == nil { if DefaultServer == nil {
DefaultServer = NewRpcServer(Address) DefaultServer = newRpcServer(Address)
} }
return DefaultServer.Init() return DefaultServer.Init()
} }
func New(address string, opt ...Option) Server {
return newRpcServer(address, opt...)
}
func NewReceiver(handler interface{}) Receiver { func NewReceiver(handler interface{}) Receiver {
return DefaultServer.NewReceiver(handler) return DefaultServer.NewReceiver(handler)
} }