franklin/agent/rabbit.go

package main

import (
	"encoding/json"
	"git.wanbits.io/joe/franklin/comp"
	"git.wanbits.io/joe/kettle/log"
	"git.wanbits.io/joe/kettle/utl"
	"git.wanbits.io/joe/nnet"
	"go.uber.org/zap"
)

type WsRespPacket struct {
	Cmd    string
	Ec     int
	Result map[string]interface{}
}

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

type WsReqPacket struct {
	Cmd    string
	Params map[string]interface{}
}

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

type RabbitCProtocol struct{}

func (self *RabbitCProtocol) 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
	}

	p := &WsRespPacket{}
	err = json.Unmarshal(buf[:n], p)
	if err != nil {
		return nil, err
	}

	return p, nil
}

func (self *RabbitCProtocol) Serialize(packet nnet.IPacket) ([]byte, error) {
	p, ok := packet.(*WsReqPacket)
	if !ok {
		return nil, utl.ErrInterfaceTransform
	}

	data, err := json.Marshal(p)
	if err != nil {
		return nil, err
	}

	return data, nil
}

func (self *RabbitCProtocol) OnClosed(ses nnet.ISession, reason int32) {
}

func (self *RabbitCProtocol) OnConnected(ses nnet.ISession) (bool, int32) {
	return true, 0
}

func (self *RabbitCProtocol) OnMessage(ses nnet.ISession, pkt nnet.IPacket) bool {
	p, ok := pkt.(*WsRespPacket)
	if !ok {
		log.Error("not a WsPacket")
		return false
	}

	// dispatch
	cb, ok := rabbit_route_table[p.Cmd]
	if !ok {
		log.Error("not supported cmd", zap.String("cmd", p.Cmd))
		return false
	}

	cb(ses, p)

	return true
}

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