U: add CallOptions & AsyncCall

rpc/client.go

@@ -24,6 +24,39 @@ func init() {
 	rand.Seed(time.Now().UnixNano())
 }
 
+type CallOptions struct {
+	// request timeout
+	RequestTimeout time.Duration
+	// response timeout
+	ResponseTimeout time.Duration
+	Meta map[interface{}]interface{}
+}
+
+type CallOption func(*CallOptions)
+
+func CallRequestTimeout(d time.Duration) CallOption {
+	return func(o *CallOptions) {
+		o.RequestTimeout = d
+	}
+}
+
+func CallResponseTimeout(d time.Duration) CallOption {
+	return func(o *CallOptions) {
+		o.ResponseTimeout = d
+	}
+}
+
+func CallMeta(k, v interface{}) CallOption {
+	return func(o *CallOptions) {
+		if o.Meta == nil {
+			o.Meta = make(map[interface{}]interface{})
+		}
+		o.Meta[k] = v
+	}
+}
+
+type AsyncHandler func(reply interface{}, opts CallOptions)
+
 type Client struct {
 	conf     ClientConfig
 	pool     *gettyRPCClientPool
@@ -47,7 +80,52 @@ func NewClient(conf *ClientConfig) (*Client, error) {
 	return c, nil
 }
 
-func (c *Client) Call(typ CodecType, addr, service, method string, args interface{}, reply interface{}) error {
+func (c *Client) Notify(typ CodecType, addr, service, method string, args interface{}, opts ...CallOption) error {
+	var copts CallOptions
+	
+	for _, o := range opts {
+		o(&copts)
+	}
+	
+	return jerrors.Trace(c.call(CT_OneWay, typ, addr, service, method, args, nil, nil, copts))
+}
+
+// if @reply is nil, the transport layer will get the response without notify the invoker.
+func (c *Client) Call(typ CodecType, addr, service, method string, args, reply interface{}, opts ...CallOption) error {
+  var copts CallOptions
+	
+	for _, o := range opts {
+		o(&copts)
+	}
+	
+	ct := CT_TwoWay
+	if reply == nil {
+		ct = CT_TwoWayNoReply
+	}
+	
+	return jerrors.Trace(c.call(ct, typ, addr, service, method, args, reply, nil, copts))
+}
+
+func (c *Client) AsyncCall(typ CodecType, addr, service, method string,
+	args, reply interface{}, handler AsyncHandler, opts ...CallOption) error {
+	var copts CallOptions
+	
+	for _, o := range opts {
+		o(&copts)
+	}
+	
+	return jerrors.Trace(c.call(CT_TwoWay, typ, addr, service, method, args, reply, handler, copts))
+}
+
+func (c *Client) call(ct CallType, typ CodecType, addr, service, method string,
+	args, reply interface{}, handler AsyncHandler, opts CallOptions) error {
+	
+	if opts.RequestTimeout == 0  {
+		opts.RequestTimeout = c.conf.GettySessionParam.tcpWriteTimeout
+	}
+	if opts.ResponseTimeout == 0 {
+	opts.ResponseTimeout = c.conf.GettySessionParam.tcpReadTimeout
+	}
 	if !typ.CheckValidity() {
 		return errInvalidCodecType
 	}
@@ -55,14 +133,16 @@ func (c *Client) Call(typ CodecType, addr, service, method string, args interfac
 	b := &GettyRPCRequest{}
 	b.header.Service = service
 	b.header.Method = method
-	b.header.CallType = CT_TwoWay
-	if reply == nil {
-		b.header.CallType = CT_TwoWayNoReply
-	}
+	b.header.CallType = ct
 	b.body = args
-
-	rsp := NewPendingResponse()
-	rsp.reply = reply
+	
+	var rsp *PendingResponse
+	if ct != CT_OneWay {
+		rsp = NewPendingResponse()
+		rsp.reply = reply
+		rsp.handler = handler
+		rsp.opts = opts
+	}
 
 	var (
 		err     error
@@ -75,12 +155,16 @@ func (c *Client) Call(typ CodecType, addr, service, method string, args interfac
 	}
 	defer c.pool.release(conn, err)
 
-	if err = c.transfer(session, typ, b, rsp); err != nil {
+	if err = c.transfer(session, typ, b, rsp, opts); err != nil {
 		return jerrors.Trace(err)
 	}
+	
+	if ct == CT_OneWay || handler != nil {
+		return  nil
+	}
 
 	select {
-	case <-getty.GetTimeWheel().After(c.conf.GettySessionParam.tcpReadTimeout):
+	case <-getty.GetTimeWheel().After(opts.ResponseTimeout):
 		err = errClientReadTimeout
 		c.removePendingResponse(SequenceType(rsp.seq))
 	case <-rsp.done:
@@ -106,11 +190,12 @@ func (c *Client) selectSession(typ CodecType, addr string) (*gettyRPCClient, get
 }
 
 func (c *Client) heartbeat(session getty.Session, typ CodecType) error {
-	rsp := NewPendingResponse()
-	return c.transfer(session, typ, nil, rsp)
+	return c.transfer(session, typ, nil, NewPendingResponse(), CallOptions{})
 }
 
-func (c *Client) transfer(session getty.Session, typ CodecType, req *GettyRPCRequest, rsp *PendingResponse) error {
+func (c *Client) transfer(session getty.Session, typ CodecType,req *GettyRPCRequest,
+	rsp *PendingResponse, opts CallOptions) error {
+	
 	var (
 		sequence uint64
 		err      error
@@ -127,11 +212,13 @@ func (c *Client) transfer(session getty.Session, typ CodecType, req *GettyRPCReq
 		pkg.H.Command = gettyCmdRPCRequest
 		pkg.B = req
 	}
+	
+	if rsp != nil {
+		rsp.seq = sequence
+		c.addPendingResponse(rsp)
+	}
 
-	rsp.seq = sequence
-	c.addPendingResponse(rsp)
-
-	err = session.WritePkg(pkg, 0)
+	err = session.WritePkg(pkg, opts.RequestTimeout)
 	if err != nil {
 		c.removePendingResponse(SequenceType(rsp.seq))
 	}

rpc/codec.go

@@ -507,10 +507,12 @@ func (resp *GettyRPCResponse) GetHeader() interface{} {
 ////////////////////////////////////////////
 
 type PendingResponse struct {
-	seq   uint64
-	err   error
-	reply interface{}
-	done  chan struct{}
+	seq     uint64
+	err     error
+	handler AsyncHandler
+	reply   interface{}
+	opts CallOptions
+	done    chan struct{}
 }
 
 func NewPendingResponse() *PendingResponse {

rpc/listener.go

 package rpc
 
 import (
-	"reflect"
+		"reflect"
 	"sync"
 	"time"
 )
@@ -90,13 +90,21 @@ func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) {
 		h.replyCmd(session, req, gettyCmdHbResponse, "")
 		return
 	}
+	if req.header.CallType == CT_OneWay {
+		function := req.methodType.method.Func
+		function.Call([]reflect.Value{req.service.rcvr, req.argv, req.replyv})
+		return
+	}
 	if req.header.CallType == CT_TwoWayNoReply {
 		h.replyCmd(session, req, gettyCmdRPCResponse, "")
 		function := req.methodType.method.Func
 		function.Call([]reflect.Value{req.service.rcvr, req.argv, req.replyv})
 		return
 	}
-	h.callService(session, req, req.service, req.methodType, req.argv, req.replyv)
+	err := h.callService(session, req, req.service, req.methodType, req.argv, req.replyv)
+	if err != nil {
+     log.Error("h.callService(session:%#v, req:%#v) = %s", session, req, jerrors.ErrorStack(err))
+	}
 }
 
 func (h *RpcServerHandler) OnCron(session getty.Session) {
@@ -142,14 +150,14 @@ func (h *RpcServerHandler) replyCmd(session getty.Session, req GettyRPCRequestPa
 }
 
 func (h *RpcServerHandler) callService(session getty.Session, req GettyRPCRequestPackage,
-	service *service, methodType *methodType, argv, replyv reflect.Value) {
+	service *service, methodType *methodType, argv, replyv reflect.Value) error {
 
 	function := methodType.method.Func
 	returnValues := function.Call([]reflect.Value{service.rcvr, argv, replyv})
 	errInter := returnValues[0].Interface()
 	if errInter != nil {
 		h.replyCmd(session, req, gettyCmdRPCResponse, errInter.(error).Error())
-		return
+		return nil
 	}
 
 	resp := GettyPackage{
@@ -161,7 +169,7 @@ func (h *RpcServerHandler) callService(session getty.Session, req GettyRPCReques
 		body: replyv.Interface(),
 	}
 
-	session.WritePkg(resp, 5*time.Second)
+	return jerrors.Trace(session.WritePkg(resp, 5*time.Second))
 }
 
 ////////////////////////////////////////////
@@ -220,8 +228,12 @@ func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) {
 	}
 	err := codec.Decode(p.body, pendingResponse.reply)
 	if err != nil {
-		pendingResponse.err = err
-		pendingResponse.done <- struct{}{}
+		if pendingResponse.handler == nil {
+			pendingResponse.err = err
+			pendingResponse.done <- struct{}{}
+		} else {
+			pendingResponse.handler(pendingResponse.reply, pendingResponse.opts)
+		}
 		return
 	}
 	pendingResponse.done <- struct{}{}