protobuf.go 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224
  1. package protobuf
  2. import (
  3. "encoding/binary"
  4. "errors"
  5. "fmt"
  6. "leafstalk/log"
  7. "leafstalk/module/handler"
  8. "math"
  9. "reflect"
  10. "google.golang.org/protobuf/proto"
  11. // "github.com/golang/protobuf/proto"
  12. )
  13. // -------------------------
  14. // | id | protobuf message |
  15. // -------------------------
  16. type Processor struct {
  17. littleEndian bool
  18. msgInfo []*MsgInfo
  19. msgID map[reflect.Type]uint16
  20. }
  21. type MsgInfo struct {
  22. msgType reflect.Type
  23. msgRouter *handler.Server
  24. msgHandler MsgHandler
  25. msgRawHandler MsgHandler
  26. }
  27. type MsgHandler func([]interface{})
  28. type MsgRaw struct {
  29. msgID uint16
  30. msgRawData []byte
  31. }
  32. func NewProcessor() *Processor {
  33. p := new(Processor)
  34. p.littleEndian = false
  35. p.msgID = make(map[reflect.Type]uint16)
  36. return p
  37. }
  38. // It's dangerous to call the method on routing or marshaling (unmarshaling)
  39. func (p *Processor) SetByteOrder(littleEndian bool) {
  40. p.littleEndian = littleEndian
  41. }
  42. // It's dangerous to call the method on routing or marshaling (unmarshaling)
  43. func (p *Processor) Register(msg proto.Message) uint16 {
  44. msgType := reflect.TypeOf(msg)
  45. if msgType == nil || msgType.Kind() != reflect.Ptr {
  46. log.Fatal("protobuf message pointer required")
  47. }
  48. if _, ok := p.msgID[msgType]; ok {
  49. log.Fatalf("message %s is already registered", msgType)
  50. }
  51. if len(p.msgInfo) >= math.MaxUint16 {
  52. log.Fatalf("too many protobuf messages (max = %v)", math.MaxUint16)
  53. }
  54. i := new(MsgInfo)
  55. i.msgType = msgType
  56. p.msgInfo = append(p.msgInfo, i)
  57. id := uint16(len(p.msgInfo) - 1)
  58. p.msgID[msgType] = id
  59. return id
  60. }
  61. // func (p *Processor) Register2(msg interface{}) uint16 {
  62. // msgType := reflect.TypeOf(msg)
  63. // if msgType == nil || msgType.Kind() != reflect.Ptr {
  64. // log.Fatal("protobuf message pointer required")
  65. // }
  66. // if _, ok := p.msgID[msgType]; ok {
  67. // log.Fatalf("message %s is already registered", msgType)
  68. // }
  69. // if len(p.msgInfo) >= math.MaxUint16 {
  70. // log.Fatalf("too many protobuf messages (max = %v)", math.MaxUint16)
  71. // }
  72. // i := new(MsgInfo)
  73. // i.msgType = msgType
  74. // p.msgInfo = append(p.msgInfo, i)
  75. // id := uint16(len(p.msgInfo) - 1)
  76. // p.msgID[msgType] = id
  77. // return id
  78. // }
  79. // func (p *Processor) IsRegisted(msg proto.Message) (bool, error) {
  80. // msgType := reflect.TypeOf(msg)
  81. // if msgType == nil || msgType.Kind() != reflect.Ptr {
  82. // return false, errors.New("protobuf message pointer required")
  83. // }
  84. // if len(p.msgInfo) >= math.MaxUint16 {
  85. // return false, fmt.Errorf("too many protobuf messages (max = %v)", math.MaxUint16)
  86. // }
  87. // if _, ok := p.msgID[msgType]; ok {
  88. // return true, nil
  89. // }
  90. // return false, nil
  91. // }
  92. // It's dangerous to call the method on routing or marshaling (unmarshaling)
  93. func (p *Processor) SetRouter(msg proto.Message, msgRouter *handler.Server) {
  94. msgType := reflect.TypeOf(msg)
  95. id, ok := p.msgID[msgType]
  96. if !ok {
  97. log.Fatalf("message %s not registered", msgType)
  98. }
  99. p.msgInfo[id].msgRouter = msgRouter
  100. }
  101. // It's dangerous to call the method on routing or marshaling (unmarshaling)
  102. func (p *Processor) SetHandler(msg proto.Message, msgHandler MsgHandler) {
  103. msgType := reflect.TypeOf(msg)
  104. id, ok := p.msgID[msgType]
  105. if !ok {
  106. log.Fatalf("message %s not registered", msgType)
  107. }
  108. p.msgInfo[id].msgHandler = msgHandler
  109. }
  110. // It's dangerous to call the method on routing or marshaling (unmarshaling)
  111. func (p *Processor) SetRawHandler(id uint16, msgRawHandler MsgHandler) {
  112. if id >= uint16(len(p.msgInfo)) {
  113. log.Fatalf("message id %v not registered", id)
  114. }
  115. p.msgInfo[id].msgRawHandler = msgRawHandler
  116. }
  117. // goroutine safe
  118. func (p *Processor) Route(msg interface{}, userData interface{}) error {
  119. // raw
  120. if msgRaw, ok := msg.(MsgRaw); ok {
  121. if msgRaw.msgID >= uint16(len(p.msgInfo)) {
  122. return fmt.Errorf("message id %v not registered", msgRaw.msgID)
  123. }
  124. i := p.msgInfo[msgRaw.msgID]
  125. if i.msgRawHandler != nil {
  126. i.msgRawHandler([]interface{}{msgRaw.msgID, msgRaw.msgRawData, userData})
  127. }
  128. return nil
  129. }
  130. // protobuf
  131. msgType := reflect.TypeOf(msg)
  132. id, ok := p.msgID[msgType]
  133. if !ok {
  134. return fmt.Errorf("message %s not registered", msgType)
  135. }
  136. i := p.msgInfo[id]
  137. if i.msgHandler != nil {
  138. i.msgHandler([]interface{}{msg, userData})
  139. }
  140. if i.msgRouter != nil {
  141. i.msgRouter.Go(msgType, msg, userData)
  142. }
  143. return nil
  144. }
  145. // goroutine safe
  146. func (p *Processor) Unmarshal(data []byte) (interface{}, error) {
  147. if len(data) < 2 {
  148. return nil, errors.New("protobuf data too short")
  149. }
  150. // id
  151. var id uint16
  152. if p.littleEndian {
  153. id = binary.LittleEndian.Uint16(data)
  154. } else {
  155. id = binary.BigEndian.Uint16(data)
  156. }
  157. if id >= uint16(len(p.msgInfo)) {
  158. return nil, fmt.Errorf("message id %v not registered", id)
  159. }
  160. // msg
  161. i := p.msgInfo[id]
  162. if i.msgRawHandler != nil {
  163. return MsgRaw{id, data[2:]}, nil
  164. } else {
  165. msg := reflect.New(i.msgType.Elem()).Interface()
  166. // return msg, proto.UnmarshalMerge(data[2:], msg.(proto.Message))
  167. return msg, proto.Unmarshal(data[2:], msg.(proto.Message))
  168. }
  169. }
  170. // goroutine safe
  171. func (p *Processor) Marshal(msg interface{}) ([][]byte, error) {
  172. msgType := reflect.TypeOf(msg)
  173. // id
  174. _id, ok := p.msgID[msgType]
  175. if !ok {
  176. err := fmt.Errorf("message %s not registered", msgType)
  177. return nil, err
  178. }
  179. id := make([]byte, 2)
  180. if p.littleEndian {
  181. binary.LittleEndian.PutUint16(id, _id)
  182. } else {
  183. binary.BigEndian.PutUint16(id, _id)
  184. }
  185. // data
  186. data, err := proto.Marshal(msg.(proto.Message))
  187. return [][]byte{id, data}, err
  188. }
  189. // goroutine safe
  190. func (p *Processor) Range(f func(id uint16, t reflect.Type)) {
  191. for id, i := range p.msgInfo {
  192. f(uint16(id), i.msgType)
  193. }
  194. }