115 lines
2.2 KiB
Go
115 lines
2.2 KiB
Go
package eventstream
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/binary"
|
|
"hash"
|
|
"hash/crc32"
|
|
"io"
|
|
)
|
|
|
|
// Encoder provides EventStream message encoding.
|
|
type Encoder struct {
|
|
w io.Writer
|
|
|
|
headersBuf *bytes.Buffer
|
|
}
|
|
|
|
// NewEncoder initializes and returns an Encoder to encode Event Stream
|
|
// messages to an io.Writer.
|
|
func NewEncoder(w io.Writer) *Encoder {
|
|
return &Encoder{
|
|
w: w,
|
|
headersBuf: bytes.NewBuffer(nil),
|
|
}
|
|
}
|
|
|
|
// Encode encodes a single EventStream message to the io.Writer the Encoder
|
|
// was created with. An error is returned if writing the message fails.
|
|
func (e *Encoder) Encode(msg Message) error {
|
|
e.headersBuf.Reset()
|
|
|
|
err := encodeHeaders(e.headersBuf, msg.Headers)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
crc := crc32.New(crc32IEEETable)
|
|
hashWriter := io.MultiWriter(e.w, crc)
|
|
|
|
headersLen := uint32(e.headersBuf.Len())
|
|
payloadLen := uint32(len(msg.Payload))
|
|
|
|
if err := encodePrelude(hashWriter, crc, headersLen, payloadLen); err != nil {
|
|
return err
|
|
}
|
|
|
|
if headersLen > 0 {
|
|
if _, err := io.Copy(hashWriter, e.headersBuf); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
if payloadLen > 0 {
|
|
if _, err := hashWriter.Write(msg.Payload); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
msgCRC := crc.Sum32()
|
|
return binary.Write(e.w, binary.BigEndian, msgCRC)
|
|
}
|
|
|
|
func encodePrelude(w io.Writer, crc hash.Hash32, headersLen, payloadLen uint32) error {
|
|
p := messagePrelude{
|
|
Length: minMsgLen + headersLen + payloadLen,
|
|
HeadersLen: headersLen,
|
|
}
|
|
if err := p.ValidateLens(); err != nil {
|
|
return err
|
|
}
|
|
|
|
err := binaryWriteFields(w, binary.BigEndian,
|
|
p.Length,
|
|
p.HeadersLen,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
p.PreludeCRC = crc.Sum32()
|
|
err = binary.Write(w, binary.BigEndian, p.PreludeCRC)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func encodeHeaders(w io.Writer, headers Headers) error {
|
|
for _, h := range headers {
|
|
hn := headerName{
|
|
Len: uint8(len(h.Name)),
|
|
}
|
|
copy(hn.Name[:hn.Len], h.Name)
|
|
if err := hn.encode(w); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := h.Value.encode(w); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func binaryWriteFields(w io.Writer, order binary.ByteOrder, vs ...interface{}) error {
|
|
for _, v := range vs {
|
|
if err := binary.Write(w, order, v); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|