Ma/origin
1
0
Fork 0
mirror of https://github.com/duanhf2012/origin.git synced 2026-02-04 06:54:45 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

扩展pbprocessor参数

Browse Source
This commit is contained in:
orgin
2022-02-17 15:50:39 +08:00
parent db964f90e1
commit b91f09b9ff
2 changed files with 43 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@@ -1,6 +1,6 @@
module github.com/duanhf2012/origin module github.com/duanhf2012/origin
go 1.15 go 1.17
require ( require (
github.com/go-sql-driver/mysql v1.5.0 github.com/go-sql-driver/mysql v1.5.0

86
network/processor/pbprocessor.go
View File

@@ -13,29 +13,30 @@ type MessageInfo struct {
msgHandler MessageHandler msgHandler MessageHandler
} }
type MessageHandler func(clientId uint64,msg proto.Message) type MessageHandler func(clientId uint64, msg proto.Message)
type ConnectHandler func(clientId uint64) type ConnectHandler func(clientId uint64)
type UnknownMessageHandler func(clientId uint64,msg []byte) type UnknownMessageHandler func(clientId uint64, msg []byte)
const MsgTypeSize = 2 const MsgTypeSize = 2
type PBProcessor struct { type PBProcessor struct {
mapMsg map[uint16]MessageInfo mapMsg map[uint16]MessageInfo
LittleEndian bool LittleEndian bool
unknownMessageHandler UnknownMessageHandler unknownMessageHandler UnknownMessageHandler
connectHandler ConnectHandler connectHandler ConnectHandler
disconnectHandler ConnectHandler disconnectHandler ConnectHandler
network.INetMempool network.INetMempool
} }
type PBPackInfo struct { type PBPackInfo struct {
typ uint16 typ uint16
msg proto.Message msg proto.Message
rawMsg []byte rawMsg []byte
} }
func NewPBProcessor() *PBProcessor { func NewPBProcessor() *PBProcessor {
processor := &PBProcessor{mapMsg:map[uint16]MessageInfo{}} processor := &PBProcessor{mapMsg: map[uint16]MessageInfo{}}
processor.INetMempool = network.NewMemAreaPool() processor.INetMempool = network.NewMemAreaPool()
return processor return processor
} }
@@ -53,65 +54,65 @@ func (slf *PBPackInfo) GetMsg() proto.Message {
} }
// must goroutine safe // must goroutine safe
func (pbProcessor *PBProcessor ) MsgRoute(msg interface{},userdata interface{}) error{ func (pbProcessor *PBProcessor) MsgRoute(clientId uint64, msg interface{}) error {
pPackInfo := msg.(*PBPackInfo) pPackInfo := msg.(*PBPackInfo)
v,ok := pbProcessor.mapMsg[pPackInfo.typ] v, ok := pbProcessor.mapMsg[pPackInfo.typ]
if ok == false { if ok == false {
return fmt.Errorf("Cannot find msgtype %d is register!",pPackInfo.typ) return fmt.Errorf("Cannot find msgtype %d is register!", pPackInfo.typ)
} }
v.msgHandler(userdata.(uint64),pPackInfo.msg) v.msgHandler(clientId, pPackInfo.msg)
return nil return nil
} }
// must goroutine safe // must goroutine safe
func (pbProcessor *PBProcessor ) Unmarshal(data []byte) (interface{}, error) { func (pbProcessor *PBProcessor) Unmarshal(clientId uint64, data []byte) (interface{}, error) {
defer pbProcessor.ReleaseByteSlice(data) defer pbProcessor.ReleaseByteSlice(data)
var msgType uint16 var msgType uint16
if pbProcessor.LittleEndian == true { if pbProcessor.LittleEndian == true {
msgType = binary.LittleEndian.Uint16(data[:2]) msgType = binary.LittleEndian.Uint16(data[:2])
}else{ } else {
msgType = binary.BigEndian.Uint16(data[:2]) msgType = binary.BigEndian.Uint16(data[:2])
} }
info,ok := pbProcessor.mapMsg[msgType] info, ok := pbProcessor.mapMsg[msgType]
if ok == false { if ok == false {
return nil,fmt.Errorf("cannot find register %d msgtype!",msgType) return nil, fmt.Errorf("cannot find register %d msgtype!", msgType)
} }
msg := reflect.New(info.msgType.Elem()).Interface() msg := reflect.New(info.msgType.Elem()).Interface()
protoMsg := msg.(proto.Message) protoMsg := msg.(proto.Message)
err := proto.Unmarshal(data[2:], protoMsg) err := proto.Unmarshal(data[2:], protoMsg)
if err != nil { if err != nil {
return nil,err return nil, err
} }
return &PBPackInfo{typ:msgType,msg:protoMsg},nil return &PBPackInfo{typ: msgType, msg: protoMsg}, nil
} }
// must goroutine safe // must goroutine safe
func (pbProcessor *PBProcessor ) Marshal(msg interface{}) ([]byte, error){ func (pbProcessor *PBProcessor) Marshal(clientId uint64, msg interface{}) ([]byte, error) {
pMsg := msg.(*PBPackInfo) pMsg := msg.(*PBPackInfo)
var err error var err error
if pMsg.msg!=nil { if pMsg.msg != nil {
pMsg.rawMsg,err = proto.Marshal(pMsg.msg) pMsg.rawMsg, err = proto.Marshal(pMsg.msg)
if err != nil { if err != nil {
return nil,err return nil, err
} }
} }
buff := make([]byte, 2, len(pMsg.rawMsg)+MsgTypeSize) buff := make([]byte, 2, len(pMsg.rawMsg)+MsgTypeSize)
if pbProcessor.LittleEndian == true { if pbProcessor.LittleEndian == true {
binary.LittleEndian.PutUint16(buff[:2],pMsg.typ) binary.LittleEndian.PutUint16(buff[:2], pMsg.typ)
}else{ } else {
binary.BigEndian.PutUint16(buff[:2],pMsg.typ) binary.BigEndian.PutUint16(buff[:2], pMsg.typ)
} }
buff = append(buff,pMsg.rawMsg...) buff = append(buff, pMsg.rawMsg...)
return buff,nil return buff, nil
} }
func (pbProcessor *PBProcessor) Register(msgtype uint16,msg proto.Message,handle MessageHandler) { func (pbProcessor *PBProcessor) Register(msgtype uint16, msg proto.Message, handle MessageHandler) {
var info MessageInfo var info MessageInfo
info.msgType = reflect.TypeOf(msg.(proto.Message)) info.msgType = reflect.TypeOf(msg.(proto.Message))
@@ -119,38 +120,35 @@ func (pbProcessor *PBProcessor) Register(msgtype uint16,msg proto.Message,handle
pbProcessor.mapMsg[msgtype] = info pbProcessor.mapMsg[msgtype] = info
} }
func (pbProcessor *PBProcessor) MakeMsg(msgType uint16,protoMsg proto.Message) *PBPackInfo { func (pbProcessor *PBProcessor) MakeMsg(msgType uint16, protoMsg proto.Message) *PBPackInfo {
return &PBPackInfo{typ:msgType,msg:protoMsg} return &PBPackInfo{typ: msgType, msg: protoMsg}
} }
func (pbProcessor *PBProcessor) MakeRawMsg(msgType uint16,msg []byte) *PBPackInfo { func (pbProcessor *PBProcessor) MakeRawMsg(msgType uint16, msg []byte) *PBPackInfo {
return &PBPackInfo{typ:msgType,rawMsg:msg} return &PBPackInfo{typ: msgType, rawMsg: msg}
} }
func (pbProcessor *PBProcessor) UnknownMsgRoute(msg interface{}, userData interface{}){ func (pbProcessor *PBProcessor) UnknownMsgRoute(clientId uint64, msg interface{}) {
pbProcessor.unknownMessageHandler(userData.(uint64),msg.([]byte)) pbProcessor.unknownMessageHandler(clientId, msg.([]byte))
} }
// connect event // connect event
func (pbProcessor *PBProcessor) ConnectedRoute(userData interface{}){ func (pbProcessor *PBProcessor) ConnectedRoute(clientId uint64) {
pbProcessor.connectHandler(userData.(uint64)) pbProcessor.connectHandler(clientId)
} }
func (pbProcessor *PBProcessor) DisConnectedRoute(userData interface{}){ func (pbProcessor *PBProcessor) DisConnectedRoute(clientId uint64) {
pbProcessor.disconnectHandler(userData.(uint64)) pbProcessor.disconnectHandler(clientId)
} }
func (pbProcessor *PBProcessor) RegisterUnknownMsg(unknownMessageHandler UnknownMessageHandler){ func (pbProcessor *PBProcessor) RegisterUnknownMsg(unknownMessageHandler UnknownMessageHandler) {
pbProcessor.unknownMessageHandler = unknownMessageHandler pbProcessor.unknownMessageHandler = unknownMessageHandler
} }
func (pbProcessor *PBProcessor) RegisterConnected(connectHandler ConnectHandler){ func (pbProcessor *PBProcessor) RegisterConnected(connectHandler ConnectHandler) {
pbProcessor.connectHandler = connectHandler pbProcessor.connectHandler = connectHandler
} }
func (pbProcessor *PBProcessor) RegisterDisConnected(disconnectHandler ConnectHandler){ func (pbProcessor *PBProcessor) RegisterDisConnected(disconnectHandler ConnectHandler) {
pbProcessor.disconnectHandler = disconnectHandler pbProcessor.disconnectHandler = disconnectHandler
} }