go_dreamfactory/modules/gateway/module.go

package gateway

import (
	"context"
	"fmt"
	"go_dreamfactory/comm"
	"go_dreamfactory/pb"
	"go_dreamfactory/sys/db"

	"go_dreamfactory/lego/base"
	"go_dreamfactory/lego/core"
	"go_dreamfactory/lego/core/cbase"
	"go_dreamfactory/lego/sys/log"
	"go_dreamfactory/lego/sys/rpcx"
)

/*
模块名:gateway
描述:提供客户端网关路由服务 管理用户socket对象 以及分发用户消息
开发:李伟
*/

func NewModule() core.IModule {
	m := new(Gateway)
	return m
}

type Gateway struct {
	cbase.ModuleBase
	options   *Options
	service   base.IRPCXService // rpcx服务接口 主要client->server
	wsService *WSServiceComp    // websocket服务 监听websocket连接
	agentMgr  *AgentMgrComp     // 客户端websocket连接管理
	queueComp *QueueComp        // 等待队列组件
	configure *configureComp
}

// GetType 获取模块服务类型
func (this *Gateway) GetType() core.M_Modules {
	return comm.ModuleGate
}

// NewOptions 模块自定义参数
func (this *Gateway) NewOptions() (options core.IModuleOptions) {
	return new(Options)
}

// Service 获取rpcx服务接口
func (this *Gateway) Service() base.IRPCXService {
	return this.service
}

// Init 模块初始化函数
func (this *Gateway) Init(service core.IService, module core.IModule, options core.IModuleOptions) (err error) {
	err = this.ModuleBase.Init(service, module, options)
	this.options = options.(*Options)
	this.service = service.(base.IRPCXService)
	return
}

// Start 模块启动函数 注册rpc服务接口提供用户相关的rpc接口服务
func (this *Gateway) Start() (err error) {

	_name2Func := map[string]any{
		// 注册用户绑定uid接口 登录成功后触发
		// string(comm.Rpc_GatewayAgentBind): this.agentMgr.Bind,
		// 注册用户解绑uid接口 登出或则切换账号是触发
		string(comm.Rpc_GatewayAgentUnBind): this.agentMgr.UnBind,
		// 向用户发送消息接口
		string(comm.Rpc_GatewayAgentSendMsg): this.agentMgr.SendMsgToAgent,
		// 向多个用户对象发送消息接口
		string(comm.Rpc_GatewaySendBatchMsg): this.agentMgr.SendMsgToAgents,
		// 向所有用户发送消息接口
		string(comm.Rpc_GatewaySendRadioMsg): this.agentMgr.SendMsgToAllAgent,
		// 关闭用户socket连接接口
		string(comm.Rpc_GatewayAgentClose): this.agentMgr.CloseAgent,
	}
	for name, fn := range _name2Func {
		this.service.RegisterFunctionName(name, fn)
	}
	err = this.ModuleBase.Start()
	//没有建立客户端 主动发起握手
	this.service.RpcGo(context.Background(), comm.Service_Mainte, rpcx.RpcX_ShakeHands, &rpcx.ServiceNode{
		ServiceTag:  this.service.GetTag(),
		ServiceId:   this.service.GetId(),
		ServiceType: this.service.GetType(),
		ServiceAddr: fmt.Sprintf("tcp@%s:%d", this.service.GetIp(), this.service.GetPort())},
		nil)
	return
}

// OnInstallComp 装备组件
func (this *Gateway) OnInstallComp() {
	this.ModuleBase.OnInstallComp()
	this.agentMgr = this.RegisterComp(new(AgentMgrComp)).(*AgentMgrComp)
	this.queueComp = this.RegisterComp(new(QueueComp)).(*QueueComp)
	this.wsService = this.RegisterComp(new(WSServiceComp)).(*WSServiceComp)
	this.configure = this.RegisterComp(new(configureComp)).(*configureComp)
}

// Connect 有新的连接对象进入
func (this *Gateway) Connect(a IAgent) {
	log.Debugf("[Module.Gateway] have new connect:Ip[%s] SessionId:[%s]", a.IP(), a.SessionId())
	this.agentMgr.Connect(a)
}

// DisConnect 有用户断开连接
func (this *Gateway) DisConnect(a IAgent) {
	log.Debugf("[Module.Gateway] have disConnect:Ip[%s] SessionId:[%s] uid:[%s]", a.IP(), a.SessionId(), a.UserId())
	this.agentMgr.DisConnect(a)
}

func (this *Gateway) LoginNotice(a IAgent) {
	if _, err := this.service.AcrossClusterRpcGo(context.Background(), db.CrossTag(), comm.Service_Worker, string(comm.Rpc_GatewayNoticeUserLogin), &pb.NoticeUserLoginReq{
		Ip:               a.IP(),
		ServiceTag:       this.service.GetTag(),
		GatewayServiceId: this.service.GetId(),
		UserSessionId:    a.SessionId(),
		UserId:           a.UserId(),
	}, nil); err != nil {
		log.Errorf("uId:%s clusterTag:%s Rpc_NoticeUserLogin  err:%v", a.UserId(), db.CrossTag(), err)
	}
}

// GetMsgDistribute 读取消息分发规则
func (this *Gateway) GetMsgDistribute(msgmid, msguid string) (rule string, ok bool) {
	return this.configure.GetMsgDistribute(msgmid, msguid)
}

// 加入登录等待队列
func (this *Gateway) InLoginQueue(sessionId string, login *pb.UserMessage) (index int32, err error) {
	return this.queueComp.inLoginQueue(sessionId, login)
}

//日志
func (this *Gateway) Enabled(lvl log.Loglevel) bool {
	return this.options.GetLog().Enabled(lvl)
}
func (this *Gateway) SetName(name string) {
	this.options.GetLog().SetName(name)
}

//日志接口
func (this *Gateway) Debug(msg string, args ...log.Field) {
	this.options.GetLog().Debug(msg, args...)
}
func (this *Gateway) Info(msg string, args ...log.Field) {
	this.options.GetLog().Info(msg, args...)
}
func (this *Gateway) Print(msg string, args ...log.Field) {
	this.options.GetLog().Print(msg, args...)
}
func (this *Gateway) Warn(msg string, args ...log.Field) {
	this.options.GetLog().Warn(msg, args...)
}
func (this *Gateway) Error(msg string, args ...log.Field) {
	this.options.GetLog().Error(msg, args...)
}
func (this *Gateway) Panic(msg string, args ...log.Field) {
	this.options.GetLog().Panic(msg, args...)
}
func (this *Gateway) Fatal(msg string, args ...log.Field) {
	this.options.GetLog().Fatal(msg, args...)
}

func (this *Gateway) Debugf(format string, args ...interface{}) {
	this.options.GetLog().Debugf(format, args...)
}
func (this *Gateway) Infof(format string, args ...interface{}) {
	this.options.GetLog().Infof(format, args...)
}
func (this *Gateway) Printf(format string, args ...interface{}) {
	this.options.GetLog().Printf(format, args...)
}
func (this *Gateway) Warnf(format string, args ...interface{}) {
	this.options.GetLog().Warnf(format, args...)
}
func (this *Gateway) Errorf(format string, args ...interface{}) {
	this.options.GetLog().Errorf(format, args...)
}
func (this *Gateway) Fatalf(format string, args ...interface{}) {
	this.options.GetLog().Fatalf(format, args...)
}
func (this *Gateway) Panicf(format string, args ...interface{}) {
	this.options.GetLog().Panicf(format, args...)
}

func (this *Gateway) Debugln(args ...interface{}) {
	this.options.GetLog().Debugln(args...)
}
func (this *Gateway) Infoln(args ...interface{}) {
	this.options.GetLog().Infoln(args...)
}
func (this *Gateway) Println(args ...interface{}) {
	this.options.GetLog().Println(args...)
}
func (this *Gateway) Warnln(args ...interface{}) {
	this.options.GetLog().Warnln(args...)
}
func (this *Gateway) Errorln(args ...interface{}) {
	this.options.GetLog().Errorln(args...)
}
func (this *Gateway) Fatalln(args ...interface{}) {
	this.options.GetLog().Fatalln(args...)
}
func (this *Gateway) Panicln(args ...interface{}) {
	this.options.GetLog().Panicln(args...)
}