From 2fc1217b18456697535b782e7e6c9c1836443e33 Mon Sep 17 00:00:00 2001
From: boyce <boyce@atbc.com>
Date: Wed, 4 Nov 2020 10:56:11 +0800
Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E7=A7=81=E6=9C=89=E6=9C=8D?=
 =?UTF-8?q?=E5=8A=A1=E5=8A=9F=E8=83=BD(=E7=BB=93=E7=82=B9=E4=B8=AD?=
 =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E9=85=8D=E7=BD=AE=E5=8A=A0=E5=89=8D=E7=BC=80?=
 =?UTF-8?q?=5F)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 cluster/configdiscovery.go      | 14 +++++++++
 cluster/parsecfg.go             |  7 +++++
 rpc/rpc.go                      |  4 +++
 rpc/rpchandler.go               | 56 ++++++++++++++++++---------------
 sysservice/tcpgateway/Router.go | 14 ++++-----
 5 files changed, 63 insertions(+), 32 deletions(-)

diff --git a/cluster/configdiscovery.go b/cluster/configdiscovery.go
index a3b1685..1735f21 100644
--- a/cluster/configdiscovery.go
+++ b/cluster/configdiscovery.go
@@ -1,5 +1,6 @@
 package cluster
 
+import "strings"
 
 type ConfigDiscovery struct {
 	funDelService FunDelNode
@@ -7,6 +8,17 @@ type ConfigDiscovery struct {
 	localNodeId int
 }
 
+func (discovery *ConfigDiscovery) privateService(nodeInfo *NodeInfo){
+	var serviceList []string
+	for _,s := range nodeInfo.ServiceList {
+		if strings.HasPrefix(s,"_") {
+			continue
+		}
+		serviceList = append(serviceList,s)
+	}
+	nodeInfo.ServiceList = serviceList
+}
+
 func (discovery *ConfigDiscovery) Init(localNodeId int) error{
 	discovery.localNodeId = localNodeId
 
@@ -20,6 +32,8 @@ func (discovery *ConfigDiscovery) Init(localNodeId int) error{
 		if nodeInfo.NodeId == localNodeId {
 			continue
 		}
+		//去除私有服务
+		discovery.privateService(&nodeInfo)
 		discovery.funSetService(&nodeInfo)
 	}
 
diff --git a/cluster/parsecfg.go b/cluster/parsecfg.go
index 346ea6a..9b0ef7c 100644
--- a/cluster/parsecfg.go
+++ b/cluster/parsecfg.go
@@ -146,6 +146,12 @@ func (cls *Cluster) parseLocalCfg(){
 	}
 }
 
+func (cls *Cluster) localPrivateService(localNodeInfo *NodeInfo){
+	for i:=0;i<len(localNodeInfo.ServiceList);i++{
+		localNodeInfo.ServiceList[i] = strings.TrimLeft(localNodeInfo.ServiceList[i],"_")
+	}
+}
+
 func (cls *Cluster) InitCfg(localNodeId int) error{
 	cls.localServiceCfg = map[string]interface{}{}
 	cls.mapRpc = map[int] NodeRpcInfo{}
@@ -158,6 +164,7 @@ func (cls *Cluster) InitCfg(localNodeId int) error{
 		return err
 	}
 	cls.localNodeInfo = nodeInfoList[0]
+	cls.localPrivateService(&cls.localNodeInfo)
 
 	//读取本地服务配置
 	err = cls.readLocalService(localNodeId)
diff --git a/rpc/rpc.go b/rpc/rpc.go
index 5f23dd4..da82272 100644
--- a/rpc/rpc.go
+++ b/rpc/rpc.go
@@ -88,6 +88,9 @@ func (slf *RpcRequest) Clear() *RpcRequest{
 	slf.localParam = nil
 	slf.requestHandle = nil
 	slf.callback = nil
+	slf.bLocalRequest = false
+	slf.inputArgs = nil
+	slf.rpcProcessor = nil
 	return slf
 }
 
@@ -101,6 +104,7 @@ func (call *Call) Clear() *Call{
 	call.ServiceMethod = ""
 	call.Reply = nil
 	call.Response = nil
+	call.done = nil
 	call.Err = nil
 	call.connId = 0
 	call.callback = nil
diff --git a/rpc/rpchandler.go b/rpc/rpchandler.go
index ea7f320..dcb45a6 100644
--- a/rpc/rpchandler.go
+++ b/rpc/rpchandler.go
@@ -360,8 +360,8 @@ func (handler *RpcHandler) goRpc(processor IRpcProcessor,bCast bool,nodeId int,s
 		if pClient.bSelfNode == true {
 			pLocalRpcServer:= handler.funcRpcServer()
 			//判断是否是同一服务
-			sMethod := strings.Split(serviceMethod,".")
-			if len(sMethod)!=2 {
+			findIndex := strings.Index(serviceMethod,".")
+			if findIndex==-1 {
 				sErr := fmt.Errorf("Call serviceMethod %s is error!",serviceMethod)
 				log.Error("%+v", sErr)
 				if sErr != nil {
@@ -369,13 +369,14 @@ func (handler *RpcHandler) goRpc(processor IRpcProcessor,bCast bool,nodeId int,s
 				}
 				continue
 			}
-			//调用自己rpcHandler处理器
-			if sMethod[0] == handler.rpcHandler.GetName() { //自己服务调用
-				//
-				return pLocalRpcServer.myselfRpcHandlerGo(sMethod[0],sMethod[1],args,nil)
+			serviceName := serviceMethod[:findIndex]
+			method := serviceMethod[findIndex+1:]
+			if serviceName == handler.rpcHandler.GetName() { //自己服务调用
+				//调用自己rpcHandler处理器
+				return pLocalRpcServer.myselfRpcHandlerGo(serviceName,method,args,nil)
 			}
 			//其他的rpcHandler的处理器
-			pCall := pLocalRpcServer.selfNodeRpcHandlerGo(processor,pClient,true,sMethod[0],sMethod[1],args,nil,nil)
+			pCall := pLocalRpcServer.selfNodeRpcHandlerGo(processor,pClient,true,serviceName,method,args,nil,nil)
 			if pCall.Err!=nil {
 				err = pCall.Err
 			}
@@ -412,19 +413,20 @@ func (handler *RpcHandler) callRpc(nodeId int,serviceMethod string,args interfac
 	if pClient.bSelfNode == true {
 		pLocalRpcServer:= handler.funcRpcServer()
 		//判断是否是同一服务
-		sMethod := strings.Split(serviceMethod,".")
-		if len(sMethod)!=2 {
+		findIndex := strings.Index(serviceMethod,".")
+		if findIndex==-1 {
 			err := fmt.Errorf("Call serviceMethod %s is error!",serviceMethod)
 			log.Error("%+v",err)
 			return err
 		}
-		//调用自己rpcHandler处理器
-		if sMethod[0] == handler.rpcHandler.GetName() { //自己服务调用
-			//
-			return pLocalRpcServer.myselfRpcHandlerGo(sMethod[0],sMethod[1],args,reply)
+		serviceName := serviceMethod[:findIndex]
+		method := serviceMethod[findIndex+1:]
+		if serviceName == handler.rpcHandler.GetName() { //自己服务调用
+			//调用自己rpcHandler处理器
+			return pLocalRpcServer.myselfRpcHandlerGo(serviceName,method,args,reply)
 		}
 		//其他的rpcHandler的处理器
-		pCall := pLocalRpcServer.selfNodeRpcHandlerGo(nil,pClient,false,sMethod[0],sMethod[1],args,reply,nil)
+		pCall := pLocalRpcServer.selfNodeRpcHandlerGo(nil,pClient,false,serviceName,method,args,reply,nil)
 		err = pCall.Done().Err
 		pClient.RemovePending(pCall.Seq)
 		ReleaseCall(pCall)
@@ -484,16 +486,18 @@ func (handler *RpcHandler) asyncCallRpc(nodeid int,serviceMethod string,args int
 	if pClient.bSelfNode == true {
 		pLocalRpcServer:= handler.funcRpcServer()
 		//判断是否是同一服务
-		sMethod := strings.Split(serviceMethod,".")
-		if len(sMethod)!=2 {
+		findIndex := strings.Index(serviceMethod,".")
+		if findIndex==-1 {
 			err := fmt.Errorf("Call serviceMethod %s is error!",serviceMethod)
 			fVal.Call([]reflect.Value{reflect.ValueOf(reply),reflect.ValueOf(err)})
 			log.Error("%+v",err)
 			return nil
 		}
+		serviceName := serviceMethod[:findIndex]
+		method := serviceMethod[findIndex+1:]
 		//调用自己rpcHandler处理器
-		if sMethod[0] == handler.rpcHandler.GetName() { //自己服务调用
-			err := pLocalRpcServer.myselfRpcHandlerGo(sMethod[0],sMethod[1],args,reply)
+		if serviceName == handler.rpcHandler.GetName() { //自己服务调用
+			err := pLocalRpcServer.myselfRpcHandlerGo(serviceName,method,args,reply)
 			if err == nil {
 				fVal.Call([]reflect.Value{reflect.ValueOf(reply),NilError})
 			}else{
@@ -503,13 +507,13 @@ func (handler *RpcHandler) asyncCallRpc(nodeid int,serviceMethod string,args int
 
 		//其他的rpcHandler的处理器
 		if callback!=nil {
-			err =  pLocalRpcServer.selfNodeRpcHandlerAsyncGo(pClient, handler,false,sMethod[0],sMethod[1],args,reply,fVal)
+			err =  pLocalRpcServer.selfNodeRpcHandlerAsyncGo(pClient, handler,false,serviceName,method,args,reply,fVal)
 			if err != nil {
 				fVal.Call([]reflect.Value{reflect.ValueOf(reply),reflect.ValueOf(err)})
 			}
 			return nil
 		}
-		pCall := pLocalRpcServer.selfNodeRpcHandlerGo(nil,pClient,false,sMethod[0],sMethod[1],args,reply,nil)
+		pCall := pLocalRpcServer.selfNodeRpcHandlerGo(nil,pClient,false,serviceName,method,args,reply,nil)
 		err = pCall.Done().Err
 		pClient.RemovePending(pCall.Seq)
 		ReleaseCall(pCall)
@@ -582,8 +586,8 @@ func (handler *RpcHandler) RawGoNode(rpcProcessorType RpcProcessorType,nodeId in
 		if pClient.bSelfNode == true {
 			pLocalRpcServer:= handler.funcRpcServer()
 			//判断是否是同一服务
-			sMethod := strings.Split(serviceMethod,".")
-			if len(sMethod)!=2 {
+			findIndex := strings.Index(serviceMethod,".")
+			if findIndex==-1 {
 				serr := fmt.Errorf("Call serviceMethod %s is error!",serviceMethod)
 				log.Error("%+v",serr)
 				if serr!= nil {
@@ -591,15 +595,17 @@ func (handler *RpcHandler) RawGoNode(rpcProcessorType RpcProcessorType,nodeId in
 				}
 				continue
 			}
+			serviceName := serviceMethod[:findIndex]
+			method := serviceMethod[findIndex+1:]
 			//调用自己rpcHandler处理器
-			if sMethod[0] == handler.rpcHandler.GetName() { //自己服务调用
-				err:= pLocalRpcServer.myselfRpcHandlerGo(sMethod[0],sMethod[1],args,nil)
+			if serviceName == handler.rpcHandler.GetName() { //自己服务调用
+				err:= pLocalRpcServer.myselfRpcHandlerGo(serviceName,method,args,nil)
 				args.DoGc()
 				return err
 			}
 
 			//其他的rpcHandler的处理器
-			pCall := pLocalRpcServer.selfNodeRpcHandlerGo(processor,pClient,true,sMethod[0],sMethod[1],nil,nil,args)
+			pCall := pLocalRpcServer.selfNodeRpcHandlerGo(processor,pClient,true,serviceName,method,nil,nil,args)
 			if pCall.Err!=nil {
 				err = pCall.Err
 			}
diff --git a/sysservice/tcpgateway/Router.go b/sysservice/tcpgateway/Router.go
index 3867a32..24a2a21 100644
--- a/sysservice/tcpgateway/Router.go
+++ b/sysservice/tcpgateway/Router.go
@@ -183,8 +183,8 @@ func (r *Router) GetRouterId(clientId uint64,serviceName *string) int {
 	return routerId
 }
 
-func (r *Router) SetRouterId(clientId uint64,serviceName *string,routerId int){
-	r.mapClientRouterCache[clientId][*serviceName] = routerId
+func (r *Router) SetRouterId(clientId uint64,serviceName string,routerId int){
+	r.mapClientRouterCache[clientId][serviceName] = routerId
 }
 
 type RawInputArgs struct {
@@ -212,21 +212,21 @@ func (args RawInputArgs) DoGc() {
 	network.ReleaseByteSlice(args.rawData)
 }
 
-func (r *Router) RouterMessage(clientId uint64,msgType uint16,msg []byte) {
+func (r *Router) RouterMessage(cliId uint64,msgType uint16,msg []byte) {
 	routerInfo:= r.GetMsgRouterService(msgType)
 	if routerInfo==nil {
 		log.Error("The message type is %d with no configured route!",msgType)
 		return
 	}
 
-	routerId := r.GetRouterId(clientId,&routerInfo.ServiceName)
+	routerId := r.GetRouterId(cliId,&routerInfo.ServiceName)
 	if routerId ==0 {
 		routerId = r.loadBalance.SelectNode(routerInfo.ServiceName)
-		r.SetRouterId(clientId,&routerInfo.ServiceName,routerId)
+		r.SetRouterId(cliId,routerInfo.ServiceName,routerId)
 	}
 
 	if routerId>0 {
-		r.rpcHandler.RawGoNode(rpc.RpcProcessorPb,routerId,routerInfo.Rpc,RawInputArgs{rawData: msg,clientId: clientId})
+		r.rpcHandler.RawGoNode(rpc.RpcProcessorPb,routerId,routerInfo.Rpc,RawInputArgs{rawData: msg,clientId: cliId})
 	}
 }
 
@@ -243,7 +243,7 @@ func (r *Router) RouterEvent(clientId uint64,eventType string) bool{
 	routerId := r.GetRouterId(clientId,&routerInfo.ServiceName)
 	if routerId ==0 {
 		routerId = r.loadBalance.SelectNode(routerInfo.ServiceName)
-		r.SetRouterId(clientId,&routerInfo.ServiceName,routerId)
+		r.SetRouterId(clientId,routerInfo.ServiceName,routerId)
 	}
 
 	if routerId>0 {