优化rpc调用错误日志

限制配置的服务必需安装
优化结点断开连接时删除结点

README.md

@@ -667,6 +667,7 @@ type InputData struct {
 	B int
 }
 
+// 注意RPC函数名的格式必需为RPC_FunctionName或者是RPCFunctionName，如下的RPC_Sum也可以写成RPCSum
 func (slf *TestService6) RPC_Sum(input *InputData,output *int) error{
 	*output = input.A+input.B
 	return nil

cluster/dynamicdiscovery.go

@@ -60,6 +60,21 @@ func (ds *DynamicDiscoveryMaster) addNodeInfo(nodeInfo *rpc.NodeInfo) {
 	ds.nodeInfo = append(ds.nodeInfo, nodeInfo)
 }
 
+func (ds *DynamicDiscoveryMaster) removeNodeInfo(nodeId int32) {
+	if _,ok:= ds.mapNodeInfo[nodeId];ok == false {
+		return
+	}
+
+	for i:=0;i<len(ds.nodeInfo);i++ {
+		if ds.nodeInfo[i].NodeId == nodeId {
+			ds.nodeInfo = append(ds.nodeInfo[:i],ds.nodeInfo[i+1:]...)
+			break
+		}
+	}
+
+	delete(ds.mapNodeInfo,nodeId)
+}
+
 func (ds *DynamicDiscoveryMaster) OnInit() error {
 	ds.mapNodeInfo = make(map[int32]struct{}, 20)
 	ds.RegRpcListener(ds)
@@ -103,6 +118,8 @@ func (ds *DynamicDiscoveryMaster) OnNodeDisconnect(nodeId int) {
 		return
 	}
 
+	ds.removeNodeInfo(int32(nodeId))
+
 	var notifyDiscover rpc.SubscribeDiscoverNotify
 	notifyDiscover.MasterNodeId = int32(cluster.GetLocalNodeInfo().NodeId)
 	notifyDiscover.DelNodeId = int32(nodeId)

go.mod

@@ -23,8 +23,8 @@ require (
 	github.com/xdg-go/scram v1.0.2 // indirect
 	github.com/xdg-go/stringprep v1.0.2 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
-	golang.org/x/crypto v0.0.0-20201216223049-8b5274cf687f // indirect
+	golang.org/x/crypto v0.1.0 // indirect
 	golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 // indirect
-	golang.org/x/text v0.3.6 // indirect
+	golang.org/x/text v0.4.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 )

go.sum

@@ -58,8 +58,9 @@ go.mongodb.org/mongo-driver v1.9.1/go.mod h1:0sQWfOeY63QTntERDJJ/0SuKK0T1uVSgKCu
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20201216223049-8b5274cf687f h1:aZp0e2vLN4MToVqnjNEYEtrEA8RH8U8FN1CU7JgqsPU=
 golang.org/x/crypto v0.0.0-20201216223049-8b5274cf687f/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
+golang.org/x/crypto v0.1.0 h1:MDRAIl0xIo9Io2xV565hzXHw3zVseKrJKodhohM5CjU=
+golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
@@ -79,8 +80,8 @@ golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXR
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
-golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.4.0 h1:BrVqGRd7+k1DiOgtnFvAkoQEWQvBc25ouMJM6429SFg=
+golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190531172133-b3315ee88b7d/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=

node/node.go

@@ -155,16 +155,21 @@ func initNode(id int) {
 	//2.顺序安装服务
 	serviceOrder := cluster.GetCluster().GetLocalNodeInfo().ServiceList
 	for _,serviceName:= range serviceOrder{
+		bSetup := false
 		for _, s := range preSetupService {
 			if s.GetName() != serviceName {
 				continue
 			}
-
+			bSetup = true
 			pServiceCfg := cluster.GetCluster().GetServiceCfg(s.GetName())
 			s.Init(s, cluster.GetRpcClient, cluster.GetRpcServer, pServiceCfg)
 
 			service.Setup(s)
 		}
+
+		if bSetup == false {
+			log.SFatal("Service name "+serviceName+" configuration error")
+		}
 	}
 
 	//3.service初始化

rpc/gogopbprocessor.go

@@ -41,7 +41,10 @@ func (slf *GoGoPBProcessor) Marshal(v interface{}) ([]byte, error){
 }
 
 func (slf *GoGoPBProcessor) Unmarshal(data []byte, msg interface{}) error{
-	protoMsg := msg.(proto.Message)
+	protoMsg,ok := msg.(proto.Message)
+	if ok == false {
+		return fmt.Errorf("%+v is not of proto.Message type",msg)
+	}
 	return proto.Unmarshal(data, protoMsg)
 }
 

rpc/rpchandler.go

@@ -138,7 +138,7 @@ func (handler *RpcHandler) isExportedOrBuiltinType(t reflect.Type) bool {
 
 func (handler *RpcHandler) suitableMethods(method reflect.Method) error {
 	//只有RPC_开头的才能被调用
-	if strings.Index(method.Name, "RPC_") != 0 {
+	if strings.Index(method.Name, "RPC_") != 0 && strings.Index(method.Name, "RPC") != 0 {
 		return nil
 	}
 
@@ -496,7 +496,11 @@ func (handler *RpcHandler) asyncCallRpc(nodeId int, serviceMethod string, args i
 	err, count := handler.funcRpcClient(nodeId, serviceMethod, pClientList[:])
 	if count == 0 || err != nil {
 		if err == nil {
-			err = fmt.Errorf("cannot find %s from nodeId %d",serviceMethod,nodeId)
+			if nodeId  > 0 {
+				err = fmt.Errorf("cannot find %s from nodeId %d",serviceMethod,nodeId)
+			}else {
+				err = fmt.Errorf("No %s service found in the origin network",serviceMethod)
+			}
 		}
 		fVal.Call([]reflect.Value{reflect.ValueOf(reply), reflect.ValueOf(err)})
 		log.SError("Call serviceMethod is error:", err.Error())

rpc/server.go

@@ -147,7 +147,6 @@ func (agent *RpcAgent) Run() {
 				ReleaseRpcRequest(req)
 				continue
 			} else {
-				//will close tcpconn
 				ReleaseRpcRequest(req)
 				break
 			}
@@ -303,21 +302,22 @@ func (server *Server) selfNodeRpcHandlerGo(processor IRpcProcessor, client *Clie
 			if reply != nil && Returns != reply && Returns != nil {
 				byteReturns, err := req.rpcProcessor.Marshal(Returns)
 				if err != nil {
-					log.SError("returns data cannot be marshal ", callSeq)
-					ReleaseRpcRequest(req)
-				}
-
-				err = req.rpcProcessor.Unmarshal(byteReturns, reply)
-				if err != nil {
-					log.SError("returns data cannot be Unmarshal ", callSeq)
-					ReleaseRpcRequest(req)
+					Err = ConvertError(err)
+					log.SError("returns data cannot be marshal,callSeq is ", callSeq," error is ",err.Error())
+				}else{
+					err = req.rpcProcessor.Unmarshal(byteReturns, reply)
+					if err != nil {
+						Err = ConvertError(err)
+						log.SError("returns data cannot be Unmarshal,callSeq is ", callSeq," error is ",err.Error())
+					}
 				}
 			}
 
+			ReleaseRpcRequest(req)
 			v := client.RemovePending(callSeq)
 			if v == nil {
 				log.SError("rpcClient cannot find seq ",callSeq, " in pending")
-				ReleaseRpcRequest(req)
+
 				return
 			}
 			if len(Err) == 0 {
@@ -326,7 +326,6 @@ func (server *Server) selfNodeRpcHandlerGo(processor IRpcProcessor, client *Clie
 				v.Err = Err
 			}
 			v.done <- v
-			ReleaseRpcRequest(req)
 		}
 	}