From 50bb5a92a8aace0c8c18ded9e924bc6acff0d6ab Mon Sep 17 00:00:00 2001 From: liwei1dao Date: Tue, 28 Jun 2022 17:09:32 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E4=BC=9A=E8=AF=9D=E5=AF=B9?= =?UTF-8?q?=E8=B1=A1=E6=B6=88=E6=81=AF=E9=98=9F=E5=88=97=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- comm/core.go | 2 +- comm/usersession.go | 2 ++ modules/comp_gate.go | 11 ++------ .../{comp_configure.go => compconfigure.go} | 0 modules/equipment/api_getlist.go | 15 +++++------ modules/gateway/agent.go | 16 +++++++---- pb/comm.pb.go | 12 ++++----- pb/proto/comm.proto | 2 +- services/comp_gateroute.go | 27 +++++++++---------- 9 files changed, 42 insertions(+), 45 deletions(-) rename modules/{comp_configure.go => compconfigure.go} (100%) diff --git a/comm/core.go b/comm/core.go index 45f6a3849..bc07dafa7 100644 --- a/comm/core.go +++ b/comm/core.go @@ -16,7 +16,7 @@ import ( type ISC_GateRouteComp interface { core.IServiceComp ReceiveMsg(ctx context.Context, args *pb.AgentMessage, reply *pb.RPCMessageReply) error - RegisterRoute(methodName string, comp reflect.Value, msg reflect.Type, reply reflect.Type, check, handle reflect.Method) + RegisterRoute(methodName string, comp reflect.Value, msg reflect.Type, check, handle reflect.Method) } //游戏类资源类型 diff --git a/comm/usersession.go b/comm/usersession.go index ba5624adc..d8ff86e01 100644 --- a/comm/usersession.go +++ b/comm/usersession.go @@ -22,6 +22,7 @@ func NewUserSession(service base.IRPCXService, ip, sessionId, gatewayServiceId s SessionId: sessionId, GatewayServiceId: gatewayServiceId, UserId: uid, + msgqueue: make([]*pb.UserMessage, 0), service: service, } } @@ -32,6 +33,7 @@ type UserSession struct { GatewayServiceId string //用户所在网关服务 UserId string service base.IRPCXService + msgqueue []*pb.UserMessage } //获取用户的会话id diff --git a/modules/comp_gate.go b/modules/comp_gate.go index 883fdf33d..99c9fc565 100644 --- a/modules/comp_gate.go +++ b/modules/comp_gate.go @@ -77,7 +77,7 @@ func (this *MCompGate) reflectionRouteHandle(typ reflect.Type, method reflect.Me if method.PkgPath != "" { return } - if mtype.NumIn() != 5 { + if mtype.NumIn() != 4 { return } ctxType := mtype.In(1) @@ -95,13 +95,6 @@ func (this *MCompGate) reflectionRouteHandle(typ reflect.Type, method reflect.Me if !this.isExportedOrBuiltinType(agrType) { return } - replyType := mtype.In(4) - if replyType.Kind() != reflect.Ptr { - return - } - if !this.isExportedOrBuiltinType(replyType) { - return - } if mtype.NumOut() != 1 { return @@ -118,7 +111,7 @@ func (this *MCompGate) reflectionRouteHandle(typ reflect.Type, method reflect.Me return } if err := this.reflectionRouteCheck(check, agrType); err == nil { - this.scomp.RegisterRoute(fmt.Sprintf("%s.%s", this.module.GetType(), strings.ToLower(mname)), reflect.ValueOf(this.comp), agrType, replyType, check, method) + this.scomp.RegisterRoute(fmt.Sprintf("%s.%s", this.module.GetType(), strings.ToLower(mname)), reflect.ValueOf(this.comp), agrType, check, method) } else { log.Panicf("反射注册用户处理函数错误 [%s-%s]校验函数格式异常:%v", this.module.GetType(), mname, err) return diff --git a/modules/comp_configure.go b/modules/compconfigure.go similarity index 100% rename from modules/comp_configure.go rename to modules/compconfigure.go diff --git a/modules/equipment/api_getlist.go b/modules/equipment/api_getlist.go index 08496c491..85fee866b 100644 --- a/modules/equipment/api_getlist.go +++ b/modules/equipment/api_getlist.go @@ -13,21 +13,20 @@ func (this *apiComp) Getlist_Check(session comm.IUserSession, req *pb.Equipment_ } ///获取用户装备列表 -func (this *apiComp) Getlist(session comm.IUserSession, agrs map[string]interface{}, req *pb.Equipment_GetList_Req, resp *pb.Equipment_GetList_Resp) (code pb.ErrorCode) { +func (this *apiComp) Getlist(session comm.IUserSession, agrs map[string]interface{}, req *pb.Equipment_GetList_Req) (code pb.ErrorCode) { var ( err error items []*pb.DB_Equipment ) - // defer func() { - // if code == pb.ErrorCode_Success { - // session.SendMsg(string(this.module.GetType()), "", &pb.Equipment_GetList_Resp{Equipments: items}) - // } - // }() - if items, err = this.module.modelEquipment.QueryUserEquipments(session.GetUserId()); err != nil { + defer func() { + if code == pb.ErrorCode_Success { + session.SendMsg(string(this.module.GetType()), "", &pb.Equipment_GetList_Resp{Equipments: items}) + } + }() + if items, err = this.module.modelequipment.QueryUserEquipments(session.GetUserId()); err != nil { log.Errorf("QueryUserPackReq err:%v", err) code = pb.ErrorCode_CacheReadError return } - resp.Equipments = items return } diff --git a/modules/gateway/agent.go b/modules/gateway/agent.go index 854a6fd24..cc47f0a47 100644 --- a/modules/gateway/agent.go +++ b/modules/gateway/agent.go @@ -208,7 +208,7 @@ func (this *Agent) Close() { } //分发用户消息 -func (this *Agent) messageDistribution(msg *pb.UserMessage) error { +func (this *Agent) messageDistribution(msg *pb.UserMessage) (err error) { reply := &pb.RPCMessageReply{} log.Debugf("agent:%s uId:%s MessageDistribution msg:%s.%s", this.sessionId, this.uId, msg.MainType, msg.SubType) servicePath := comm.Service_Worker @@ -219,7 +219,7 @@ func (this *Agent) messageDistribution(msg *pb.UserMessage) error { servicePath = fmt.Sprintf("%s/%s", comm.Service_Worker, this.wId) } } - if err := this.gateway.Service().RpcCall(context.Background(), servicePath, string(comm.Rpc_GatewayRoute), &pb.AgentMessage{ + if err = this.gateway.Service().RpcCall(context.Background(), servicePath, string(comm.Rpc_GatewayRoute), &pb.AgentMessage{ Ip: this.IP(), UserSessionId: this.sessionId, UserId: this.uId, @@ -229,16 +229,22 @@ func (this *Agent) messageDistribution(msg *pb.UserMessage) error { Message: msg.Data, }, reply); err != nil { log.Errorf("agent:%s uId:%s MessageDistribution err:%v", this.sessionId, this.uId, err) - return err + return } if reply.Code != pb.ErrorCode_Success { data, _ := anypb.New(&pb.ErrorNotify{ReqMainType: msg.MainType, ReqSubType: msg.SubType, Code: pb.ErrorCode(reply.Code.Number())}) - err := this.WriteMsg(&pb.UserMessage{ + err = this.WriteMsg(&pb.UserMessage{ MainType: comm.MainType_Notify, SubType: comm.SubType_ErrorNotify, Data: data, }) - return err + return + } else { + for _, v := range reply.Reply { + if err = this.WriteMsg(v); err != nil { + return + } + } } return nil } diff --git a/pb/comm.pb.go b/pb/comm.pb.go index 88b90b6a8..a21273297 100644 --- a/pb/comm.pb.go +++ b/pb/comm.pb.go @@ -252,10 +252,10 @@ type RPCMessageReply struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Code ErrorCode `protobuf:"varint,1,opt,name=Code,proto3,enum=ErrorCode" json:"Code"` - ErrorMessage string `protobuf:"bytes,2,opt,name=ErrorMessage,proto3" json:"ErrorMessage"` - ErrorData string `protobuf:"bytes,3,opt,name=ErrorData,proto3" json:"ErrorData"` - Reply *UserMessage `protobuf:"bytes,4,opt,name=Reply,proto3" json:"Reply"` + Code ErrorCode `protobuf:"varint,1,opt,name=Code,proto3,enum=ErrorCode" json:"Code"` + ErrorMessage string `protobuf:"bytes,2,opt,name=ErrorMessage,proto3" json:"ErrorMessage"` + ErrorData string `protobuf:"bytes,3,opt,name=ErrorData,proto3" json:"ErrorData"` + Reply []*UserMessage `protobuf:"bytes,4,rep,name=Reply,proto3" json:"Reply"` } func (x *RPCMessageReply) Reset() { @@ -311,7 +311,7 @@ func (x *RPCMessageReply) GetErrorData() string { return "" } -func (x *RPCMessageReply) GetReply() *UserMessage { +func (x *RPCMessageReply) GetReply() []*UserMessage { if x != nil { return x.Reply } @@ -771,7 +771,7 @@ var file_comm_proto_rawDesc = []byte{ 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x44, 0x61, 0x74, 0x61, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x44, 0x61, 0x74, 0x61, - 0x12, 0x22, 0x0a, 0x05, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, + 0x12, 0x22, 0x0a, 0x05, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0c, 0x2e, 0x55, 0x73, 0x65, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x05, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x22, 0x69, 0x0a, 0x0d, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x42, 0x75, 0x69, 0x6c, 0x64, 0x52, 0x65, 0x71, 0x12, 0x24, 0x0a, 0x0d, 0x55, 0x73, 0x65, 0x72, 0x53, 0x65, 0x73, diff --git a/pb/proto/comm.proto b/pb/proto/comm.proto index 2e32a2ee0..69696ae52 100644 --- a/pb/proto/comm.proto +++ b/pb/proto/comm.proto @@ -28,7 +28,7 @@ message RPCMessageReply { ErrorCode Code = 1; string ErrorMessage = 2; string ErrorData = 3; - UserMessage Reply = 4; + repeated UserMessage Reply = 4; } //用户代理绑定Uid请求 diff --git a/services/comp_gateroute.go b/services/comp_gateroute.go index 6a761e498..566991e05 100644 --- a/services/comp_gateroute.go +++ b/services/comp_gateroute.go @@ -33,11 +33,10 @@ func NewGateRouteComp() comm.ISC_GateRouteComp { //用户协议处理函数注册的反射对象 type msghandle struct { - rcvr reflect.Value - msgType reflect.Type //消息请求类型 - replyType reflect.Type //消息回应类型 - check reflect.Method //校验函数 - handle reflect.Method //处理函数 + rcvr reflect.Value + msgType reflect.Type //消息请求类型 + check reflect.Method //校验函数 + handle reflect.Method //处理函数 } //服务网关组件 @@ -70,7 +69,7 @@ func (this *SCompGateRoute) Start() (err error) { } //业务模块注册用户消息处理路由 -func (this *SCompGateRoute) RegisterRoute(methodName string, comp reflect.Value, msg reflect.Type, replyType reflect.Type, check, handele reflect.Method) { +func (this *SCompGateRoute) RegisterRoute(methodName string, comp reflect.Value, msg reflect.Type, check, handele reflect.Method) { log.Debugf("注册用户路由【%s】", methodName) this.mrlock.RLock() _, ok := this.msghandles[methodName] @@ -81,11 +80,10 @@ func (this *SCompGateRoute) RegisterRoute(methodName string, comp reflect.Value, } this.mrlock.Lock() this.msghandles[methodName] = &msghandle{ - rcvr: comp, - msgType: msg, - replyType: replyType, - check: check, - handle: handele, + rcvr: comp, + msgType: msg, + check: check, + handle: handele, } this.mrlock.Unlock() } @@ -117,9 +115,6 @@ func (this *SCompGateRoute) ReceiveMsg(ctx context.Context, args *pb.AgentMessag return err } - //序列化用户消息对象 - replymsg := reflect.New(msghandle.replyType.Elem()).Interface() - //调用校验接口 checkreturn := msghandle.check.Func.Call([]reflect.Value{msghandle.rcvr, reflect.ValueOf(session), reflect.ValueOf(msg)}) //读取校验结果 有错误直接返回错误码给用户 @@ -141,7 +136,7 @@ func (this *SCompGateRoute) ReceiveMsg(ctx context.Context, args *pb.AgentMessag //校验结果成功 处理临时数据转移 result := checkreturn[0].Interface().(map[string]interface{}) //调用用户处理函数 - handlereturn := msghandle.handle.Func.Call([]reflect.Value{msghandle.rcvr, reflect.ValueOf(session), reflect.ValueOf(result), reflect.ValueOf(msg), reflect.ValueOf(replymsg)}) + handlereturn := msghandle.handle.Func.Call([]reflect.Value{msghandle.rcvr, reflect.ValueOf(session), reflect.ValueOf(result), reflect.ValueOf(msg)}) errcode := pb.ErrorCode(handlereturn[0].Int()) if errcode != pb.ErrorCode_Success { //处理返货错误码 返回用户错误信息 log.Errorf("HandleUserMsg:%s msg:%v code:%d", method, msg, code) @@ -149,6 +144,7 @@ func (this *SCompGateRoute) ReceiveMsg(ctx context.Context, args *pb.AgentMessag reply.ErrorMessage = pb.GetErrorCodeMsg(errcode) return nil } + } else { //未找到消息处理函数 reply.Code = pb.ErrorCode_ReqParameterError } @@ -158,5 +154,6 @@ func (this *SCompGateRoute) ReceiveMsg(ctx context.Context, args *pb.AgentMessag //RPC_NoticeUserClose 接收用户离线通知 func (this *SCompGateRoute) NoticeUserClose(ctx context.Context, args *pb.NoticeUserCloseReq, reply *pb.RPCMessageReply) error { event.TriggerEvent(comm.Event_UserOffline, args.UserId) + return nil }