增加消息发送

This commit is contained in:
duanhf2012
2020-03-28 12:02:22 +08:00
parent 7745ed39f3
commit d66995dd0f
3 changed files with 40 additions and 5 deletions

View File

@@ -1,5 +1,6 @@
package network
type Processor interface {
// must goroutine safe
Route(msg interface{}, userData interface{}) error

View File

@@ -12,7 +12,8 @@ type MessageInfo struct {
msgHandler MessageHandler
}
type MessageHandler func(clientid uint64,msg proto.Message)
const MsgTypeSize = 2
type PBProcessor struct {
mapMsg map[uint16]MessageInfo
LittleEndian bool
@@ -65,10 +66,22 @@ func (slf *PBProcessor ) Unmarshal(data []byte) (interface{}, error) {
// must goroutine safe
func (slf *PBProcessor ) Marshal(msg interface{}) ([]byte, error){
return proto.Marshal(msg.(proto.Message))
}
bytes,err := proto.Marshal(msg.(proto.Message))
if err != nil {
return nil,err
}
type MessageHandler func(clientid uint64,msg proto.Message)
buff := make([]byte, 0, len(bytes)+MsgTypeSize)
pMsg := msg.(*PBPackInfo)
if slf.LittleEndian == true {
binary.LittleEndian.PutUint16(buff[:2],pMsg.typ)
}else{
binary.BigEndian.PutUint16(buff[:2],pMsg.typ)
}
buff = append(buff,bytes...)
return buff,nil
}
func (slf *PBProcessor) Register(msgtype uint16,msg proto.Message,handle MessageHandler) {
var info MessageInfo
@@ -77,3 +90,7 @@ func (slf *PBProcessor) Register(msgtype uint16,msg proto.Message,handle Message
info.msgHandler = handle
slf.mapMsg[msgtype] = info
}
func (slf *PBProcessor) MakeMsg(msgType uint16,protoMsg proto.Message) *PBPackInfo {
return &PBPackInfo{typ:msgType,msg:protoMsg}
}

View File

@@ -64,6 +64,7 @@ func (slf *TcpService) OnInit() error{
if ok == true {
slf.tcpServer.MaxMsgLen = uint32(MaxMsgLen.(float64))
}
slf.mapClient = make( map[uint64] *Client,slf.tcpServer.MaxConnNum)
slf.tcpServer.NewAgent =slf.NewClient
slf.tcpServer.Start()
//加载配置
@@ -86,6 +87,7 @@ func (slf *TcpService) NewClient(conn *network.TCPConn) network.Agent {
}
pClient := &Client{tcpConn:conn}
pClient.tcpService = slf
slf.mapClient[slf.initClientId] = pClient
return pClient
}
@@ -124,7 +126,6 @@ func (slf *Client) Run() {
continue
}
slf.tcpService.GetEventReciver().NotifyEvent(&event.Event{Type:event.Sys_Event_Tcp_RecvPack,Data:&TcpPack{ClientId:slf.id,Data:data}})
}
}
@@ -135,3 +136,19 @@ func (slf *Client) OnClose(){
defer slf.tcpService.mapClientLocker.Unlock()
delete (slf.tcpService.mapClient,slf.GetId())
}
func (slf *TcpService) SendMsg(clientid uint64,msg interface{}) error{
slf.mapClientLocker.Lock()
client,ok := slf.mapClient[clientid]
if ok == false{
slf.mapClientLocker.Unlock()
return fmt.Errorf("client %d is disconnect!",clientid)
}
slf.mapClientLocker.Unlock()
bytes,err := slf.process.Marshal(msg)
if err != nil {
return err
}
return client.tcpConn.WriteMsg(bytes)
}