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

优化原始RPC，支持interface{}参数

Browse Source
This commit is contained in:
boyce
2020-11-05 20:25:14 +08:00
parent 2455054b0f
commit 80dfe2ee2e
2 changed files with 59 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
rpc/rpchandler.go
View File

@@ -147,11 +147,15 @@ func (handler *RpcHandler) suitableMethods(method reflect.Method) error {
return fmt.Errorf("%s Unsupported parameter types!",method.Name)
}
}
rpcMethodInfo.inParamValue = reflect.New(typ.In(parIdx).Elem()) //append(rpcMethodInfo.iparam,)
rpcMethodInfo.inParam = reflect.New(typ.In(parIdx).Elem()).Interface()
pt,_ := GetProcessorType(rpcMethodInfo.inParamValue.Interface())
rpcMethodInfo.rpcProcessorType = pt
a := typ.In(parIdx).Kind()
if a == reflect.Interface {
rpcMethodInfo.inParam = nil
}else{
rpcMethodInfo.inParamValue = reflect.New(typ.In(parIdx).Elem()) //append(rpcMethodInfo.iparam,)
rpcMethodInfo.inParam = reflect.New(typ.In(parIdx).Elem()).Interface()
pt,_ := GetProcessorType(rpcMethodInfo.inParamValue.Interface())
rpcMethodInfo.rpcProcessorType = pt
}
parIdx++
if parIdx< typ.NumIn() {
@@ -251,17 +255,23 @@ func (handler *RpcHandler) HandlerRpcRequest(request *RpcRequest) {
}
if request.bLocalRequest == false {
err = request.rpcProcessor.Unmarshal(request.RpcRequestData.GetInParam(),iParam)
if err!=nil {
rErr := "Call Rpc "+request.RpcRequestData.GetServiceMethod()+" Param error "+err.Error()
log.Error(rErr)
if request.requestHandle!=nil {
request.requestHandle(nil, RpcError(rErr))
if iParam == nil {
iParam = request.RpcRequestData.GetInParam()
}else{
err = request.rpcProcessor.Unmarshal(request.RpcRequestData.GetInParam(),iParam)
if err!=nil {
rErr := "Call Rpc "+request.RpcRequestData.GetServiceMethod()+" Param error "+err.Error()
log.Error(rErr)
if request.requestHandle!=nil {
request.requestHandle(nil, RpcError(rErr))
}
return
}
return
}
}else {
if request.inputArgs!=nil {
if iParam == nil {
iParam = request.inputArgs.GetRawData()
}else if request.inputArgs!=nil {
err = request.rpcProcessor.Unmarshal(request.inputArgs.GetRawData(),iParam)
if err!=nil {
rErr := "Call Rpc "+request.RpcRequestData.GetServiceMethod()+" Param error "+err.Error()

38
sysservice/tcpgateway/Router.go
View File

@@ -6,6 +6,7 @@ import (
"github.com/duanhf2012/origin/node"
"github.com/duanhf2012/origin/rpc"
"github.com/duanhf2012/origin/sysservice/tcpservice"
"strconv"
"strings"
)
@@ -88,11 +89,41 @@ func (r *Router) loadCfg(cfg interface{}){
//error ...
continue
}
msgId,ok := iMsgId.(float64)
msgIdList,ok := iMsgId.(string)
if ok == false {
//error ...
continue
}
var startMsgId int
var endMsgId int
var err error
sliceId := strings.Split(msgIdList,"-")
if len(sliceId) == 1 {
startMsgId,err = strconv.Atoi(msgIdList)
if err != nil {
log.Fatal("TcpGateService %s config is error!",iRpc.(string))
continue
}
endMsgId = startMsgId
} else if len(sliceId) == 2 {
startMsgId,err = strconv.Atoi(sliceId[0])
if err != nil {
log.Fatal("TcpGateService %s config is error!",iRpc.(string))
continue
}
endMsgId,err = strconv.Atoi(sliceId[1])
if err != nil {
log.Fatal("TcpGateService %s config is error!",iRpc.(string))
continue
}
if startMsgId>endMsgId {
log.Fatal("TcpGateService %s config is error!",iRpc.(string))
continue
}
}else {
log.Fatal("TcpGateService %s config is error!",iRpc.(string))
continue
}
strService := strings.Split(iRpc.(string),".")
if len(strService)!=2 {
@@ -100,7 +131,10 @@ func (r *Router) loadCfg(cfg interface{}){
continue
}
r.mapMsgRouterInfo[uint16(msgId)] = &MsgRouterInfo{ServiceName: strService[0],Rpc: iRpc.(string),LoadBalanceType: iLoadBalanceType.(string)}
msgInfo := &MsgRouterInfo{ServiceName: strService[0],Rpc: iRpc.(string),LoadBalanceType: iLoadBalanceType.(string)}
for i:=startMsgId;i<=endMsgId;i++{
r.mapMsgRouterInfo[uint16(i)] = msgInfo
}
}
//parse EventRouter