franklin/agent/logics.go

package main

import (
	"encoding/binary"
	"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"
)

// protocol && callback, Using TCP
type ALProtocol struct{}

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

	n, err := conn.Read(buf[:4])
	n, err = conn.Read(buf[4 : 4+n])
	p := &SCPacket{}
	err = proto.Unmarshal(buf[4:], p.resp)
	return p, err
}

func (self *ALProtocol) Serialize(packet nnet.IPacket) ([]byte, error) {
	csp, ok := packet.(*CSPacket)
	if !ok {
		log.Error("can not transform")
		return nil, utl.ErrInterfaceTransform
	}
	data, err := proto.Marshal(csp.Request)
	if err != nil {
		return nil, err
	}
	buf := make([]byte, len(data)+4)
	binary.LittleEndian.PutUint32(buf, uint32(len(data)))
	copy(buf[4:], data)
	return buf, nil
}

func (self *ALProtocol) OnClosed(ses nnet.ISession, reason int32) {
	log.Info("logic offline:", zap.String("addr", ses.ServerAddr()), zap.Int32("reason", reason))
}

func (self *ALProtocol) OnConnected(ses nnet.ISession) (bool, int32) {
	log.Debug("connect loggic", zap.String("addr", ses.ServerAddr()))
	return true, 0
}

func (self *ALProtocol) OnMessage(ses nnet.ISession, pkt nnet.IPacket) bool {
	csp, ok := pkt.(*SCPacket)
	if !ok {
		return false
	}
	switch pb.MsgGossipId(csp.resp.MsgId) {
	case pb.MsgGossipId_A_Heartbeat:
	case pb.MsgGossipId_A_Offline:
	case pb.MsgGossipId_A_Online:
	default:
		ForwardClient(csp)
	}
	return true
}

// server-side dont handle heartbeat.
func (self *ALProtocol) OnHeartbeat(ses nnet.ISession) bool {
	SendToLogic(ses.Id(), 0, &pb.Q_Heartbeat{Ts: time.Now().Unix()})
	return true
}

func ForwardLogic(logicId uint64, msg *CSPacket) {
	ses, err := g_logics.GetSession(logicId)
	if err != nil {
		log.Error("cant find logic server", zap.Uint64("logicId", logicId), zap.Error(err))
		return
	}
	err = ses.AWrite(msg, 3*time.Second)
	if err != nil {
		log.Error("post logic", zap.Error(err))
	}
}

func SendToLogic(logicId, userId uint64, inner proto.Message) {
	ses, err := g_logics.GetSession(logicId)
	if err != nil {
		log.Error("cant find logic server", zap.Uint64("logicId", logicId), zap.Error(err))
		return
	}

	// 不做 guid 合法性检查，而直接发出消息
	bst, err := proto.Marshal(inner)
	if nil != err {
		log.Error("proto.Marshal", zap.Error(err))
		return
	}
	name := comp.GetMsgName(inner)
	id := comp.GetMsgId(name)
	req := &pb.Request{
		MsgId:   id,
		UserId:  userId,
		MsgName: name,
		Req:     bst,
	}

	err = ses.AWrite(&CSPacket{req}, time.Second)
	if nil != err {
		log.Error("proto.AWrite", zap.Error(err))
	}
}