origin/rpc/client.go

package rpc

import (
	"errors"
	"fmt"
	"github.com/duanhf2012/origin/v2/log"
	"github.com/duanhf2012/origin/v2/network"
	"reflect"
	"time"
)

const (
	DefaultRpcConnNum           = 1
	DefaultRpcLenMsgLen         = 4
	DefaultRpcMinMsgLen         = 2
	DefaultMaxCheckCallRpcCount = 1000
	DefaultMaxPendingWriteNum   = 1000000

	DefaultConnectInterval             = 2 * time.Second
	DefaultCheckRpcCallTimeoutInterval = 1 * time.Second
	DefaultRpcTimeout                  = 15 * time.Second
)

var clientSeq uint32

type IWriter interface {
	WriteMsg(nodeId string, args ...[]byte) error
	IsConnected() bool
}

type IRealClient interface {
	SetConn(conn *network.NetConn)
	Close(waitDone bool)

	AsyncCall(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}) (CancelRpc, error)
	Go(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, noReply bool, serviceMethod string, args interface{}, reply interface{}) *Call
	RawGo(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, processor IRpcProcessor, noReply bool, rpcMethodId uint32, serviceMethod string, rawArgs []byte, reply interface{}) *Call
	IsConnected() bool

	Run()
	OnClose()

	Bind(server IServer)
}

type Client struct {
	clientId         uint32
	targetNodeId     string
	compressBytesLen int

	*CallSet
	IRealClient
}

func (client *Client) NewClientAgent(conn *network.NetConn) network.Agent {
	client.SetConn(conn)

	return client
}

func (client *Client) GetTargetNodeId() string {
	return client.targetNodeId
}

func (client *Client) GetClientId() uint32 {
	return client.clientId
}

func (client *Client) processRpcResponse(responseData []byte) error {
	bCompress := (responseData[0] >> 7) > 0
	processor := GetProcessor(responseData[0] & 0x7f)
	if processor == nil {
		//rc.conn.ReleaseReadMsg(responseData)
		err := errors.New(fmt.Sprintf("cannot find process %d", responseData[0]&0x7f))
		log.Error(err.Error())
		return err
	}

	//1.解析head
	response := RpcResponse{}
	response.RpcResponseData = processor.MakeRpcResponse(0, "", nil)

	//解压缩
	byteData := responseData[1:]
	var compressBuff []byte

	if bCompress == true {
		var unCompressErr error
		compressBuff, unCompressErr = compressor.UncompressBlock(byteData)
		if unCompressErr != nil {
			//rc.conn.ReleaseReadMsg(responseData)
			err := fmt.Errorf("uncompressBlock failed,err :%s", unCompressErr.Error())
			return err
		}

		byteData = compressBuff
	}

	err := processor.Unmarshal(byteData, response.RpcResponseData)
	if cap(compressBuff) > 0 {
		compressor.UnCompressBufferCollection(compressBuff)
	}

	//rc.conn.ReleaseReadMsg(bytes)
	if err != nil {
		processor.ReleaseRpcResponse(response.RpcResponseData)
		log.Error("rpcClient Unmarshal head error", log.ErrorField("error", err))
		return nil
	}

	v := client.RemovePending(response.RpcResponseData.GetSeq())
	if v == nil {
		log.Error("rpcClient cannot find seq", log.Uint64("seq", response.RpcResponseData.GetSeq()))
	} else {
		v.Err = nil
		if len(response.RpcResponseData.GetReply()) > 0 {
			err = processor.Unmarshal(response.RpcResponseData.GetReply(), v.Reply)
			if err != nil {
				log.Error("rpcClient Unmarshal body failed", log.ErrorField("error", err))
				v.Err = err
			}
		}

		if response.RpcResponseData.GetErr() != nil {
			v.Err = response.RpcResponseData.GetErr()
		}

		if v.callback != nil && v.callback.IsValid() {
			v.rpcHandler.PushRpcResponse(v)
		} else {
			v.done <- v
		}
	}

	processor.ReleaseRpcResponse(response.RpcResponseData)

	return nil
}

//func (rc *Client) Go(timeout time.Duration,rpcHandler IRpcHandler,noReply bool, serviceMethod string, args interface{}, reply interface{}) *Call {
//	_, processor := GetProcessorType(args)
//	InParam, err := processor.Marshal(args)
//	if err != nil {
//		log.Error("Marshal is fail",log.ErrorAttr("error",err))
//		call := MakeCall()
//		call.DoError(err)
//		return call
//	}
//
//	return rc.RawGo(timeout,rpcHandler,processor, noReply, 0, serviceMethod, InParam, reply)
//}

func (client *Client) rawGo(nodeId string, w IWriter, timeout time.Duration, rpcHandler IRpcHandler, processor IRpcProcessor, noReply bool, rpcMethodId uint32, serviceMethod string, rawArgs []byte, reply interface{}) *Call {
	call := MakeCall()
	call.ServiceMethod = serviceMethod
	call.Reply = reply
	call.Seq = client.generateSeq()
	call.TimeOut = timeout

	request := MakeRpcRequest(processor, call.Seq, rpcMethodId, serviceMethod, noReply, rawArgs)
	bytes, err := processor.Marshal(request.RpcRequestData)
	ReleaseRpcRequest(request)

	if err != nil {
		call.Seq = 0
		log.Error("marshal is fail", log.String("error", err.Error()))
		call.DoError(err)
		return call
	}

	if w == nil || w.IsConnected() == false {
		call.Seq = 0
		sErr := errors.New(serviceMethod + "  was called failed,rpc client is disconnect")
		log.Error("conn is disconnect", log.String("error", sErr.Error()))
		call.DoError(sErr)
		return call
	}

	var compressBuff []byte
	bCompress := uint8(0)
	if client.compressBytesLen > 0 && len(bytes) >= client.compressBytesLen {
		var cErr error
		compressBuff, cErr = compressor.CompressBlock(bytes)
		if cErr != nil {
			call.Seq = 0
			log.Error("compress fail", log.String("error", cErr.Error()))
			call.DoError(cErr)
			return call
		}
		if len(compressBuff) < len(bytes) {
			bytes = compressBuff
			bCompress = 1 << 7
		}
	}

	if noReply == false {
		client.AddPending(call)
	}

	err = w.WriteMsg(nodeId, []byte{uint8(processor.GetProcessorType()) | bCompress}, bytes)
	if cap(compressBuff) > 0 {
		compressor.CompressBufferCollection(compressBuff)
	}
	if err != nil {
		client.RemovePending(call.Seq)
		log.Error("WriteMsg is fail", log.ErrorField("error", err))
		call.Seq = 0
		call.DoError(err)
	}

	return call
}

func (client *Client) asyncCall(nodeId string, w IWriter, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}) (CancelRpc, error) {
	processorType, processor := GetProcessorType(args)
	InParam, herr := processor.Marshal(args)
	if herr != nil {
		return emptyCancelRpc, herr
	}

	seq := client.generateSeq()
	request := MakeRpcRequest(processor, seq, 0, serviceMethod, false, InParam)
	bytes, err := processor.Marshal(request.RpcRequestData)
	ReleaseRpcRequest(request)
	if err != nil {
		return emptyCancelRpc, err
	}

	if w == nil || w.IsConnected() == false {
		return emptyCancelRpc, errors.New("Rpc server is disconnect,call " + serviceMethod)
	}

	var compressBuff []byte
	bCompress := uint8(0)
	if client.compressBytesLen > 0 && len(bytes) >= client.compressBytesLen {
		var cErr error
		compressBuff, cErr = compressor.CompressBlock(bytes)
		if cErr != nil {
			return emptyCancelRpc, cErr
		}

		if len(compressBuff) < len(bytes) {
			bytes = compressBuff
			bCompress = 1 << 7
		}
	}

	call := MakeCall()
	call.Reply = replyParam
	call.callback = &callback
	call.rpcHandler = rpcHandler
	call.ServiceMethod = serviceMethod
	call.Seq = seq
	call.TimeOut = timeout
	client.AddPending(call)

	err = w.WriteMsg(nodeId, []byte{uint8(processorType) | bCompress}, bytes)
	if cap(compressBuff) > 0 {
		compressor.CompressBufferCollection(compressBuff)
	}
	if err != nil {
		client.RemovePending(call.Seq)
		ReleaseCall(call)
		return emptyCancelRpc, err
	}


	rpcCancel := RpcCancel{CallSeq: seq, Cli: client}
	return rpcCancel.CancelRpc, nil
}