encode.go 2.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114
  1. package eventstream
  2. import (
  3. "bytes"
  4. "encoding/binary"
  5. "hash"
  6. "hash/crc32"
  7. "io"
  8. )
  9. // Encoder provides EventStream message encoding.
  10. type Encoder struct {
  11. w io.Writer
  12. headersBuf *bytes.Buffer
  13. }
  14. // NewEncoder initializes and returns an Encoder to encode Event Stream
  15. // messages to an io.Writer.
  16. func NewEncoder(w io.Writer) *Encoder {
  17. return &Encoder{
  18. w: w,
  19. headersBuf: bytes.NewBuffer(nil),
  20. }
  21. }
  22. // Encode encodes a single EventStream message to the io.Writer the Encoder
  23. // was created with. An error is returned if writing the message fails.
  24. func (e *Encoder) Encode(msg Message) error {
  25. e.headersBuf.Reset()
  26. err := encodeHeaders(e.headersBuf, msg.Headers)
  27. if err != nil {
  28. return err
  29. }
  30. crc := crc32.New(crc32IEEETable)
  31. hashWriter := io.MultiWriter(e.w, crc)
  32. headersLen := uint32(e.headersBuf.Len())
  33. payloadLen := uint32(len(msg.Payload))
  34. if err := encodePrelude(hashWriter, crc, headersLen, payloadLen); err != nil {
  35. return err
  36. }
  37. if headersLen > 0 {
  38. if _, err := io.Copy(hashWriter, e.headersBuf); err != nil {
  39. return err
  40. }
  41. }
  42. if payloadLen > 0 {
  43. if _, err := hashWriter.Write(msg.Payload); err != nil {
  44. return err
  45. }
  46. }
  47. msgCRC := crc.Sum32()
  48. return binary.Write(e.w, binary.BigEndian, msgCRC)
  49. }
  50. func encodePrelude(w io.Writer, crc hash.Hash32, headersLen, payloadLen uint32) error {
  51. p := messagePrelude{
  52. Length: minMsgLen + headersLen + payloadLen,
  53. HeadersLen: headersLen,
  54. }
  55. if err := p.ValidateLens(); err != nil {
  56. return err
  57. }
  58. err := binaryWriteFields(w, binary.BigEndian,
  59. p.Length,
  60. p.HeadersLen,
  61. )
  62. if err != nil {
  63. return err
  64. }
  65. p.PreludeCRC = crc.Sum32()
  66. err = binary.Write(w, binary.BigEndian, p.PreludeCRC)
  67. if err != nil {
  68. return err
  69. }
  70. return nil
  71. }
  72. func encodeHeaders(w io.Writer, headers Headers) error {
  73. for _, h := range headers {
  74. hn := headerName{
  75. Len: uint8(len(h.Name)),
  76. }
  77. copy(hn.Name[:hn.Len], h.Name)
  78. if err := hn.encode(w); err != nil {
  79. return err
  80. }
  81. if err := h.Value.encode(w); err != nil {
  82. return err
  83. }
  84. }
  85. return nil
  86. }
  87. func binaryWriteFields(w io.Writer, order binary.ByteOrder, vs ...interface{}) error {
  88. for _, v := range vs {
  89. if err := binary.Write(w, order, v); err != nil {
  90. return err
  91. }
  92. }
  93. return nil
  94. }