Ma/origin
1
0
Fork 0
mirror of https://github.com/duanhf2012/origin.git synced 2026-02-04 06:54:45 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

优化日志格式

Browse Source
This commit is contained in:
orgin
2022-03-10 17:56:11 +08:00
parent 484e505ea4
commit e71e92ecdc
1 changed files with 187 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

372
rpc/rpchandler.go
View File

@@ -13,17 +13,21 @@ import (
) )
const maxClusterNode int = 128 const maxClusterNode int = 128
type FuncRpcClient func(nodeId int,serviceMethod string,client []*Client) (error,int)
type FuncRpcClient func(nodeId int, serviceMethod string, client []*Client) (error, int)
type FuncRpcServer func() *Server type FuncRpcServer func() *Server
var nilError = reflect.Zero(reflect.TypeOf((*error)(nil)).Elem()) var nilError = reflect.Zero(reflect.TypeOf((*error)(nil)).Elem())
type RpcError string type RpcError string
var NilError RpcError var NilError RpcError
func (e RpcError) Error() string { func (e RpcError) Error() string {
return string(e) return string(e)
} }
func ConvertError(e error) RpcError{ func ConvertError(e error) RpcError {
if e == nil { if e == nil {
return NilError return NilError
} }
@@ -33,16 +37,16 @@ func ConvertError(e error) RpcError{
} }
type RpcMethodInfo struct { type RpcMethodInfo struct {
method reflect.Method method reflect.Method
inParamValue reflect.Value inParamValue reflect.Value
inParam interface{} inParam interface{}
outParamValue reflect.Value outParamValue reflect.Value
hasResponder bool hasResponder bool
rpcProcessorType RpcProcessorType rpcProcessorType RpcProcessorType
} }
type RawRpcCallBack interface { type RawRpcCallBack interface {
Unmarshal(data []byte) (interface{},error) Unmarshal(data []byte) (interface{}, error)
CB(data interface{}) CB(data interface{})
} }
@@ -54,16 +58,16 @@ type IRpcHandlerChannel interface {
type RpcHandler struct { type RpcHandler struct {
IRpcHandlerChannel IRpcHandlerChannel
rpcHandler IRpcHandler rpcHandler IRpcHandler
mapFunctions map[string]RpcMethodInfo mapFunctions map[string]RpcMethodInfo
mapRawFunctions map[uint32] RawRpcCallBack mapRawFunctions map[uint32]RawRpcCallBack
funcRpcClient FuncRpcClient funcRpcClient FuncRpcClient
funcRpcServer FuncRpcServer funcRpcServer FuncRpcServer
pClientList []*Client pClientList []*Client
} }
type TriggerRpcEvent func(bConnect bool,clientSeq uint32,nodeId int) type TriggerRpcEvent func(bConnect bool, clientSeq uint32, nodeId int)
type IRpcListener interface { type IRpcListener interface {
OnNodeConnected(nodeId int) OnNodeConnected(nodeId int)
OnNodeDisconnect(nodeId int) OnNodeDisconnect(nodeId int)
@@ -72,44 +76,44 @@ type IRpcListener interface {
type IRpcHandler interface { type IRpcHandler interface {
IRpcHandlerChannel IRpcHandlerChannel
GetName() string GetName() string
InitRpcHandler(rpcHandler IRpcHandler,getClientFun FuncRpcClient,getServerFun FuncRpcServer,rpcHandlerChannel IRpcHandlerChannel) InitRpcHandler(rpcHandler IRpcHandler, getClientFun FuncRpcClient, getServerFun FuncRpcServer, rpcHandlerChannel IRpcHandlerChannel)
GetRpcHandler() IRpcHandler GetRpcHandler() IRpcHandler
HandlerRpcRequest(request *RpcRequest) HandlerRpcRequest(request *RpcRequest)
HandlerRpcResponseCB(call *Call) HandlerRpcResponseCB(call *Call)
CallMethod(ServiceMethod string,param interface{},reply interface{}) error CallMethod(ServiceMethod string, param interface{}, reply interface{}) error
AsyncCall(serviceMethod string,args interface{},callback interface{}) error AsyncCall(serviceMethod string, args interface{}, callback interface{}) error
Call(serviceMethod string,args interface{},reply interface{}) error Call(serviceMethod string, args interface{}, reply interface{}) error
Go(serviceMethod string,args interface{}) error Go(serviceMethod string, args interface{}) error
AsyncCallNode(nodeId int,serviceMethod string,args interface{},callback interface{}) error AsyncCallNode(nodeId int, serviceMethod string, args interface{}, callback interface{}) error
CallNode(nodeId int,serviceMethod string,args interface{},reply interface{}) error CallNode(nodeId int, serviceMethod string, args interface{}, reply interface{}) error
GoNode(nodeId int,serviceMethod string,args interface{}) error GoNode(nodeId int, serviceMethod string, args interface{}) error
RawGoNode(rpcProcessorType RpcProcessorType,nodeId int,rpcMethodId uint32,serviceName string,rawArgs IRawInputArgs) error RawGoNode(rpcProcessorType RpcProcessorType, nodeId int, rpcMethodId uint32, serviceName string, rawArgs IRawInputArgs) error
CastGo(serviceMethod string,args interface{})error CastGo(serviceMethod string, args interface{}) error
IsSingleCoroutine() bool IsSingleCoroutine() bool
UnmarshalInParam(rpcProcessor IRpcProcessor,serviceMethod string,rawRpcMethodId uint32,inParam []byte) (interface{},error) UnmarshalInParam(rpcProcessor IRpcProcessor, serviceMethod string, rawRpcMethodId uint32, inParam []byte) (interface{}, error)
} }
func reqHandlerNull(Returns interface{}, Err RpcError) {
func reqHandlerNull(Returns interface{},Err RpcError) {
} }
var requestHandlerNull reflect.Value var requestHandlerNull reflect.Value
func init(){
func init() {
requestHandlerNull = reflect.ValueOf(reqHandlerNull) requestHandlerNull = reflect.ValueOf(reqHandlerNull)
} }
func (handler *RpcHandler) GetRpcHandler() IRpcHandler{ func (handler *RpcHandler) GetRpcHandler() IRpcHandler {
return handler.rpcHandler return handler.rpcHandler
} }
func (handler *RpcHandler) InitRpcHandler(rpcHandler IRpcHandler,getClientFun FuncRpcClient,getServerFun FuncRpcServer,rpcHandlerChannel IRpcHandlerChannel) { func (handler *RpcHandler) InitRpcHandler(rpcHandler IRpcHandler, getClientFun FuncRpcClient, getServerFun FuncRpcServer, rpcHandlerChannel IRpcHandlerChannel) {
handler.IRpcHandlerChannel = rpcHandlerChannel handler.IRpcHandlerChannel = rpcHandlerChannel
handler.mapRawFunctions = make(map[uint32] RawRpcCallBack) handler.mapRawFunctions = make(map[uint32]RawRpcCallBack)
handler.rpcHandler = rpcHandler handler.rpcHandler = rpcHandler
handler.mapFunctions = map[string]RpcMethodInfo{} handler.mapFunctions = map[string]RpcMethodInfo{}
handler.funcRpcClient = getClientFun handler.funcRpcClient = getClientFun
handler.funcRpcServer = getServerFun handler.funcRpcServer = getServerFun
handler.pClientList = make([]*Client,maxClusterNode) handler.pClientList = make([]*Client, maxClusterNode)
handler.RegisterRpc(rpcHandler) handler.RegisterRpc(rpcHandler)
} }
@@ -131,7 +135,7 @@ func (handler *RpcHandler) isExportedOrBuiltinType(t reflect.Type) bool {
func (handler *RpcHandler) suitableMethods(method reflect.Method) error { func (handler *RpcHandler) suitableMethods(method reflect.Method) error {
//只有RPC_开头的才能被调用 //只有RPC_开头的才能被调用
if strings.Index(method.Name,"RPC_")!=0 { if strings.Index(method.Name, "RPC_") != 0 {
return nil return nil
} }
@@ -139,15 +143,15 @@ func (handler *RpcHandler) suitableMethods(method reflect.Method) error {
var rpcMethodInfo RpcMethodInfo var rpcMethodInfo RpcMethodInfo
typ := method.Type typ := method.Type
if typ.NumOut() != 1 { if typ.NumOut() != 1 {
return fmt.Errorf("%s The number of returned arguments must be 1",method.Name) return fmt.Errorf("%s The number of returned arguments must be 1", method.Name)
} }
if typ.Out(0).String() != "error" { if typ.Out(0).String() != "error" {
return fmt.Errorf("%s The return parameter must be of type error",method.Name) return fmt.Errorf("%s The return parameter must be of type error", method.Name)
} }
if typ.NumIn() <2 || typ.NumIn() > 4 { if typ.NumIn() < 2 || typ.NumIn() > 4 {
return fmt.Errorf("%s Unsupported parameter format",method.Name) return fmt.Errorf("%s Unsupported parameter format", method.Name)
} }
//1.判断第一个参数 //1.判断第一个参数
@@ -157,19 +161,19 @@ func (handler *RpcHandler) suitableMethods(method reflect.Method) error {
rpcMethodInfo.hasResponder = true rpcMethodInfo.hasResponder = true
} }
for i:= parIdx ;i<typ.NumIn();i++{ for i := parIdx; i < typ.NumIn(); i++ {
if handler.isExportedOrBuiltinType(typ.In(i)) == false { if handler.isExportedOrBuiltinType(typ.In(i)) == false {
return fmt.Errorf("%s Unsupported parameter types",method.Name) return fmt.Errorf("%s Unsupported parameter types", method.Name)
} }
} }
rpcMethodInfo.inParamValue = reflect.New(typ.In(parIdx).Elem()) rpcMethodInfo.inParamValue = reflect.New(typ.In(parIdx).Elem())
rpcMethodInfo.inParam = reflect.New(typ.In(parIdx).Elem()).Interface() rpcMethodInfo.inParam = reflect.New(typ.In(parIdx).Elem()).Interface()
pt,_ := GetProcessorType(rpcMethodInfo.inParamValue.Interface()) pt, _ := GetProcessorType(rpcMethodInfo.inParamValue.Interface())
rpcMethodInfo.rpcProcessorType = pt rpcMethodInfo.rpcProcessorType = pt
parIdx++ parIdx++
if parIdx< typ.NumIn() { if parIdx < typ.NumIn() {
rpcMethodInfo.outParamValue = reflect.New(typ.In(parIdx).Elem()) rpcMethodInfo.outParamValue = reflect.New(typ.In(parIdx).Elem())
} }
@@ -178,9 +182,9 @@ func (handler *RpcHandler) suitableMethods(method reflect.Method) error {
return nil return nil
} }
func (handler *RpcHandler) RegisterRpc(rpcHandler IRpcHandler) error { func (handler *RpcHandler) RegisterRpc(rpcHandler IRpcHandler) error {
typ := reflect.TypeOf(rpcHandler) typ := reflect.TypeOf(rpcHandler)
for m:=0;m<typ.NumMethod();m++{ for m := 0; m < typ.NumMethod(); m++ {
method := typ.Method(m) method := typ.Method(m)
err := handler.suitableMethods(method) err := handler.suitableMethods(method)
if err != nil { if err != nil {
@@ -191,20 +195,20 @@ func (handler *RpcHandler) RegisterRpc(rpcHandler IRpcHandler) error {
return nil return nil
} }
func (handler *RpcHandler) HandlerRpcResponseCB(call *Call){ func (handler *RpcHandler) HandlerRpcResponseCB(call *Call) {
defer func() { defer func() {
if r := recover(); r != nil { if r := recover(); r != nil {
buf := make([]byte, 4096) buf := make([]byte, 4096)
l := runtime.Stack(buf, false) l := runtime.Stack(buf, false)
errString := fmt.Sprint(r) errString := fmt.Sprint(r)
log.SError("core dump info[",errString,"]\n",string(buf[:l])) log.SError("core dump info[", errString, "]\n", string(buf[:l]))
} }
}() }()
if call.Err == nil { if call.Err == nil {
call.callback.Call([]reflect.Value{reflect.ValueOf(call.Reply),nilError}) call.callback.Call([]reflect.Value{reflect.ValueOf(call.Reply), nilError})
}else{ } else {
call.callback.Call([]reflect.Value{reflect.ValueOf(call.Reply),reflect.ValueOf(call.Err)}) call.callback.Call([]reflect.Value{reflect.ValueOf(call.Reply), reflect.ValueOf(call.Err)})
} }
ReleaseCall(call) ReleaseCall(call)
} }
@@ -216,25 +220,23 @@ func (handler *RpcHandler) HandlerRpcRequest(request *RpcRequest) {
defer func() { defer func() {
if r := recover(); r != nil { if r := recover(); r != nil {
buf := make([]byte, 4096) buf := make([]byte, 4096)
l := runtime.Stack(buf, false) l := runtime.Stack(buf, false)
errString := fmt.Sprint(r) errString := fmt.Sprint(r)
log.SError("Handler Rpc ",request.RpcRequestData.GetServiceMethod()," Core dump info[",errString,"]\n",string(buf[:l])) log.SError("Handler Rpc ", request.RpcRequestData.GetServiceMethod(), " Core dump info[", errString, "]\n", string(buf[:l]))
rpcErr := RpcError("call error : core dumps") rpcErr := RpcError("call error : core dumps")
if request.requestHandle!=nil { if request.requestHandle != nil {
request.requestHandle(nil,rpcErr) request.requestHandle(nil, rpcErr)
} }
} }
}() }()
//如果是原始RPC请求 //如果是原始RPC请求
rawRpcId := request.RpcRequestData.GetRpcMethodId() rawRpcId := request.RpcRequestData.GetRpcMethodId()
if rawRpcId>0 { if rawRpcId > 0 {
v,ok := handler.mapRawFunctions[rawRpcId] v, ok := handler.mapRawFunctions[rawRpcId]
if ok == false { if ok == false {
log.SError("RpcHandler cannot find request rpc id",rawRpcId) log.SError("RpcHandler cannot find request rpc id", rawRpcId)
return return
} }
@@ -243,12 +245,12 @@ func (handler *RpcHandler) HandlerRpcRequest(request *RpcRequest) {
} }
//普通的rpc请求 //普通的rpc请求
v,ok := handler.mapFunctions[request.RpcRequestData.GetServiceMethod()] v, ok := handler.mapFunctions[request.RpcRequestData.GetServiceMethod()]
if ok == false { if ok == false {
err := "RpcHandler "+handler.rpcHandler.GetName()+"cannot find "+request.RpcRequestData.GetServiceMethod() err := "RpcHandler " + handler.rpcHandler.GetName() + "cannot find " + request.RpcRequestData.GetServiceMethod()
log.SError(err) log.SError(err)
if request.requestHandle!=nil { if request.requestHandle != nil {
request.requestHandle(nil,RpcError(err)) request.requestHandle(nil, RpcError(err))
} }
return return
} }
@@ -256,27 +258,27 @@ func (handler *RpcHandler) HandlerRpcRequest(request *RpcRequest) {
var paramList []reflect.Value var paramList []reflect.Value
var err error var err error
//生成Call参数 //生成Call参数
paramList = append(paramList,reflect.ValueOf(handler.GetRpcHandler())) //接受者 paramList = append(paramList, reflect.ValueOf(handler.GetRpcHandler())) //接受者
if v.hasResponder == true { if v.hasResponder == true {
if request.requestHandle!=nil { if request.requestHandle != nil {
responder := reflect.ValueOf(request.requestHandle) responder := reflect.ValueOf(request.requestHandle)
paramList = append(paramList,responder) paramList = append(paramList, responder)
}else{ } else {
paramList = append(paramList,requestHandlerNull) paramList = append(paramList, requestHandlerNull)
} }
} }
paramList = append(paramList,reflect.ValueOf(request.inParam)) paramList = append(paramList, reflect.ValueOf(request.inParam))
var oParam reflect.Value var oParam reflect.Value
if v.outParamValue.IsValid() { if v.outParamValue.IsValid() {
if request.localReply!=nil { if request.localReply != nil {
oParam = reflect.ValueOf(request.localReply) //输出参数 oParam = reflect.ValueOf(request.localReply) //输出参数
}else{ } else {
oParam = reflect.New(v.outParamValue.Type().Elem()) oParam = reflect.New(v.outParamValue.Type().Elem())
} }
paramList = append(paramList,oParam) //输出参数 paramList = append(paramList, oParam) //输出参数
}else if request.requestHandle != nil && v.hasResponder==false{ //调用方有返回值，但被调用函数没有返回参数 } else if request.requestHandle != nil && v.hasResponder == false { //调用方有返回值，但被调用函数没有返回参数
rErr := "Call Rpc "+request.RpcRequestData.GetServiceMethod()+" without return parameter!" rErr := "Call Rpc " + request.RpcRequestData.GetServiceMethod() + " without return parameter!"
log.SError(rErr) log.SError(rErr)
request.requestHandle(nil, RpcError(rErr)) request.requestHandle(nil, RpcError(rErr))
return return
@@ -287,24 +289,24 @@ func (handler *RpcHandler) HandlerRpcRequest(request *RpcRequest) {
err = errInter.(error) err = errInter.(error)
} }
if request.requestHandle!=nil && v.hasResponder==false { if request.requestHandle != nil && v.hasResponder == false {
request.requestHandle(oParam.Interface(), ConvertError(err)) request.requestHandle(oParam.Interface(), ConvertError(err))
} }
} }
func (handler *RpcHandler) CallMethod(ServiceMethod string,param interface{},reply interface{}) error{ func (handler *RpcHandler) CallMethod(ServiceMethod string, param interface{}, reply interface{}) error {
var err error var err error
v,ok := handler.mapFunctions[ServiceMethod] v, ok := handler.mapFunctions[ServiceMethod]
if ok == false { if ok == false {
err = errors.New("RpcHandler "+ handler.rpcHandler.GetName()+" cannot find"+ServiceMethod) err = errors.New("RpcHandler " + handler.rpcHandler.GetName() + " cannot find" + ServiceMethod)
log.SError(err.Error()) log.SError(err.Error())
return err return err
} }
var paramList []reflect.Value var paramList []reflect.Value
paramList = append(paramList,reflect.ValueOf(handler.GetRpcHandler())) //接受者 paramList = append(paramList, reflect.ValueOf(handler.GetRpcHandler())) //接受者
paramList = append(paramList,reflect.ValueOf(param)) paramList = append(paramList, reflect.ValueOf(param))
paramList = append(paramList,reflect.ValueOf(reply)) //输出参数 paramList = append(paramList, reflect.ValueOf(reply)) //输出参数
returnValues := v.method.Func.Call(paramList) returnValues := v.method.Func.Call(paramList)
errInter := returnValues[0].Interface() errInter := returnValues[0].Interface()
@@ -315,32 +317,32 @@ func (handler *RpcHandler) CallMethod(ServiceMethod string,param interface{},rep
return err return err
} }
func (handler *RpcHandler) goRpc(processor IRpcProcessor,bCast bool,nodeId int,serviceMethod string,args interface{}) error { func (handler *RpcHandler) goRpc(processor IRpcProcessor, bCast bool, nodeId int, serviceMethod string, args interface{}) error {
var pClientList [maxClusterNode]*Client var pClientList [maxClusterNode]*Client
err,count := handler.funcRpcClient(nodeId,serviceMethod,pClientList[:]) err, count := handler.funcRpcClient(nodeId, serviceMethod, pClientList[:])
if count==0 { if count == 0 {
if err != nil { if err != nil {
log.SError("Call ",serviceMethod," is error:",err.Error()) log.SError("Call ", serviceMethod, " is error:", err.Error())
}else{ } else {
log.SError("Can not find ",serviceMethod) log.SError("Can not find ", serviceMethod)
} }
return err return err
} }
if count > 1 && bCast == false{ if count > 1 && bCast == false {
log.SError("Cannot call %s more then 1 node!",serviceMethod) log.SError("Cannot call %s more then 1 node!", serviceMethod)
return errors.New("cannot call more then 1 node") return errors.New("cannot call more then 1 node")
} }
//2.rpcClient调用 //2.rpcClient调用
//如果调用本结点服务 //如果调用本结点服务
for i:=0;i<count;i++{ for i := 0; i < count; i++ {
if pClientList[i].bSelfNode == true { if pClientList[i].bSelfNode == true {
pLocalRpcServer:= handler.funcRpcServer() pLocalRpcServer := handler.funcRpcServer()
//判断是否是同一服务 //判断是否是同一服务
findIndex := strings.Index(serviceMethod,".") findIndex := strings.Index(serviceMethod, ".")
if findIndex==-1 { if findIndex == -1 {
sErr := errors.New("Call serviceMethod "+serviceMethod+" is error!") sErr := errors.New("Call serviceMethod " + serviceMethod + " is error!")
log.SError(sErr.Error()) log.SError(sErr.Error())
err = sErr err = sErr
@@ -349,11 +351,11 @@ func (handler *RpcHandler) goRpc(processor IRpcProcessor,bCast bool,nodeId int,s
serviceName := serviceMethod[:findIndex] serviceName := serviceMethod[:findIndex]
if serviceName == handler.rpcHandler.GetName() { //自己服务调用 if serviceName == handler.rpcHandler.GetName() { //自己服务调用
//调用自己rpcHandler处理器 //调用自己rpcHandler处理器
return pLocalRpcServer.myselfRpcHandlerGo(serviceName,serviceMethod,args,nil) return pLocalRpcServer.myselfRpcHandlerGo(serviceName, serviceMethod, args, nil)
} }
//其他的rpcHandler的处理器 //其他的rpcHandler的处理器
pCall := pLocalRpcServer.selfNodeRpcHandlerGo(processor,pClientList[i],true,serviceName,0,serviceMethod,args,nil,nil) pCall := pLocalRpcServer.selfNodeRpcHandlerGo(processor, pClientList[i], true, serviceName, 0, serviceMethod, args, nil, nil)
if pCall.Err!=nil { if pCall.Err != nil {
err = pCall.Err err = pCall.Err
} }
pClientList[i].RemovePending(pCall.Seq) pClientList[i].RemovePending(pCall.Seq)
@@ -362,8 +364,8 @@ func (handler *RpcHandler) goRpc(processor IRpcProcessor,bCast bool,nodeId int,s
} }
//跨node调用 //跨node调用
pCall := pClientList[i].Go(true,serviceMethod,args,nil) pCall := pClientList[i].Go(true, serviceMethod, args, nil)
if pCall.Err!=nil { if pCall.Err != nil {
err = pCall.Err err = pCall.Err
} }
pClientList[i].RemovePending(pCall.Seq) pClientList[i].RemovePending(pCall.Seq)
@@ -373,17 +375,17 @@ func (handler *RpcHandler) goRpc(processor IRpcProcessor,bCast bool,nodeId int,s
return err return err
} }
func (handler *RpcHandler) callRpc(nodeId int,serviceMethod string,args interface{},reply interface{}) error { func (handler *RpcHandler) callRpc(nodeId int, serviceMethod string, args interface{}, reply interface{}) error {
var pClientList [maxClusterNode]*Client var pClientList [maxClusterNode]*Client
err,count := handler.funcRpcClient(nodeId,serviceMethod,pClientList[:]) err, count := handler.funcRpcClient(nodeId, serviceMethod, pClientList[:])
if err != nil { if err != nil {
log.SError("Call serviceMethod is error:",err.Error()) log.SError("Call serviceMethod is error:", err.Error())
return err return err
}else if count <=0 { } else if count <= 0 {
err = errors.New("Call serviceMethod is error:cannot find "+serviceMethod) err = errors.New("Call serviceMethod is error:cannot find " + serviceMethod)
log.SError(err.Error()) log.SError(err.Error())
return err return err
}else if count > 1 { } else if count > 1 {
log.SError("Cannot call more then 1 node!") log.SError("Cannot call more then 1 node!")
return errors.New("cannot call more then 1 node") return errors.New("cannot call more then 1 node")
} }
@@ -392,21 +394,21 @@ func (handler *RpcHandler) callRpc(nodeId int,serviceMethod string,args interfac
//如果调用本结点服务 //如果调用本结点服务
pClient := pClientList[0] pClient := pClientList[0]
if pClient.bSelfNode == true { if pClient.bSelfNode == true {
pLocalRpcServer:= handler.funcRpcServer() pLocalRpcServer := handler.funcRpcServer()
//判断是否是同一服务 //判断是否是同一服务
findIndex := strings.Index(serviceMethod,".") findIndex := strings.Index(serviceMethod, ".")
if findIndex==-1 { if findIndex == -1 {
err := errors.New("Call serviceMethod "+serviceMethod+"is error!") err := errors.New("Call serviceMethod " + serviceMethod + "is error!")
log.SError(err.Error()) log.SError(err.Error())
return err return err
} }
serviceName := serviceMethod[:findIndex] serviceName := serviceMethod[:findIndex]
if serviceName == handler.rpcHandler.GetName() { //自己服务调用 if serviceName == handler.rpcHandler.GetName() { //自己服务调用
//调用自己rpcHandler处理器 //调用自己rpcHandler处理器
return pLocalRpcServer.myselfRpcHandlerGo(serviceName,serviceMethod,args,reply) return pLocalRpcServer.myselfRpcHandlerGo(serviceName, serviceMethod, args, reply)
} }
//其他的rpcHandler的处理器 //其他的rpcHandler的处理器
pCall := pLocalRpcServer.selfNodeRpcHandlerGo(nil,pClient,false,serviceName,0,serviceMethod,args,reply,nil) pCall := pLocalRpcServer.selfNodeRpcHandlerGo(nil, pClient, false, serviceName, 0, serviceMethod, args, reply, nil)
err = pCall.Done().Err err = pCall.Done().Err
pClient.RemovePending(pCall.Seq) pClient.RemovePending(pCall.Seq)
ReleaseCall(pCall) ReleaseCall(pCall)
@@ -414,7 +416,7 @@ func (handler *RpcHandler) callRpc(nodeId int,serviceMethod string,args interfac
} }
//跨node调用 //跨node调用
pCall := pClient.Go(false,serviceMethod,args,reply) pCall := pClient.Go(false, serviceMethod, args, reply)
if pCall.Err != nil { if pCall.Err != nil {
err = pCall.Err err = pCall.Err
ReleaseCall(pCall) ReleaseCall(pCall)
@@ -426,42 +428,42 @@ func (handler *RpcHandler) callRpc(nodeId int,serviceMethod string,args interfac
return err return err
} }
func (handler *RpcHandler) asyncCallRpc(nodeId int,serviceMethod string,args interface{},callback interface{}) error { func (handler *RpcHandler) asyncCallRpc(nodeId int, serviceMethod string, args interface{}, callback interface{}) error {
fVal := reflect.ValueOf(callback) fVal := reflect.ValueOf(callback)
if fVal.Kind()!=reflect.Func{ if fVal.Kind() != reflect.Func {
err := errors.New("call "+serviceMethod+" input callback param is error!") err := errors.New("call " + serviceMethod + " input callback param is error!")
log.SError(err.Error()) log.SError(err.Error())
return err return err
} }
if fVal.Type().NumIn()!= 2 { if fVal.Type().NumIn() != 2 {
err := errors.New("call "+serviceMethod+" callback param function is error!") err := errors.New("call " + serviceMethod + " callback param function is error!")
log.SError(err.Error()) log.SError(err.Error())
return err return err
} }
if fVal.Type().In(0).Kind() != reflect.Ptr || fVal.Type().In(1).String() != "error"{ if fVal.Type().In(0).Kind() != reflect.Ptr || fVal.Type().In(1).String() != "error" {
err := errors.New("call "+serviceMethod+" callback param function is error!") err := errors.New("call " + serviceMethod + " callback param function is error!")
log.SError(err.Error()) log.SError(err.Error())
return err return err
} }
reply := reflect.New(fVal.Type().In(0).Elem()).Interface() reply := reflect.New(fVal.Type().In(0).Elem()).Interface()
var pClientList [maxClusterNode]*Client var pClientList [maxClusterNode]*Client
err,count := handler.funcRpcClient(nodeId,serviceMethod,pClientList[:]) err, count := handler.funcRpcClient(nodeId, serviceMethod, pClientList[:])
if count==0||err != nil { if count == 0 || err != nil {
strNodeId := strconv.Itoa(nodeId) strNodeId := strconv.Itoa(nodeId)
if err == nil { if err == nil {
err = errors.New("cannot find rpcClient from nodeId "+strNodeId+" "+serviceMethod) err = errors.New("cannot find rpcClient from nodeId " + strNodeId + " " + serviceMethod)
} }
fVal.Call([]reflect.Value{reflect.ValueOf(reply),reflect.ValueOf(err)}) fVal.Call([]reflect.Value{reflect.ValueOf(reply), reflect.ValueOf(err)})
log.SError("Call serviceMethod is error:",err.Error()) log.SError("Call serviceMethod is error:", err.Error())
return nil return nil
} }
if count > 1 { if count > 1 {
err := errors.New("cannot call more then 1 node") err := errors.New("cannot call more then 1 node")
fVal.Call([]reflect.Value{reflect.ValueOf(reply),reflect.ValueOf(err)}) fVal.Call([]reflect.Value{reflect.ValueOf(reply), reflect.ValueOf(err)})
log.SError(err.Error()) log.SError(err.Error())
return nil return nil
} }
@@ -470,84 +472,84 @@ func (handler *RpcHandler) asyncCallRpc(nodeId int,serviceMethod string,args int
//如果调用本结点服务 //如果调用本结点服务
pClient := pClientList[0] pClient := pClientList[0]
if pClient.bSelfNode == true { if pClient.bSelfNode == true {
pLocalRpcServer:= handler.funcRpcServer() pLocalRpcServer := handler.funcRpcServer()
//判断是否是同一服务 //判断是否是同一服务
findIndex := strings.Index(serviceMethod,".") findIndex := strings.Index(serviceMethod, ".")
if findIndex==-1 { if findIndex == -1 {
err := errors.New("Call serviceMethod "+serviceMethod+" is error!") err := errors.New("Call serviceMethod " + serviceMethod + " is error!")
fVal.Call([]reflect.Value{reflect.ValueOf(reply),reflect.ValueOf(err)}) fVal.Call([]reflect.Value{reflect.ValueOf(reply), reflect.ValueOf(err)})
log.SError(err.Error()) log.SError(err.Error())
return nil return nil
} }
serviceName := serviceMethod[:findIndex] serviceName := serviceMethod[:findIndex]
//调用自己rpcHandler处理器 //调用自己rpcHandler处理器
if serviceName == handler.rpcHandler.GetName() { //自己服务调用 if serviceName == handler.rpcHandler.GetName() { //自己服务调用
err := pLocalRpcServer.myselfRpcHandlerGo(serviceName,serviceMethod,args,reply) err := pLocalRpcServer.myselfRpcHandlerGo(serviceName, serviceMethod, args, reply)
if err == nil { if err == nil {
fVal.Call([]reflect.Value{reflect.ValueOf(reply),nilError}) fVal.Call([]reflect.Value{reflect.ValueOf(reply), nilError})
}else{ } else {
fVal.Call([]reflect.Value{reflect.ValueOf(reply),reflect.ValueOf(err)}) fVal.Call([]reflect.Value{reflect.ValueOf(reply), reflect.ValueOf(err)})
} }
} }
//其他的rpcHandler的处理器 //其他的rpcHandler的处理器
err = pLocalRpcServer.selfNodeRpcHandlerAsyncGo(pClient, handler,false,serviceName,serviceMethod,args,reply,fVal) err = pLocalRpcServer.selfNodeRpcHandlerAsyncGo(pClient, handler, false, serviceName, serviceMethod, args, reply, fVal)
if err != nil { if err != nil {
fVal.Call([]reflect.Value{reflect.ValueOf(reply),reflect.ValueOf(err)}) fVal.Call([]reflect.Value{reflect.ValueOf(reply), reflect.ValueOf(err)})
} }
return nil return nil
} }
//跨node调用 //跨node调用
err = pClient.AsyncCall(handler,serviceMethod,fVal,args,reply) err = pClient.AsyncCall(handler, serviceMethod, fVal, args, reply)
if err != nil { if err != nil {
fVal.Call([]reflect.Value{reflect.ValueOf(reply),reflect.ValueOf(err)}) fVal.Call([]reflect.Value{reflect.ValueOf(reply), reflect.ValueOf(err)})
} }
return nil return nil
} }
func (handler *RpcHandler) GetName() string{ func (handler *RpcHandler) GetName() string {
return handler.rpcHandler.GetName() return handler.rpcHandler.GetName()
} }
func (handler *RpcHandler) IsSingleCoroutine() bool{ func (handler *RpcHandler) IsSingleCoroutine() bool {
return handler.rpcHandler.IsSingleCoroutine() return handler.rpcHandler.IsSingleCoroutine()
} }
func (handler *RpcHandler) AsyncCall(serviceMethod string,args interface{},callback interface{}) error { func (handler *RpcHandler) AsyncCall(serviceMethod string, args interface{}, callback interface{}) error {
return handler.asyncCallRpc(0,serviceMethod,args,callback) return handler.asyncCallRpc(0, serviceMethod, args, callback)
} }
func (handler *RpcHandler) Call(serviceMethod string,args interface{},reply interface{}) error { func (handler *RpcHandler) Call(serviceMethod string, args interface{}, reply interface{}) error {
return handler.callRpc(0,serviceMethod,args,reply) return handler.callRpc(0, serviceMethod, args, reply)
} }
func (handler *RpcHandler) Go(serviceMethod string,args interface{}) error { func (handler *RpcHandler) Go(serviceMethod string, args interface{}) error {
return handler.goRpc(nil,false,0,serviceMethod,args) return handler.goRpc(nil, false, 0, serviceMethod, args)
} }
func (handler *RpcHandler) AsyncCallNode(nodeId int,serviceMethod string,args interface{},callback interface{}) error { func (handler *RpcHandler) AsyncCallNode(nodeId int, serviceMethod string, args interface{}, callback interface{}) error {
return handler.asyncCallRpc(nodeId,serviceMethod,args,callback) return handler.asyncCallRpc(nodeId, serviceMethod, args, callback)
} }
func (handler *RpcHandler) CallNode(nodeId int,serviceMethod string,args interface{},reply interface{}) error { func (handler *RpcHandler) CallNode(nodeId int, serviceMethod string, args interface{}, reply interface{}) error {
return handler.callRpc(nodeId,serviceMethod,args,reply) return handler.callRpc(nodeId, serviceMethod, args, reply)
} }
func (handler *RpcHandler) GoNode(nodeId int,serviceMethod string,args interface{}) error { func (handler *RpcHandler) GoNode(nodeId int, serviceMethod string, args interface{}) error {
return handler.goRpc(nil,false,nodeId,serviceMethod,args) return handler.goRpc(nil, false, nodeId, serviceMethod, args)
} }
func (handler *RpcHandler) CastGo(serviceMethod string,args interface{}) error{ func (handler *RpcHandler) CastGo(serviceMethod string, args interface{}) error {
return handler.goRpc(nil,true,0,serviceMethod,args) return handler.goRpc(nil, true, 0, serviceMethod, args)
} }
func (handler *RpcHandler) RawGoNode(rpcProcessorType RpcProcessorType,nodeId int,rpcMethodId uint32,serviceName string,rawArgs IRawInputArgs) error { func (handler *RpcHandler) RawGoNode(rpcProcessorType RpcProcessorType, nodeId int, rpcMethodId uint32, serviceName string, rawArgs IRawInputArgs) error {
processor := GetProcessor(uint8(rpcProcessorType)) processor := GetProcessor(uint8(rpcProcessorType))
err,count := handler.funcRpcClient(nodeId,serviceName,handler.pClientList) err, count := handler.funcRpcClient(nodeId, serviceName, handler.pClientList)
if count==0||err != nil { if count == 0 || err != nil {
//args.DoGc() //args.DoGc()
log.SError("Call serviceMethod is error:",err.Error()) log.SError("Call serviceMethod is error:", err.Error())
return err return err
} }
if count > 1 { if count > 1 {
@@ -559,20 +561,20 @@ func (handler *RpcHandler) RawGoNode(rpcProcessorType RpcProcessorType,nodeId in
//2.rpcClient调用 //2.rpcClient调用
//如果调用本结点服务 //如果调用本结点服务
for i:=0;i<count;i++{ for i := 0; i < count; i++ {
if handler.pClientList[i].bSelfNode == true { if handler.pClientList[i].bSelfNode == true {
pLocalRpcServer:= handler.funcRpcServer() pLocalRpcServer := handler.funcRpcServer()
//调用自己rpcHandler处理器 //调用自己rpcHandler处理器
if serviceName == handler.rpcHandler.GetName() { //自己服务调用 if serviceName == handler.rpcHandler.GetName() { //自己服务调用
err:= pLocalRpcServer.myselfRpcHandlerGo(serviceName,serviceName,rawArgs.GetRawData(),nil) err := pLocalRpcServer.myselfRpcHandlerGo(serviceName, serviceName, rawArgs.GetRawData(), nil)
//args.DoGc() //args.DoGc()
return err return err
} }
//其他的rpcHandler的处理器 //其他的rpcHandler的处理器
pCall := pLocalRpcServer.selfNodeRpcHandlerGo(processor,handler.pClientList[i],true,serviceName,rpcMethodId,serviceName,nil,nil,rawArgs.GetRawData()) pCall := pLocalRpcServer.selfNodeRpcHandlerGo(processor, handler.pClientList[i], true, serviceName, rpcMethodId, serviceName, nil, nil, rawArgs.GetRawData())
rawArgs.DoEscape() rawArgs.DoEscape()
if pCall.Err!=nil { if pCall.Err != nil {
err = pCall.Err err = pCall.Err
} }
handler.pClientList[i].RemovePending(pCall.Seq) handler.pClientList[i].RemovePending(pCall.Seq)
@@ -581,9 +583,9 @@ func (handler *RpcHandler) RawGoNode(rpcProcessorType RpcProcessorType,nodeId in
} }
//跨node调用 //跨node调用
pCall := handler.pClientList[i].RawGo(processor,true,rpcMethodId,serviceName,rawArgs.GetRawData(),nil) pCall := handler.pClientList[i].RawGo(processor, true, rpcMethodId, serviceName, rawArgs.GetRawData(), nil)
rawArgs.DoFree() rawArgs.DoFree()
if pCall.Err!=nil { if pCall.Err != nil {
err = pCall.Err err = pCall.Err
} }
handler.pClientList[i].RemovePending(pCall.Seq) handler.pClientList[i].RemovePending(pCall.Seq)
@@ -593,38 +595,38 @@ func (handler *RpcHandler) RawGoNode(rpcProcessorType RpcProcessorType,nodeId in
return err return err
} }
func (handler *RpcHandler) RegRawRpc(rpcMethodId uint32,rawRpcCB RawRpcCallBack){ func (handler *RpcHandler) RegRawRpc(rpcMethodId uint32, rawRpcCB RawRpcCallBack) {
handler.mapRawFunctions[rpcMethodId] = rawRpcCB handler.mapRawFunctions[rpcMethodId] = rawRpcCB
} }
func (handler *RpcHandler) UnmarshalInParam(rpcProcessor IRpcProcessor,serviceMethod string,rawRpcMethodId uint32,inParam []byte) (interface{},error){ func (handler *RpcHandler) UnmarshalInParam(rpcProcessor IRpcProcessor, serviceMethod string, rawRpcMethodId uint32, inParam []byte) (interface{}, error) {
if rawRpcMethodId>0 { if rawRpcMethodId > 0 {
v,ok := handler.mapRawFunctions[rawRpcMethodId] v, ok := handler.mapRawFunctions[rawRpcMethodId]
if ok == false { if ok == false {
strRawRpcMethodId := strconv.FormatUint(uint64(rawRpcMethodId),10) strRawRpcMethodId := strconv.FormatUint(uint64(rawRpcMethodId), 10)
err := errors.New("RpcHandler cannot find request rpc id "+strRawRpcMethodId) err := errors.New("RpcHandler cannot find request rpc id " + strRawRpcMethodId)
log.SError(err.Error()) log.SError(err.Error())
return nil,err return nil, err
} }
msg,err := v.Unmarshal(inParam) msg, err := v.Unmarshal(inParam)
if err != nil { if err != nil {
strRawRpcMethodId := strconv.FormatUint(uint64(rawRpcMethodId),10) strRawRpcMethodId := strconv.FormatUint(uint64(rawRpcMethodId), 10)
err := errors.New("RpcHandler cannot Unmarshal rpc id "+strRawRpcMethodId) err := errors.New("RpcHandler cannot Unmarshal rpc id " + strRawRpcMethodId)
log.SError(err.Error()) log.SError(err.Error())
return nil,err return nil, err
} }
return msg,err return msg, err
} }
v,ok := handler.mapFunctions[serviceMethod] v, ok := handler.mapFunctions[serviceMethod]
if ok == false { if ok == false {
return nil,errors.New( "RpcHandler "+handler.rpcHandler.GetName()+"cannot find "+serviceMethod) return nil, errors.New("RpcHandler " + handler.rpcHandler.GetName() + " cannot find " + serviceMethod)
} }
var err error var err error
param := reflect.New(v.inParamValue.Type().Elem()).Interface() param := reflect.New(v.inParamValue.Type().Elem()).Interface()
err = rpcProcessor.Unmarshal(inParam,param) err = rpcProcessor.Unmarshal(inParam, param)
return param,err return param, err
} }