package comm import ( "context" "fmt" "go_dreamfactory/pb" "go_dreamfactory/lego/base" "go_dreamfactory/lego/sys/log" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" ) /* 用户会话对象 跨服操作用户的对象 */ func NewUserSession(service base.IRPCXService, ip, sessionId, gatewayServiceId string, uid string) IUserSession { return &UserSession{ IP: ip, SessionId: sessionId, GatewayServiceId: gatewayServiceId, UserId: uid, msgqueue: make([]*pb.UserMessage, 0), service: service, } } func NewUserSessionByPools(service base.IRPCXService) IUserSession { return &UserSession{ msgqueue: make([]*pb.UserMessage, 0), service: service, } } type UserSession struct { IP string SessionId string GatewayServiceId string //用户所在网关服务 UserId string service base.IRPCXService msgqueue []*pb.UserMessage } //重置 func (this *UserSession) SetSession(ip, sessionId, gatewayServiceId, uid string) { this.IP = ip this.SessionId = sessionId this.GatewayServiceId = gatewayServiceId this.UserId = uid this.msgqueue = this.msgqueue[:0] } //重置 func (this *UserSession) Reset() { this.IP = "" this.SessionId = "" this.GatewayServiceId = "" this.UserId = "" this.msgqueue = this.msgqueue[:0] } //获取用户的会话id func (this *UserSession) GetSessionId() string { return this.SessionId } //获取用户的uid func (this *UserSession) GetUserId() string { return this.UserId } //获取用户的远程ip地址 func (this *UserSession) GetIP() string { return this.IP } //用户当先所在网关服务 func (this *UserSession) GetGatewayServiceId() string { return this.GatewayServiceId } //是否登录 func (this *UserSession) IsLogin() bool { return this.UserId != "" } ///绑定uid 登录后操作 ///uid 用户id ///wokerId 用户绑定worker服务id func (this *UserSession) Bind(uid string, wokerId string) (err error) { reply := &pb.RPCMessageReply{} if err = this.service.RpcCall(context.Background(), fmt.Sprintf("%s/%s", Service_Gateway, this.GatewayServiceId), string(Rpc_GatewayAgentBind), &pb.AgentBuildReq{ UserSessionId: this.SessionId, UserId: uid, WorkerId: wokerId, }, reply); err != nil { log.Errorf("Bind UserSession:%s UserId:%s err:%v", this.SessionId, this.UserId, err) return } this.UserId = uid return } //解绑uid 注销和切换账号是处理 func (this *UserSession) UnBind() (err error) { reply := &pb.RPCMessageReply{} if err = this.service.RpcCall(context.Background(), fmt.Sprintf("%s/%s", Service_Gateway, this.GatewayServiceId), string(Rpc_GatewayAgentUnBind), &pb.AgentUnBuildReq{ UserSessionId: this.SessionId, }, reply); err != nil { log.Errorf("UnBuild UserSession:%s UserId:%s err:%v", this.SessionId, this.UserId, err) return } this.UserId = "" return } //向用户发送消息 func (this *UserSession) SendMsg(mainType, subType string, msg proto.Message) (err error) { // log.Debugf("SendMsg to UserId:[%s] Data: %v", this.UserId, msg) data, _ := anypb.New(msg) this.msgqueue = append(this.msgqueue, &pb.UserMessage{ MainType: mainType, SubType: subType, Data: data, }) return } //关闭用户连接对象 func (this *UserSession) Close() (err error) { reply := &pb.RPCMessageReply{} if err := this.service.RpcCall(context.Background(), fmt.Sprintf("%s/%s", Service_Gateway, this.GatewayServiceId), string(Rpc_GatewayAgentSendMsg), &pb.AgentCloseeReq{ UserSessionId: this.SessionId, }, reply); err != nil { log.Errorf("Close UserSession:%s UserId:%s err:%v", this.SessionId, this.UserId, err) } return } //清空消息队列 func (this *UserSession) Polls() []*pb.UserMessage { msgs := this.msgqueue this.msgqueue = this.msgqueue[:0] return msgs } func (this *UserSession) Push() (err error) { reply := &pb.RPCMessageReply{} if _, err = this.service.RpcGo(context.Background(), fmt.Sprintf("%s/%s", Service_Gateway, this.GatewayServiceId), string(Rpc_GatewayAgentSendMsg), &pb.AgentSendMessageReq{ UserSessionId: this.SessionId, Reply: this.msgqueue, }, reply); err != nil { log.Errorf("SendMsgToUsers:%v err:%v", this, err) } return } //打印日志需要 func (this *UserSession) ToString() string { return fmt.Sprintf("SessionId:%s UserId:%s GatewayServiceId:%s", this.SessionId, this.UserId, this.GatewayServiceId) }