franklin/agent/server_router.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"git.wanbits.io/joe/franklin/comp"
	"git.wanbits.io/joe/franklin/mod"
	pb "git.wanbits.io/joe/franklin/protos"
	"git.wanbits.io/joe/kettle/log"
	"git.wanbits.io/joe/nnet"
	"github.com/go-redis/redis"
	"github.com/golang/protobuf/proto"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"time"
)

var (
	routes = map[string]fnCSHandler{
		"C_Heartbeat":  on_C_Heartbeat,
		"C_Login":      on_C_Login,
		"C_Logout":     on_C_Logout,
		"C_EnterGame":  on_C_EnterGame,
		"C_LeaveGame":  on_C_LeaveGame,
		"C_EnterRoom":  on_C_EnterRoom,
		"C_EnterTable": on_C_EnterTable,
		"C_LeaveRoom":  on_C_LeaveRoom,
		"C_LeaveTable": on_C_LeaveTable,
	}
)

type fnCSHandler func(nnet.ISession, *CSPacket)

func dispatch(ses nnet.ISession, pkt *CSPacket) {
	handler, ok := routes[pkt.MsgName]
	if !ok {
		// route to lobbies
		if pkt.MsgId <= 5000 {
			ForwardLobby(pkt)
			return
		}
		logicId, err:= mod.LoadUserLogic(pkt.UserId)
		if err != nil {
			if err != redis.Nil {
				log.Error("user logic", zap.Error(err))
			}
			return
		}
		ForwardLogic(logicId, pkt)
		return
	}
	handler(ses, pkt)
}

func on_C_Heartbeat(ses nnet.ISession, p *CSPacket) {
	resp := &pb.S_Heartbeat{
		Ts: time.Now().Unix(),
	}
	SendToClient(ses, ses.Id(), pb.ErrCode_OK, resp)
}

func on_C_Login(ses nnet.ISession, p *CSPacket) {
	prm := &pb.C_Login{}
	ses.SetId(p.UserId) 	// IMPORTANT
	err := proto.Unmarshal(p.Req, prm)
	if err != nil {
		response_c_login(ses, pb.ErrCode_BAD_FORMAT, nil)
		return
	}
	// verify by rabbit
	seq := time.Now().UnixNano()
	err = g_rabbitProto.Send(NewReqPacket("bl_login_verify", seq, &pb.VerifyParams{
		UserId:   p.UserId,
		Username: prm.Username,
		Token:    prm.Token,
	}), ses)
	if err != nil {
		response_c_login(ses, pb.ErrCode_FAILED, nil)
	}
}

func response_c_login(ses nnet.ISession, errCode pb.ErrCode, res *pb.S_Login) {
	var ec = pb.ErrCode_OK
	var reason = pb.NetCloseReason_KICK

	defer func() {
		p := []int32{}
		if ec != pb.ErrCode_OK {
			p = append(p, 1)
			p = append(p, int32(reason))
			res = &pb.S_Login{}
			//SendToClient(ses, ses.Id(), ec, &pb.S_Login{})
			//KillUserSession(ses, ses.Id(), pb.NetCloseReason_ERR_AUTH)
		}
		SendToClient(ses, ses.Id(), ec, res, p...)
	}()

	if errCode != pb.ErrCode_OK {
		ec = errCode
		reason = pb.NetCloseReason_ERR_AUTH
		return
	}
	// check agent offline
	if g_offline {
		ec = pb.ErrCode_MAINTAINING
		reason = pb.NetCloseReason_KICK
		return
	}
	// check user kicked off
	if mod.LoadKick(ses.Id()) {
		ec = pb.ErrCode_FORBIDDEN
		reason = pb.NetCloseReason_KICK
		return
	}
	//TODO: check if have logged in any agents already
	doCheckUserLoginAlready(ses.Id())
	// check reconnect state
	logicId, err := mod.LoadOffline(ses.Id())
	if err == nil {
		// route to logic
		SendToLogic(logicId, ses.Id(), &pb.Q_Online{UserId: ses.Id()})
	} else {
		if err != redis.Nil {
			log.Error("", zap.Error(err))
		}
		// route to a lobby
		SendToLobby(ses.Id(), &pb.Q_Online{UserId: ses.Id()})
	}
	// update user location
	err = mod.SaveUserLogin(ses.Id(), g_conf.Id)
	if err != nil {
		log.Error("", zap.Error(err))
	}
	// update session id
	ses.UpdateId(ses.Id())
}

func on_C_Logout(ses nnet.ISession, p *CSPacket) {
	//
	if err := mod.SaveUserLogout(ses.Id()); err != nil {
		log.Error("logout", zap.Error(err))
	}
}

func on_C_EnterGame(ses nnet.ISession, p *CSPacket) {

}

func on_C_LeaveGame(ses nnet.ISession, p *CSPacket) {

}

func on_C_EnterRoom(ses nnet.ISession, p *CSPacket) {

}

func on_C_EnterTable(ses nnet.ISession, p *CSPacket) {

}

func on_C_LeaveRoom(ses nnet.ISession, p *CSPacket) {

}

func on_C_LeaveTable(ses nnet.ISession, p *CSPacket) {

}

// check if specified user had logged in any agents, if its, request older login to logout
// and keep this login process going on
func doCheckUserLoginAlready(userId uint64) {
	agentId, err := mod.LoadUserAgent(userId)
	if err != nil {
		return
	}
	if agentId == g_conf.Id {
		// logged in self
		ses, err := g_server.GetSession(userId)
		if err != nil {
			log.Error("logic error", zap.Uint64("userId", userId), zap.Uint64("agentId", agentId))
			return
		}
		KillUserSession(ses, userId, pb.NetCloseReason_DUP)
		return
	}
	// agentId != g_conf.Id
	key := fmt.Sprintf("%s/%v", comp.PathAgents(g_conf), g_conf.Id)
	resp, err := comp.GEtcdc.Get(key)
	if err != nil {
		log.Error("", zap.Error(err))
		return
	}
	if len(resp.Kvs) <= 0 {
		log.Error("")
		return
	}
	agentConf := &pb.AppConfConf{}
	err = json.Unmarshal(resp.Kvs[0].Value, agentConf)
	if err != nil {
		return
	}

	// grpc call
	conn, err := grpc.Dial(agentConf.RpcAddr, grpc.WithInsecure(), grpc.WithBlock())
	if err != nil {
		log.Error("", zap.Error(err))
		return
	}
	defer conn.Close()
	client := pb.NewRpcAgentClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), 3 * time.Second)
	defer cancel()

	_, err = client.Kick(ctx, &pb.KickUser{
		UserId: userId,
		Reason: int32(pb.NetCloseReason_DUP),
	})
	if err != nil {
		log.Error("", zap.Error(err))
	}
}