unistack-org/micro
4
0
Fork 3
Code Issues 18 Pull Requests 1 Actions Packages Projects Releases 263 Wiki Activity

Add call and publish options

Browse Source
This commit is contained in:
Asim 2015-12-08 19:25:42 +00:00
parent 3d279ffcfd
commit 02985c35d5
4 changed files with 26 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
client/client.go
View File

@ -30,11 +30,11 @@ type Client interface {
NewRequest(service, method string, req interface{}) Request NewRequest(service, method string, req interface{}) Request
NewProtoRequest(service, method string, req interface{}) Request NewProtoRequest(service, method string, req interface{}) Request
NewJsonRequest(service, method string, req interface{}) Request NewJsonRequest(service, method string, req interface{}) Request
Call(ctx context.Context, req Request, rsp interface{}) error Call(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error
CallRemote(ctx context.Context, addr string, req Request, rsp interface{}) error CallRemote(ctx context.Context, addr string, req Request, rsp interface{}, opts ...CallOption) error
Stream(ctx context.Context, req Request, rspChan interface{}) (Streamer, error) Stream(ctx context.Context, req Request, rspChan interface{}, opts ...CallOption) (Streamer, error)
StreamRemote(ctx context.Context, addr string, req Request, rspChan interface{}) (Streamer, error) StreamRemote(ctx context.Context, addr string, req Request, rspChan interface{}, opts ...CallOption) (Streamer, error)
Publish(ctx context.Context, p Publication) error Publish(ctx context.Context, p Publication, opts ...PublishOption) error
} }
type Publication interface { type Publication interface {
@ -57,30 +57,32 @@ type Streamer interface {
} }
type Option func(*options) type Option func(*options)
type CallOption func(*callOptions)
type PublishOption func(*publishOptions)
var ( var (
DefaultClient Client = newRpcClient() DefaultClient Client = newRpcClient()
) )
// Makes a synchronous call to a service using the default client // Makes a synchronous call to a service using the default client
func Call(ctx context.Context, request Request, response interface{}) error { func Call(ctx context.Context, request Request, response interface{}, opts ...CallOption) error {
return DefaultClient.Call(ctx, request, response) return DefaultClient.Call(ctx, request, response, opts...)
} }
// Makes a synchronous call to the specified address using the default client // Makes a synchronous call to the specified address using the default client
func CallRemote(ctx context.Context, address string, request Request, response interface{}) error { func CallRemote(ctx context.Context, address string, request Request, response interface{}, opts ...CallOption) error {
return DefaultClient.CallRemote(ctx, address, request, response) return DefaultClient.CallRemote(ctx, address, request, response, opts...)
} }
// Creates a streaming connection with a service and returns responses on the // Creates a streaming connection with a service and returns responses on the
// channel passed in. It's upto the user to close the streamer. // channel passed in. It's upto the user to close the streamer.
func Stream(ctx context.Context, request Request, responseChan interface{}) (Streamer, error) { func Stream(ctx context.Context, request Request, responseChan interface{}, opts ...CallOption) (Streamer, error) {
return DefaultClient.Stream(ctx, request, responseChan) return DefaultClient.Stream(ctx, request, responseChan, opts...)
} }
// Creates a streaming connection to the address specified. // Creates a streaming connection to the address specified.
func StreamRemote(ctx context.Context, address string, request Request, responseChan interface{}) (Streamer, error) { func StreamRemote(ctx context.Context, address string, request Request, responseChan interface{}, opts ...CallOption) (Streamer, error) {
return DefaultClient.StreamRemote(ctx, address, request, responseChan) return DefaultClient.StreamRemote(ctx, address, request, responseChan, opts...)
} }
// Publishes a publication using the default client. Using the underlying broker // Publishes a publication using the default client. Using the underlying broker

4
client/options.go
View File

@ -17,6 +17,10 @@ type options struct {
selector Selector selector Selector
} }
type callOptions struct{}
type publishOptions struct{}
// Broker to be used for pub/sub // Broker to be used for pub/sub
func Broker(b broker.Broker) Option { func Broker(b broker.Broker) Option {
return func(o *options) { return func(o *options) {

11
client/rpc_client.go
View File

@ -148,12 +148,11 @@ func (r *rpcClient) stream(ctx context.Context, address string, request Request,
}, nil }, 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{}, opts ...CallOption) error {
return r.call(ctx, address, request, response) return r.call(ctx, address, request, response)
} }
// TODO: Call(..., opts *Options) error { func (r *rpcClient) Call(ctx context.Context, request Request, response interface{}, opts ...CallOption) error {
func (r *rpcClient) Call(ctx context.Context, request Request, response interface{}) error {
node, err := r.sel.Select(ctx, request) node, err := r.sel.Select(ctx, request)
if err != nil { if err != nil {
return err return err
@ -169,11 +168,11 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
return err return err
} }
func (r *rpcClient) StreamRemote(ctx context.Context, address string, request Request, responseChan interface{}) (Streamer, error) { func (r *rpcClient) StreamRemote(ctx context.Context, address string, request Request, responseChan interface{}, opts ...CallOption) (Streamer, error) {
return r.stream(ctx, address, request, responseChan) return r.stream(ctx, address, request, responseChan)
} }
func (r *rpcClient) Stream(ctx context.Context, request Request, responseChan interface{}) (Streamer, error) { func (r *rpcClient) Stream(ctx context.Context, request Request, responseChan interface{}, opts ...CallOption) (Streamer, error) {
node, err := r.sel.Select(ctx, request) node, err := r.sel.Select(ctx, request)
if err != nil { if err != nil {
return nil, err return nil, err
@ -189,7 +188,7 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, responseChan in
return stream, err return stream, err
} }
func (r *rpcClient) Publish(ctx context.Context, p Publication) error { func (r *rpcClient) Publish(ctx context.Context, p Publication, opts ...PublishOption) error {
md, ok := c.GetMetadata(ctx) md, ok := c.GetMetadata(ctx)
if !ok { if !ok {
md = make(map[string]string) md = make(map[string]string)

4
examples/client/wrapper/wrapper.go
View File

@ -18,7 +18,7 @@ type logWrapper struct {
client.Client client.Client
} }
func (l *logWrapper) Call(ctx context.Context, req client.Request, rsp interface{}) error { func (l *logWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
md, _ := c.GetMetadata(ctx) md, _ := c.GetMetadata(ctx)
fmt.Printf("[Log Wrapper] ctx: %v service: %s method: %s\n", md, req.Service(), req.Method()) fmt.Printf("[Log Wrapper] ctx: %v service: %s method: %s\n", md, req.Service(), req.Method())
return l.Client.Call(ctx, req, rsp) return l.Client.Call(ctx, req, rsp)
@ -29,7 +29,7 @@ type traceWrapper struct {
client.Client client.Client
} }
func (t *traceWrapper) Call(ctx context.Context, req client.Request, rsp interface{}) error { func (t *traceWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
ctx = c.WithMetadata(ctx, map[string]string{ ctx = c.WithMetadata(ctx, map[string]string{
"X-Trace-Id": fmt.Sprintf("%d", time.Now().Unix()), "X-Trace-Id": fmt.Sprintf("%d", time.Now().Unix()),
}) })