main.go

package main

import (
	"context"
	"flag"
	"fmt"
	"gorm.io/gorm/schema"
	"net"
	"net/url"
	"strconv"
	"time"

	"github.com/go-redis/redis/v8"
	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"

	"hilo-userCenter/common"
	appConfig "hilo-userCenter/common/config"
	"hilo-userCenter/common/dingding"
	"hilo-userCenter/common/mylogrus"
	"hilo-userCenter/manager"
	"hilo-userCenter/protocol/biz"
	"hilo-userCenter/protocol/userCenter"
)

const (
	port          = 50040
	redis_section = 1
)

// 控制异步消息协程
const (
	monitorLength     = 3500 // 队列告警数量
	kickChanSize      = 500
	broadcastChanSize = 3500
	areacastChanSize  = 3500
	levelcastChanSize = 3500
)

var (
	kickChan      chan KickChanMsg
	broadcastChan chan BroadcastChanMsg
	areacastChan  chan AreaChanMsg
	levelcastChan chan LevelChanMsg
)

type KickChanMsg struct {
	userId    uint64
	proxyAddr string
}

type BroadcastChanMsg struct {
	ProxyAddr string
	UserIds   []uint64
	in        *userCenter.BroadcastMessage
}

type AreaChanMsg struct {
	ProxyAddr string
	UserIds   []uint64
	in        *userCenter.AreaMessage
}

type LevelChanMsg struct {
	ProxyAddr string
	UserIds   []uint64
	in        *userCenter.LevelMessage
}

var (
	userManager *manager.UserManager     = nil
	termManager *manager.TerminalManager = nil
	roomManager *manager.RoomManager     = nil
)

type server struct {
	userCenter.UnimplementedUserServer
}

var kasp = keepalive.ServerParameters{
	MaxConnectionIdle:     15 * time.Second, // If a client is idle for 15 seconds, send a GOAWAY
	MaxConnectionAge:      30 * time.Second, // If any connection is alive for more than 30 seconds, send a GOAWAY
	MaxConnectionAgeGrace: 5 * time.Second,  // Allow 5 seconds for pending RPCs to complete before forcibly closing connections
	Time:                  5 * time.Second,  // Ping the client if it is idle for 5 seconds to ensure the connection is still active
	Timeout:               1 * time.Second,  // Wait 1 second for the ping ack before assuming the connection is dead
}

func (s *server) Login(ctx context.Context, in *userCenter.LoginMessage) (*userCenter.LoginMessageRsp, error) {
	mylogrus.MyLog.Infof("Received loginMsg: %s, from proxy %s, client %s\n", in.Token, in.ProxyAddr, in.ClientAddr)

	var loginStatus uint32 = common.Login_success
	claim, err := common.ParseToken(in.GetToken())
	if err != nil {
		mylogrus.MyLog.Errorf("Invalid token %s\n", in.GetToken())
		loginStatus = common.Login_valid_token
	} else if time.Now().Unix() > claim.ExpiresAt {
		loginStatus = common.Login_token_expired
	}

	if loginStatus != common.Login_success || claim == nil {
		return &userCenter.LoginMessageRsp{Status: loginStatus, Uid: 0}, nil
	} else {
		// FIXME: 发现用户已经登录，要踢走旧连接
		proxyAddr := userManager.GetUser(claim.UserId)
		if proxyAddr != nil {
			mylogrus.MyLog.Infof("%d has existing value %s", claim.UserId, *proxyAddr)
			kickChan <- KickChanMsg{
				userId:    claim.UserId,
				proxyAddr: *proxyAddr,
			}
			//clientAddr := termManager.GetTerminal(claim.UserId)
			//if clientAddr == nil {
			//	mylogrus.MyLog.Infof("No terminal found for %d", claim.UserId)
			//} else {
			//	client := manager.UserProxyMgr.GetClient(*proxyAddr)
			//	if client == nil {
			//		mylogrus.MyLog.Infof("No userProxy found for %d, %s\n", claim.UserId, *proxyAddr)
			//	} else {
			//		toRouterClient := userCenter.NewRouterClient(client)
			//		msg := &userCenter.KickMessage{Uid: claim.UserId, Addr: *clientAddr}
			//		go sendKickMessage(toRouterClient, msg)
			//	}
			//}
		} else {
			mylogrus.MyLog.Errorf("wrong user %d", claim.UserId)
		}
		mylogrus.MyLog.Infof("adding user %d", claim.UserId)

		// save to redis
		userManager.AddUser(claim.UserId, in.ProxyAddr)
		termManager.SetTerminal(claim.UserId, in.ClientAddr)

		// 为登录用户建立反向连接，如果需要的话
		go manager.UserProxyMgr.MakeClient(in.ProxyAddr)
	}
	return &userCenter.LoginMessageRsp{Status: loginStatus, Uid: claim.UserId}, nil
}

func (s *server) Logout(ctx context.Context, in *userCenter.LogoutMessage) (*userCenter.LogoutMessageRsp, error) {
	//mylogrus.MyLog.Infof("Received logoutMsg: %s, %d\n", in.GetClientAddr(), in.GetUid())

	addr := termManager.GetTerminal(in.Uid)
	if addr != nil && *addr == in.ClientAddr {
		termManager.RemoveTerminal(in.Uid)
	}

	// 去掉队列push
	//msg := protocol.LogoutMsg{
	//	UserId:    in.Uid,
	//	Timestamp: time.Now().Unix(),
	//}
	//buf, err := json.Marshal(msg)
	//if err == nil {
	//r, err := termManager.RedisClient.RPush(context.Background(), protocol.LogoutMsgQueue, string(buf)).Result()
	//if err == nil {
	//	mylogrus.MyLog.Infof("RPush OK, length = %v ", r)
	//} else {
	//	mylogrus.MyLog.Infof("RPush failed %v", err)
	//}
	//}

	return &userCenter.LogoutMessageRsp{Status: 0}, nil
}

func (s *server) Multicast(ctx context.Context, in *userCenter.MulticastMessage) (*userCenter.MulticastMessageRsp, error) {
	//mylogrus.MyLog.Infof("Multicasting msgType = %d to %v, size = %d\n", in.MsgType, in.Uids, len(in.PayLoad))

	failed := []uint64{}
	for _, uid := range in.Uids {
		ok := false
		addr := userManager.GetUser(uid)
		if addr == nil {
			mylogrus.MyLog.Errorf("Unknown user %d\n", uid)
		} else {
			client := manager.UserProxyMgr.MakeClient(*addr)
			if client == nil {
				mylogrus.MyLog.Infof("Failed in making client for %d, %s\n", uid, *addr)
			} else {
				toRouterClient := userCenter.NewRouterClient(client)
				status, err := routeMessage(toRouterClient, uid, in.MsgType, in.PayLoad)
				if err == nil && status == common.ROUTE_SUCCESS {
					ok = true
				}
			}
		}
		if !ok {
			failed = append(failed, uid)
		}
	}
	if len(failed) > 0 {
		mylogrus.MyLog.Infof("Multicast failed for %v\n", failed)
	}
	return &userCenter.MulticastMessageRsp{FailedUids: failed}, nil
}

func (s *server) Broadcast(ctx context.Context, in *userCenter.BroadcastMessage) (*userCenter.BroadcastMessageRsp, error) {
	//mylogrus.MyLog.Infof("Broadcasting msgType = %d, size = %d\n", in.MsgType, len(in.PayLoad))

	failed := []uint64{}
	terminals := termManager.GetAll()
	if terminals != nil {
		m := make(map[string][]uint64, 0)
		for u, _ := range *terminals {
			uid, err := strconv.ParseUint(u, 10, 64)
			ok := false
			if err == nil {
				addr := userManager.GetUser(uid)
				if addr != nil {
					if _, ok := m[*addr]; !ok {
						m[*addr] = make([]uint64, 0)
					}
					m[*addr] = append(m[*addr], uid)
					ok = true
				} else {
					mylogrus.MyLog.Errorf("Unknown user %d\n", uid)
				}
			} else {
				mylogrus.MyLog.Infof("Invalid user str: %s\n", u)
			}
			if !ok {
				failed = append(failed, uid)
			}
		}
		for addr, users := range m {
			//addr = strings.Replace(addr, "47.91.121.73", "172.26.95.24", -1)
			//mylogrus.MyLog.Infof("Broadcasting: Addr %s: %d users", addr, len(users))
			//if !strings.Contains(addr, "172.26.95.48:50050") && !strings.Contains(addr, "172.26.95.24:50050") {
			//	mylogrus.MyLog.Errorf("Broadcasting: Addr error %s: %d users", addr, len(users))
			//}

			const sendBatchSize = 5
			for i := 0; i < len(users); i += sendBatchSize {
				end := i + sendBatchSize
				if end > len(users) {
					end = len(users)
				}
				broadcastChan <- BroadcastChanMsg{
					ProxyAddr: addr,
					UserIds:   users[i:end],
					in:        in,
				}
				//go realBroadcast(addr, users[i:end], in)
			}
		}
	}
	return &userCenter.BroadcastMessageRsp{FailedUids: failed}, nil
}

func (s *server) Areacast(ctx context.Context, in *userCenter.AreaMessage) (*userCenter.AreaMessageRsp, error) {
	var failed []uint64
	terminals := termManager.GetAll()
	if terminals != nil {
		var uids []uint64
		for uidStr := range *terminals {
			if uid, _ := strconv.ParseUint(uidStr, 10, 64); uid > 0 {
				uids = append(uids, uid)
			}
		}
		// 处理分区用户
		areaUids := userManager.GetAreaUsers(uids, int8(in.Area))
		if len(areaUids) <= 0 {
			return &userCenter.AreaMessageRsp{FailedUids: failed}, nil
		}
		m := make(map[string][]uint64, 0)
		for uid := range areaUids {
			ok := false
			addr := userManager.GetUser(uid)
			if addr != nil {
				if _, ok := m[*addr]; !ok {
					m[*addr] = make([]uint64, 0)
				}
				m[*addr] = append(m[*addr], uid)
				ok = true
			} else {
				mylogrus.MyLog.Errorf("Unknown user %d\n", uid)
			}
			if !ok {
				failed = append(failed, uid)
			}
		}
		for addr, users := range m {
			const sendBatchSize = 5
			for i := 0; i < len(users); i += sendBatchSize {
				end := i + sendBatchSize
				if end > len(users) {
					end = len(users)
				}
				areacastChan <- AreaChanMsg{
					ProxyAddr: addr,
					UserIds:   users[i:end],
					in:        in,
				}
			}
		}
	}
	return &userCenter.AreaMessageRsp{FailedUids: failed}, nil
}

func (s *server) Levelcast(ctx context.Context, in *userCenter.LevelMessage) (*userCenter.LevelMessageRsp, error) {
	var failed []uint64
	terminals := termManager.GetAll()
	if terminals != nil {
		var uids []uint64
		for uidStr := range *terminals {
			if uid, _ := strconv.ParseUint(uidStr, 10, 64); uid > 0 {
				uids = append(uids, uid)
			}
		}
		// 处理等级用户
		levelUserIds, userIds := userManager.GetLevelUsers(uids, in.Level)
		if len(levelUserIds) <= 0 {
			return &userCenter.LevelMessageRsp{FailedUids: failed}, nil
		}
		if in.Area > 0 {
			// 处理分区用户
			levelUserIds = userManager.GetAreaUsers(userIds, int8(in.Area))
			if len(levelUserIds) <= 0 {
				return &userCenter.LevelMessageRsp{FailedUids: failed}, nil
			}
		}
		m := make(map[string][]uint64, 0)
		for uid := range levelUserIds {
			ok := false
			addr := userManager.GetUser(uid)
			if addr != nil {
				if _, ok := m[*addr]; !ok {
					m[*addr] = make([]uint64, 0)
				}
				m[*addr] = append(m[*addr], uid)
				ok = true
			} else {
				mylogrus.MyLog.Errorf("Unknown user %d\n", uid)
			}
			if !ok {
				failed = append(failed, uid)
			}
		}
		for addr, users := range m {
			const sendBatchSize = 5
			for i := 0; i < len(users); i += sendBatchSize {
				end := i + sendBatchSize
				if end > len(users) {
					end = len(users)
				}
				levelcastChan <- LevelChanMsg{
					ProxyAddr: addr,
					UserIds:   users[i:end],
					in:        in,
				}
			}
		}
	}
	return &userCenter.LevelMessageRsp{FailedUids: failed}, nil
}

func (s *server) Transmit(ctx context.Context, in *userCenter.BizMessage) (*userCenter.BizMessageRsp, error) {
	mylogrus.MyLog.Infof("Transmiting msgType = %d, uid = %d, payLoad: %s\n", in.MsgType, in.Uid, in.PayLoad)

	// fixme:
	addr := "localhost:50060"

	rsp := &userCenter.BizMessageRsp{}
	client := manager.BizMgr.MakeClient(addr)
	if client == nil {
		mylogrus.MyLog.Errorf("Failed in making client for %d, %s\n", in.Uid, addr)
	} else {
		transmitterClient := biz.NewTransmitterClient(client)
		status, err := transmitMessage(transmitterClient, in.MsgType, in.PayLoad)
		mylogrus.MyLog.Infof("transmit uid = %d, msgType = %d, status = %d, %v", in.Uid, in.MsgType, status, err)
	}

	return rsp, nil
}

func (s *server) EnterRoom(ctx context.Context, in *userCenter.EnterRoomMessage) (*userCenter.EnterRoomMessageRsp, error) {
	if err := roomManager.AddRoomUser(in.GetUid(), in.GetGroupId()); err != nil {
		return nil, err
	}
	return &userCenter.EnterRoomMessageRsp{
		Status: 0,
	}, nil
}

func (s *server) LeaveRoom(ctx context.Context, in *userCenter.LeaveRoomMessage) (*userCenter.LeaveRoomMessageRsp, error) {
	if err := roomManager.DelRoomUser(in.GetUid(), in.GetGroupId()); err != nil {
		return nil, err
	}
	return &userCenter.LeaveRoomMessageRsp{
		Status: 0,
	}, nil
}

func (s *server) RoomHeartbeat(ctx context.Context, in *userCenter.RoomHeartbeatMessage) (*userCenter.RoomHeartbeatMessageRsp, error) {
	if err := roomManager.UpdateRoomUser(in.GetUid(), in.GetGroupId()); err != nil {
		return nil, err
	}
	return &userCenter.RoomHeartbeatMessageRsp{
		Status: 0,
	}, nil
}

func (s *server) GetLastRoomHeartbeat(ctx context.Context, in *userCenter.GetLastRoomHeartbeatMessage) (*userCenter.GetLastRoomHeartbeatMessageResp, error) {
	ts, err := roomManager.GetLastRoomUserHeartbeat(in.GetUid(), in.GetGroupId())
	if err != nil {
		return nil, err
	}
	return &userCenter.GetLastRoomHeartbeatMessageResp{
		Timestamp: ts,
	}, nil
}

func realBroadcast(addr string, uids []uint64, msg *userCenter.BroadcastMessage) {
	//mylogrus.MyLog.Infof("Broadcasting: Addr %s: users: %v", addr, uids)

	for _, uid := range uids {
		client := manager.UserProxyMgr.MakeClient(addr)
		if client == nil {
			mylogrus.MyLog.Errorf("Failed in making client for %d, %s\n", uid, addr)
		} else {

			toRouterClient := userCenter.NewRouterClient(client)
			status, err := routeMessage(toRouterClient, uid, msg.MsgType, msg.PayLoad)
			if err != nil {
				mylogrus.MyLog.Errorf("routeMessage uid = %d, msgType = %d, status = %d, %v", uid, msg.MsgType, status, err)
			}
		}
	}
}

func realAreacast(addr string, uids []uint64, msg *userCenter.AreaMessage) {
	for _, uid := range uids {
		client := manager.UserProxyMgr.MakeClient(addr)
		if client == nil {
			mylogrus.MyLog.Errorf("Failed in making client for %d, %s\n", uid, addr)
		} else {

			toRouterClient := userCenter.NewRouterClient(client)
			status, err := routeMessage(toRouterClient, uid, msg.MsgType, msg.PayLoad)
			if err != nil {
				mylogrus.MyLog.Errorf("routeMessage uid = %d, msgType = %d, status = %d, %v", uid, msg.MsgType, status, err)
			}
		}
	}
}

func realLevelcast(addr string, uids []uint64, msg *userCenter.LevelMessage) {
	for _, uid := range uids {
		client := manager.UserProxyMgr.MakeClient(addr)
		if client == nil {
			mylogrus.MyLog.Errorf("Failed in making client for %d, %s\n", uid, addr)
		} else {

			toRouterClient := userCenter.NewRouterClient(client)
			status, err := routeMessage(toRouterClient, uid, msg.MsgType, msg.PayLoad)
			if err != nil {
				mylogrus.MyLog.Errorf("routeMessage uid = %d, msgType = %d, status = %d, %v", uid, msg.MsgType, status, err)
			}
		}
	}
}

func routeMessage(c userCenter.RouterClient, uid uint64, msgType uint32, data []byte) (uint32, error) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	r, err := c.Route(ctx, &userCenter.RouteMessage{
		Uid:     uid,
		MsgType: msgType,
		PayLoad: data,
	})
	if err != nil {
		mylogrus.MyLog.Errorf("Route message to user %d, err: %s\n", uid, err.Error())
	} else if r != nil {
		//mylogrus.MyLog.Infof("Route message to user %d, status = %d", uid, r.Status)
		return r.Status, err
	}
	return 0, err
}

func transmitMessage(c biz.TransmitterClient, msgType uint32, data string) (uint32, error) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	r, err := c.Process(ctx, &biz.BizMessage{
		Type:    msgType,
		PayLoad: data,
	})
	if err != nil {
		mylogrus.MyLog.Errorf("Transmit message type %d, err: %s\n", msgType, err.Error())
	} else if r != nil {
		mylogrus.MyLog.Infof("Transmit message type %d, status = %d", msgType, r.Status)
		return r.Status, err
	}
	return 0, err
}

func sendKickMessage(c userCenter.RouterClient, msg *userCenter.KickMessage) error {
	mylogrus.MyLog.Infof("sendKickMessage %s", msg.String())
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	r, err := c.KickUser(ctx, msg)
	if err != nil && r != nil {
		mylogrus.MyLog.Errorf("sendKickMessage message status = %d", r.Status)
	}
	return err
}

const (
	RegisterName = "userCenter"
)

func RegisterToRedis(RedisClusterClient *redis.Client, port int, init bool) {
	// 本地不注册
	if appConfig.AppIsLocal() {
		return
	}
	if RedisClusterClient == nil {
		failMsg := fmt.Sprintf("RegisterToRedis fail,redisClusterNotInit,serviceName:%v", RegisterName)
		_ = dingding.SendDingRobot(dingding.ROBOTWEBHOOK, failMsg, true)
		mylogrus.MyLog.Errorf(failMsg)
		return
	}
	redisKey := "service:" + RegisterName
	ip, err := common.GetClientIpV2()
	if err != nil {
		failMsg := fmt.Sprintf("RegisterToRedis fail,ip fail,err:%v,serviceName:%v", err, RegisterName)
		mylogrus.MyLog.Errorf(failMsg)
		_ = dingding.SendDingRobot(dingding.ROBOTWEBHOOK, failMsg, true)
		return
	}
	ipPort := fmt.Sprintf("%s:%d", ip, port)
	if err := RedisClusterClient.ZAdd(context.Background(), redisKey, &redis.Z{
		Score:  float64(time.Now().Unix()),
		Member: ipPort,
	}).Err(); err != nil {
		failMsg := fmt.Sprintf("RegisterToRedis fail,redis fail,err:%v,serviceName:%v", err, RegisterName)
		mylogrus.MyLog.Errorf(failMsg)
		_ = dingding.SendDingRobot(dingding.ROBOTWEBHOOK, failMsg, true)
	}
	// 初始化注册自我检查 selfCheck
	if init {
		go func() {
			ticker := time.NewTicker(time.Second * 5)
			defer ticker.Stop()
			for {
				select {
				case <-ticker.C:
					RegisterToRedis(RedisClusterClient, port, false) // 刷新注册
				}
			}
		}()
	}
}

type HiloConfigs struct {
	Name  string `gorm:"primary_key"`
	Value string
}

func main() {
	flag.Parse()

	// init redis cluster
	rdbCluster := redis.NewClient(&redis.Options{
		Addr:     appConfig.GetConfigRedis().REDIS_CLUSTER_HOST,
		Password: appConfig.GetConfigRedis().REDIS_CLUSTER_PASSWORD,
	})
	// 注册到redis
	RegisterToRedis(rdbCluster, port, true)

	// init redis
	rdb := redis.NewClient(&redis.Options{
		Addr:     appConfig.GetConfigRedis().REDIS_HOST,
		Password: appConfig.GetConfigRedis().REDIS_PASSWORD,
		DB:       redis_section,
	})
	if rdb == nil {
		mylogrus.MyLog.Fatalf("failed to connect redis %s\n", appConfig.GetConfigRedis().REDIS_HOST)
	}
	result, err := rdb.Ping(context.Background()).Result()
	if err != nil {
		mylogrus.MyLog.Fatal(err)
	} else if result != "PONG" {
		mylogrus.MyLog.Fatalf("Invalid ping response %s", result)
	}

	// init db
	options := "?charset=utf8mb4&parseTime=True&loc=Local&time_zone=" + url.QueryEscape("'+8:00'")
	fmt.Println("options = ", options)

	dsn := appConfig.GetConfigMysql().MYSQL_USERNAME + ":" + appConfig.GetConfigMysql().MYSQL_PASSWORD + "@(" + appConfig.GetConfigMysql().MYSQL_HOST + ")/" + appConfig.GetConfigMysql().MYSQL_DB
	db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{
		Logger: logger.Default.LogMode(logger.Info),
		NamingStrategy: schema.NamingStrategy{
			SingularTable: true,
		},
	})
	if err != nil {
		mylogrus.MyLog.Fatal("mysql connect error %v", err)
	} else {
		mylogrus.MyLog.Infof("mysql connect success")
	}

	var jwtConfig = HiloConfigs{}
	db.First(&jwtConfig, "name = 'jwt_secret'")
	if len(jwtConfig.Value) == 0 {
		mylogrus.MyLog.Fatalln("Empty jwt secret")
	}
	mylogrus.MyLog.Infof("jwt secret is %s", jwtConfig.Value)
	common.SetJWTSecret(jwtConfig.Value)

	userManager = &manager.UserManager{
		Ctx:         context.Background(),
		RedisClient: rdb,
		MysqlDB:     db,
	}
	go func() {
		for {
			// 同步区域
			userManager.SyncArea()
			time.Sleep(time.Minute * 15)
		}
	}()
	termManager = &manager.TerminalManager{
		Ctx:         context.Background(),
		RedisClient: rdb,
	}
	roomManager = &manager.RoomManager{
		Ctx:         context.Background(),
		RedisClient: rdb,
	}

	go func() {
		ticker := time.NewTicker(time.Second * 30)
		defer ticker.Stop()

		for {
			select {
			case a := <-ticker.C:
				mylogrus.MyLog.Infof("Tick at %s", a.String())
				terminals := termManager.GetAll()
				if terminals != nil {
					if len(*terminals) <= 100 {
						mylogrus.MyLog.Infof("%v", *terminals)
					}
					if len(*terminals) >= 2 {
						mylogrus.MyLog.Infof("%d on-line users found", len(*terminals))
					}
				}
			}
		}
	}()

	// 初始化协程chan
	kickChan = make(chan KickChanMsg, kickChanSize)
	broadcastChan = make(chan BroadcastChanMsg, broadcastChanSize)
	areacastChan = make(chan AreaChanMsg, areacastChanSize)
	levelcastChan = make(chan LevelChanMsg, levelcastChanSize)
	go check() // 检查长度
	for i := 0; i < kickChanSize; i++ {
		go func(n int) {
			kick(n)
		}(i)
	}
	for i := 0; i < broadcastChanSize; i++ {
		go func(n int) {
			broadcast(n)
		}(i)
	}
	for i := 0; i < areacastChanSize; i++ {
		go func(n int) {
			areacast(n)
		}(i)
	}
	for i := 0; i < levelcastChanSize; i++ {
		go func(n int) {
			levelcast(n)
		}(i)
	}

	fmt.Println("Go RPC listening on ", port)
	lis, err := net.Listen("tcp4", ":"+strconv.Itoa(port))
	if err != nil {
		mylogrus.MyLog.Fatalf("failed to listen: %v", err)
	}
	s := grpc.NewServer(grpc.KeepaliveParams(kasp))
	userCenter.RegisterUserServer(s, &server{})
	if err := s.Serve(lis); err != nil {
		mylogrus.MyLog.Fatalf("failed to serve: %v", err)
	}
}

func kick(n int) {
	for msg := range kickChan {
		mylogrus.MyLog.Infof("handling kick in:%d,msg:%+v", n, msg)
		clientAddr := termManager.GetTerminal(msg.userId)
		if clientAddr == nil {
			mylogrus.MyLog.Errorf("No terminal found for %d", msg.userId)
		} else {
			client := manager.UserProxyMgr.GetClient(msg.proxyAddr)
			if client == nil {
				mylogrus.MyLog.Errorf("No userProxy found for %d, %s\n", msg.userId, msg.proxyAddr)
			} else {
				toRouterClient := userCenter.NewRouterClient(client)
				in := &userCenter.KickMessage{Uid: msg.userId, Addr: *clientAddr}
				if err := sendKickMessage(toRouterClient, in); err != nil {
					mylogrus.MyLog.Errorf("sendKickMessage fail,uid:%v,proxyAddr:%v,clientAddr:%v,err:%v",
						msg.userId, msg.proxyAddr, clientAddr, err)
				}
			}
		}
	}
}

func broadcast(n int) {
	for msg := range broadcastChan {
		//mylogrus.MyLog.Infof("handling broadcast in:%d,msg:%+v", n, msg)
		realBroadcast(msg.ProxyAddr, msg.UserIds, msg.in) // fixme: 这里还有优化空间,广播能否在proxy层做批量
	}
}

func areacast(n int) {
	for msg := range areacastChan {
		realAreacast(msg.ProxyAddr, msg.UserIds, msg.in)
	}
}

func levelcast(n int) {
	for msg := range levelcastChan {
		realLevelcast(msg.ProxyAddr, msg.UserIds, msg.in)
	}
}

var lastDingTime time.Time
var dingIntervalMin float64 = 5 // 5min 告警间隔

func check() {
	lastDingTime = time.Now()
	tick := time.NewTicker(time.Second * 3)
	defer tick.Stop()
	for {
		select {
		case <-tick.C:
			l, l2, l3, l4 := len(kickChan), len(broadcastChan), len(areacastChan), len(levelcastChan)
			if l >= monitorLength || l2 >= monitorLength || l3 >= monitorLength || l4 >= monitorLength {
				if time.Now().Sub(lastDingTime).Minutes() > dingIntervalMin {
					go func() {
						if sErr := dingding.SendDingRobot(dingding.ROBOTWEBHOOK, fmt.Sprintf("userCenter通知延迟,队列长度:kickChan:%d,broadcastChan:%d,areacastChan:%d,levelcastChan:%d",
							l, l2, l3, l4), true); sErr != nil {
							mylogrus.MyLog.Errorf("dingding msg fail:%v", sErr)
						} else {
							lastDingTime = time.Now()
						}
					}()
				}
			}
			if l > 0 || l2 > 0 {
				mylogrus.MyLog.Infof("userCenter msg,left kick:%v,broadcast:%v", l, l2)
			}
		}
	}
}