franklin/agent/server.go

package main

import (
	"git.wanbits.io/joe/franklin/comp"
	pb "git.wanbits.io/joe/franklin/protos"
	"git.wanbits.io/joe/kettle/log"
	"git.wanbits.io/joe/kettle/utl"
	"git.wanbits.io/joe/nnet"
	"github.com/golang/protobuf/proto"
	"go.uber.org/zap"
	"time"
)

// C/S packet
type CSPacket struct {
	*pb.Request
}

func (self *CSPacket) ShouldClose() (bool, int32) {
	return false, 0
}

type SCPacket struct {
	resp    *pb.Response
	closing int32
	reason  int32
}

func NewSCPacket(resp *pb.Response, params ...int32) *SCPacket {
	p := &SCPacket{
		resp: resp,
	}
	if len(params) > 0 {
		p.closing = params[0]
	}
	if len(params) > 1 {
		p.reason = params[1]
	}
	return p
}

func (self *SCPacket) ShouldClose() (bool, int32) {
	close_session := false
	if self.closing > 0 {
		close_session = true
	}
	return close_session, self.reason
}

// protocol && callback
type CSProtocol struct {
}

func (self *CSProtocol) ReadPacket(conn nnet.IConn) (nnet.IPacket, error) {
	buf := comp.GBufPool.Get()
	defer comp.GBufPool.Put(buf)

	n, err := conn.Read(buf)
	if err != nil {
		return nil, err
	}
	var req *pb.Request
	err = proto.Unmarshal(buf[:n], req)
	if err != nil {
		return nil, err
	}
	p := &CSPacket{
		req,
	}
	return p, nil
}

func (self *CSProtocol) Serialize(packet nnet.IPacket) ([]byte, error) {
	p, ok := packet.(*SCPacket)
	if !ok {
		return nil, utl.ErrInterfaceTransform
	}
	return proto.Marshal(p.resp)
}

func (self *CSProtocol) OnClosed(ses nnet.ISession, reason int32) {
	log.Debug("lost connection", zap.String("remote", ses.GetRawConn().RemoteAddr().String()),
		zap.Uint64("id", ses.Id()))
	on_C_Logout(ses, nil)
}

func (self *CSProtocol) OnConnected(ses nnet.ISession) (bool, int32) {
	log.Debug("new client", zap.String("remote", ses.GetRawConn().RemoteAddr().String()))
	return true, 0
}

func (self *CSProtocol) OnMessage(ses nnet.ISession, pkt nnet.IPacket) bool {
	p, ok := pkt.(*CSPacket)
	if !ok {
		return false
	}
	dispatch(ses, p)
	return true
}

// server-side dont handle heartbeat.
func (self *CSProtocol) OnHeartbeat(ses nnet.ISession) bool {
	return true
}

// @ses: session
// @st: Response 内部结构的 struct
// @params: [0] 发送后是否关闭 [1] 关闭原因
func SendToClient(ses nnet.ISession, userId uint64, ec pb.ErrCode, st proto.Message, params ...int32) {
	// 不做 guid 合法性检查，而直接发出消息
	bst, err := proto.Marshal(st)
	if nil != err {
		log.Error("proto.Marshal", zap.Error(err))
		return
	}
	name := comp.GetMsgName(st)
	id := comp.GetMsgId(name)
	res := &pb.Response{
		MsgId:   id,
		UserId:  userId,
		MsgName: name,
		Ec:      int32(ec),
		Res:     bst,
	}

	err = ses.AWrite(NewSCPacket(res, params...), time.Second)
	if nil != err {
		log.Error("proto.AWrite", zap.Error(err))
	}
}

func ForwardClient(pkt *SCPacket) {
	ses, err := g_server.GetSession(pkt.resp.UserId)
	if err != nil {
		log.Error("no userId", zap.Uint64("userId", pkt.resp.UserId))
		return
	}

	err = ses.AWrite(pkt, time.Second)
	if err != nil {
		log.Error("forward to client", zap.Error(err))
	}
}

// 关闭某个用户连接
func KillUserSession(ses nnet.ISession, userId uint64, reason pb.NetCloseReason) {
	cmd_offline := &pb.S_Offline{}
	cmd_offline.Reason = int32(reason)
	SendToClient(ses, userId, pb.ErrCode_OK, cmd_offline, 1, int32(reason))
}