mirror of
https://github.com/aptly-dev/aptly.git
synced 2026-05-08 22:30:41 +00:00
Update Go AWS SDK to the latest version
This commit is contained in:
committed by
Andrey Smirnov
parent
d08be990ef
commit
94a72b23ff
+144
@@ -0,0 +1,144 @@
|
||||
package eventstream
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
type decodedMessage struct {
|
||||
rawMessage
|
||||
Headers decodedHeaders `json:"headers"`
|
||||
}
|
||||
type jsonMessage struct {
|
||||
Length json.Number `json:"total_length"`
|
||||
HeadersLen json.Number `json:"headers_length"`
|
||||
PreludeCRC json.Number `json:"prelude_crc"`
|
||||
Headers decodedHeaders `json:"headers"`
|
||||
Payload []byte `json:"payload"`
|
||||
CRC json.Number `json:"message_crc"`
|
||||
}
|
||||
|
||||
func (d *decodedMessage) UnmarshalJSON(b []byte) (err error) {
|
||||
var jsonMsg jsonMessage
|
||||
if err = json.Unmarshal(b, &jsonMsg); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
d.Length, err = numAsUint32(jsonMsg.Length)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
d.HeadersLen, err = numAsUint32(jsonMsg.HeadersLen)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
d.PreludeCRC, err = numAsUint32(jsonMsg.PreludeCRC)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
d.Headers = jsonMsg.Headers
|
||||
d.Payload = jsonMsg.Payload
|
||||
d.CRC, err = numAsUint32(jsonMsg.CRC)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *decodedMessage) MarshalJSON() ([]byte, error) {
|
||||
jsonMsg := jsonMessage{
|
||||
Length: json.Number(strconv.Itoa(int(d.Length))),
|
||||
HeadersLen: json.Number(strconv.Itoa(int(d.HeadersLen))),
|
||||
PreludeCRC: json.Number(strconv.Itoa(int(d.PreludeCRC))),
|
||||
Headers: d.Headers,
|
||||
Payload: d.Payload,
|
||||
CRC: json.Number(strconv.Itoa(int(d.CRC))),
|
||||
}
|
||||
|
||||
return json.Marshal(jsonMsg)
|
||||
}
|
||||
|
||||
func numAsUint32(n json.Number) (uint32, error) {
|
||||
v, err := n.Int64()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("failed to get int64 json number, %v", err)
|
||||
}
|
||||
|
||||
return uint32(v), nil
|
||||
}
|
||||
|
||||
func (d decodedMessage) Message() Message {
|
||||
return Message{
|
||||
Headers: Headers(d.Headers),
|
||||
Payload: d.Payload,
|
||||
}
|
||||
}
|
||||
|
||||
type decodedHeaders Headers
|
||||
|
||||
func (hs *decodedHeaders) UnmarshalJSON(b []byte) error {
|
||||
var jsonHeaders []struct {
|
||||
Name string `json:"name"`
|
||||
Type valueType `json:"type"`
|
||||
Value interface{} `json:"value"`
|
||||
}
|
||||
|
||||
decoder := json.NewDecoder(bytes.NewReader(b))
|
||||
decoder.UseNumber()
|
||||
if err := decoder.Decode(&jsonHeaders); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var headers Headers
|
||||
for _, h := range jsonHeaders {
|
||||
value, err := valueFromType(h.Type, h.Value)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
headers.Set(h.Name, value)
|
||||
}
|
||||
(*hs) = decodedHeaders(headers)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func valueFromType(typ valueType, val interface{}) (Value, error) {
|
||||
switch typ {
|
||||
case trueValueType:
|
||||
return BoolValue(true), nil
|
||||
case falseValueType:
|
||||
return BoolValue(false), nil
|
||||
case int8ValueType:
|
||||
v, err := val.(json.Number).Int64()
|
||||
return Int8Value(int8(v)), err
|
||||
case int16ValueType:
|
||||
v, err := val.(json.Number).Int64()
|
||||
return Int16Value(int16(v)), err
|
||||
case int32ValueType:
|
||||
v, err := val.(json.Number).Int64()
|
||||
return Int32Value(int32(v)), err
|
||||
case int64ValueType:
|
||||
v, err := val.(json.Number).Int64()
|
||||
return Int64Value(v), err
|
||||
case bytesValueType:
|
||||
v, err := base64.StdEncoding.DecodeString(val.(string))
|
||||
return BytesValue(v), err
|
||||
case stringValueType:
|
||||
v, err := base64.StdEncoding.DecodeString(val.(string))
|
||||
return StringValue(string(v)), err
|
||||
case timestampValueType:
|
||||
v, err := val.(json.Number).Int64()
|
||||
return TimestampValue(timeFromEpochMilli(v)), err
|
||||
case uuidValueType:
|
||||
v, err := base64.StdEncoding.DecodeString(val.(string))
|
||||
var tv UUIDValue
|
||||
copy(tv[:], v)
|
||||
return tv, err
|
||||
default:
|
||||
panic(fmt.Sprintf("unknown type, %s, %T", typ.String(), val))
|
||||
}
|
||||
}
|
||||
+199
@@ -0,0 +1,199 @@
|
||||
package eventstream
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"hash"
|
||||
"hash/crc32"
|
||||
"io"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
)
|
||||
|
||||
// Decoder provides decoding of an Event Stream messages.
|
||||
type Decoder struct {
|
||||
r io.Reader
|
||||
logger aws.Logger
|
||||
}
|
||||
|
||||
// NewDecoder initializes and returns a Decoder for decoding event
|
||||
// stream messages from the reader provided.
|
||||
func NewDecoder(r io.Reader) *Decoder {
|
||||
return &Decoder{
|
||||
r: r,
|
||||
}
|
||||
}
|
||||
|
||||
// Decode attempts to decode a single message from the event stream reader.
|
||||
// Will return the event stream message, or error if Decode fails to read
|
||||
// the message from the stream.
|
||||
func (d *Decoder) Decode(payloadBuf []byte) (m Message, err error) {
|
||||
reader := d.r
|
||||
if d.logger != nil {
|
||||
debugMsgBuf := bytes.NewBuffer(nil)
|
||||
reader = io.TeeReader(reader, debugMsgBuf)
|
||||
defer func() {
|
||||
logMessageDecode(d.logger, debugMsgBuf, m, err)
|
||||
}()
|
||||
}
|
||||
|
||||
crc := crc32.New(crc32IEEETable)
|
||||
hashReader := io.TeeReader(reader, crc)
|
||||
|
||||
prelude, err := decodePrelude(hashReader, crc)
|
||||
if err != nil {
|
||||
return Message{}, err
|
||||
}
|
||||
|
||||
if prelude.HeadersLen > 0 {
|
||||
lr := io.LimitReader(hashReader, int64(prelude.HeadersLen))
|
||||
m.Headers, err = decodeHeaders(lr)
|
||||
if err != nil {
|
||||
return Message{}, err
|
||||
}
|
||||
}
|
||||
|
||||
if payloadLen := prelude.PayloadLen(); payloadLen > 0 {
|
||||
buf, err := decodePayload(payloadBuf, io.LimitReader(hashReader, int64(payloadLen)))
|
||||
if err != nil {
|
||||
return Message{}, err
|
||||
}
|
||||
m.Payload = buf
|
||||
}
|
||||
|
||||
msgCRC := crc.Sum32()
|
||||
if err := validateCRC(reader, msgCRC); err != nil {
|
||||
return Message{}, err
|
||||
}
|
||||
|
||||
return m, nil
|
||||
}
|
||||
|
||||
// UseLogger specifies the Logger that that the decoder should use to log the
|
||||
// message decode to.
|
||||
func (d *Decoder) UseLogger(logger aws.Logger) {
|
||||
d.logger = logger
|
||||
}
|
||||
|
||||
func logMessageDecode(logger aws.Logger, msgBuf *bytes.Buffer, msg Message, decodeErr error) {
|
||||
w := bytes.NewBuffer(nil)
|
||||
defer func() { logger.Log(w.String()) }()
|
||||
|
||||
fmt.Fprintf(w, "Raw message:\n%s\n",
|
||||
hex.Dump(msgBuf.Bytes()))
|
||||
|
||||
if decodeErr != nil {
|
||||
fmt.Fprintf(w, "Decode error: %v\n", decodeErr)
|
||||
return
|
||||
}
|
||||
|
||||
rawMsg, err := msg.rawMessage()
|
||||
if err != nil {
|
||||
fmt.Fprintf(w, "failed to create raw message, %v\n", err)
|
||||
return
|
||||
}
|
||||
|
||||
decodedMsg := decodedMessage{
|
||||
rawMessage: rawMsg,
|
||||
Headers: decodedHeaders(msg.Headers),
|
||||
}
|
||||
|
||||
fmt.Fprintf(w, "Decoded message:\n")
|
||||
encoder := json.NewEncoder(w)
|
||||
if err := encoder.Encode(decodedMsg); err != nil {
|
||||
fmt.Fprintf(w, "failed to generate decoded message, %v\n", err)
|
||||
}
|
||||
}
|
||||
|
||||
func decodePrelude(r io.Reader, crc hash.Hash32) (messagePrelude, error) {
|
||||
var p messagePrelude
|
||||
|
||||
var err error
|
||||
p.Length, err = decodeUint32(r)
|
||||
if err != nil {
|
||||
return messagePrelude{}, err
|
||||
}
|
||||
|
||||
p.HeadersLen, err = decodeUint32(r)
|
||||
if err != nil {
|
||||
return messagePrelude{}, err
|
||||
}
|
||||
|
||||
if err := p.ValidateLens(); err != nil {
|
||||
return messagePrelude{}, err
|
||||
}
|
||||
|
||||
preludeCRC := crc.Sum32()
|
||||
if err := validateCRC(r, preludeCRC); err != nil {
|
||||
return messagePrelude{}, err
|
||||
}
|
||||
|
||||
p.PreludeCRC = preludeCRC
|
||||
|
||||
return p, nil
|
||||
}
|
||||
|
||||
func decodePayload(buf []byte, r io.Reader) ([]byte, error) {
|
||||
w := bytes.NewBuffer(buf[0:0])
|
||||
|
||||
_, err := io.Copy(w, r)
|
||||
return w.Bytes(), err
|
||||
}
|
||||
|
||||
func decodeUint8(r io.Reader) (uint8, error) {
|
||||
type byteReader interface {
|
||||
ReadByte() (byte, error)
|
||||
}
|
||||
|
||||
if br, ok := r.(byteReader); ok {
|
||||
v, err := br.ReadByte()
|
||||
return uint8(v), err
|
||||
}
|
||||
|
||||
var b [1]byte
|
||||
_, err := io.ReadFull(r, b[:])
|
||||
return uint8(b[0]), err
|
||||
}
|
||||
func decodeUint16(r io.Reader) (uint16, error) {
|
||||
var b [2]byte
|
||||
bs := b[:]
|
||||
_, err := io.ReadFull(r, bs)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return binary.BigEndian.Uint16(bs), nil
|
||||
}
|
||||
func decodeUint32(r io.Reader) (uint32, error) {
|
||||
var b [4]byte
|
||||
bs := b[:]
|
||||
_, err := io.ReadFull(r, bs)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return binary.BigEndian.Uint32(bs), nil
|
||||
}
|
||||
func decodeUint64(r io.Reader) (uint64, error) {
|
||||
var b [8]byte
|
||||
bs := b[:]
|
||||
_, err := io.ReadFull(r, bs)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return binary.BigEndian.Uint64(bs), nil
|
||||
}
|
||||
|
||||
func validateCRC(r io.Reader, expect uint32) error {
|
||||
msgCRC, err := decodeUint32(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if msgCRC != expect {
|
||||
return ChecksumError{}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
+168
@@ -0,0 +1,168 @@
|
||||
package eventstream
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/hex"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestWriteEncodedFromDecoded(t *testing.T) {
|
||||
cases, err := readPositiveTests("testdata")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to load positive tests, %v", err)
|
||||
}
|
||||
|
||||
for _, c := range cases {
|
||||
f, err := os.Create(filepath.Join("testdata", "encoded", "positive", c.Name))
|
||||
if err != nil {
|
||||
t.Fatalf("failed to open %q, %v", c.Name, err)
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
encoder := NewEncoder(f)
|
||||
|
||||
msg := c.Decoded.Message()
|
||||
if err := encoder.Encode(msg); err != nil {
|
||||
t.Errorf("failed to encode %q, %v", c.Name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecoder_Decode(t *testing.T) {
|
||||
cases, err := readPositiveTests("testdata")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to load positive tests, %v", err)
|
||||
}
|
||||
|
||||
for _, c := range cases {
|
||||
decoder := NewDecoder(bytes.NewBuffer(c.Encoded))
|
||||
|
||||
msg, err := decoder.Decode(nil)
|
||||
if err != nil {
|
||||
t.Fatalf("%s, expect no decode error, got %v", c.Name, err)
|
||||
}
|
||||
|
||||
raw, err := msg.rawMessage() // rawMessage will fail if payload read CRC fails
|
||||
if err != nil {
|
||||
t.Fatalf("%s, failed to get raw decoded message %v", c.Name, err)
|
||||
}
|
||||
|
||||
if e, a := c.Decoded.Length, raw.Length; e != a {
|
||||
t.Errorf("%s, expect %v length, got %v", c.Name, e, a)
|
||||
}
|
||||
if e, a := c.Decoded.HeadersLen, raw.HeadersLen; e != a {
|
||||
t.Errorf("%s, expect %v HeadersLen, got %v", c.Name, e, a)
|
||||
}
|
||||
if e, a := c.Decoded.PreludeCRC, raw.PreludeCRC; e != a {
|
||||
t.Errorf("%s, expect %v PreludeCRC, got %v", c.Name, e, a)
|
||||
}
|
||||
if e, a := Headers(c.Decoded.Headers), msg.Headers; !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("%s, expect %v headers, got %v", c.Name, e, a)
|
||||
}
|
||||
if e, a := c.Decoded.Payload, raw.Payload; !bytes.Equal(e, a) {
|
||||
t.Errorf("%s, expect %v payload, got %v", c.Name, e, a)
|
||||
}
|
||||
if e, a := c.Decoded.CRC, raw.CRC; e != a {
|
||||
t.Errorf("%s, expect %v CRC, got %v", c.Name, e, a)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecoder_Decode_Negative(t *testing.T) {
|
||||
cases, err := readNegativeTests("testdata")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to load negative tests, %v", err)
|
||||
}
|
||||
|
||||
for _, c := range cases {
|
||||
decoder := NewDecoder(bytes.NewBuffer(c.Encoded))
|
||||
|
||||
msg, err := decoder.Decode(nil)
|
||||
if err == nil {
|
||||
rawMsg, rawMsgErr := msg.rawMessage()
|
||||
t.Fatalf("%s, expect error, got none, %s,\n%s\n%#v, %v\n", c.Name,
|
||||
c.Err, hex.Dump(c.Encoded), rawMsg, rawMsgErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var testEncodedMsg = []byte{0, 0, 0, 61, 0, 0, 0, 32, 7, 253, 131, 150, 12, 99, 111, 110, 116, 101, 110, 116, 45, 116, 121, 112, 101, 7, 0, 16, 97, 112, 112, 108, 105, 99, 97, 116, 105, 111, 110, 47, 106, 115, 111, 110, 123, 39, 102, 111, 111, 39, 58, 39, 98, 97, 114, 39, 125, 141, 156, 8, 177}
|
||||
|
||||
func TestDecoder_DecodeMultipleMessages(t *testing.T) {
|
||||
const (
|
||||
expectMsgCount = 10
|
||||
expectPayloadLen = 13
|
||||
)
|
||||
|
||||
r := bytes.NewBuffer(nil)
|
||||
for i := 0; i < expectMsgCount; i++ {
|
||||
r.Write(testEncodedMsg)
|
||||
}
|
||||
|
||||
decoder := NewDecoder(r)
|
||||
|
||||
var err error
|
||||
var msg Message
|
||||
var count int
|
||||
for {
|
||||
msg, err = decoder.Decode(nil)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
count++
|
||||
|
||||
if e, a := expectPayloadLen, len(msg.Payload); e != a {
|
||||
t.Errorf("expect %v payload len, got %v", e, a)
|
||||
}
|
||||
|
||||
if e, a := []byte(`{'foo':'bar'}`), msg.Payload; !bytes.Equal(e, a) {
|
||||
t.Errorf("expect %v payload, got %v", e, a)
|
||||
}
|
||||
}
|
||||
|
||||
type causer interface {
|
||||
Cause() error
|
||||
}
|
||||
if err != nil && count != expectMsgCount {
|
||||
t.Fatalf("expect, no error, got %v", err)
|
||||
}
|
||||
|
||||
if e, a := expectMsgCount, count; e != a {
|
||||
t.Errorf("expect %v messages read, got %v", e, a)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkDecode(b *testing.B) {
|
||||
r := bytes.NewReader(testEncodedMsg)
|
||||
decoder := NewDecoder(r)
|
||||
payloadBuf := make([]byte, 0, 5*1024)
|
||||
b.ResetTimer()
|
||||
|
||||
for i := 0; i < b.N; i++ {
|
||||
msg, err := decoder.Decode(payloadBuf)
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
|
||||
// Release the payload buffer
|
||||
payloadBuf = msg.Payload[0:0]
|
||||
r.Seek(0, 0)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkDecode_NoPayloadBuf(b *testing.B) {
|
||||
r := bytes.NewReader(testEncodedMsg)
|
||||
decoder := NewDecoder(r)
|
||||
b.ResetTimer()
|
||||
|
||||
for i := 0; i < b.N; i++ {
|
||||
_, err := decoder.Decode(nil)
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
r.Seek(0, 0)
|
||||
}
|
||||
}
|
||||
+114
@@ -0,0 +1,114 @@
|
||||
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
|
||||
}
|
||||
+50
@@ -0,0 +1,50 @@
|
||||
package eventstream
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/hex"
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestEncoder_Encode(t *testing.T) {
|
||||
cases, err := readPositiveTests("testdata")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to load positive tests, %v", err)
|
||||
}
|
||||
|
||||
for _, c := range cases {
|
||||
var w bytes.Buffer
|
||||
encoder := NewEncoder(&w)
|
||||
|
||||
err = encoder.Encode(c.Decoded.Message())
|
||||
if err != nil {
|
||||
t.Fatalf("%s, failed to encode message, %v", c.Name, err)
|
||||
}
|
||||
|
||||
if e, a := c.Encoded, w.Bytes(); !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("%s, expect:\n%v\nactual:\n%v\n", c.Name,
|
||||
hex.Dump(e), hex.Dump(a))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkEncode(b *testing.B) {
|
||||
var w bytes.Buffer
|
||||
encoder := NewEncoder(&w)
|
||||
msg := Message{
|
||||
Headers: Headers{
|
||||
{Name: "event-id", Value: Int16Value(123)},
|
||||
},
|
||||
Payload: []byte(`{"abc":123}`),
|
||||
}
|
||||
|
||||
b.ResetTimer()
|
||||
|
||||
for i := 0; i < b.N; i++ {
|
||||
err := encoder.Encode(msg)
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
package eventstream
|
||||
|
||||
import "fmt"
|
||||
|
||||
// LengthError provides the error for items being larger than a maximum length.
|
||||
type LengthError struct {
|
||||
Part string
|
||||
Want int
|
||||
Have int
|
||||
Value interface{}
|
||||
}
|
||||
|
||||
func (e LengthError) Error() string {
|
||||
return fmt.Sprintf("%s length invalid, %d/%d, %v",
|
||||
e.Part, e.Want, e.Have, e.Value)
|
||||
}
|
||||
|
||||
// ChecksumError provides the error for message checksum invalidation errors.
|
||||
type ChecksumError struct{}
|
||||
|
||||
func (e ChecksumError) Error() string {
|
||||
return "message checksum mismatch"
|
||||
}
|
||||
Generated
Vendored
+196
@@ -0,0 +1,196 @@
|
||||
package eventstreamapi
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/private/protocol"
|
||||
"github.com/aws/aws-sdk-go/private/protocol/eventstream"
|
||||
)
|
||||
|
||||
// Unmarshaler provides the interface for unmarshaling a EventStream
|
||||
// message into a SDK type.
|
||||
type Unmarshaler interface {
|
||||
UnmarshalEvent(protocol.PayloadUnmarshaler, eventstream.Message) error
|
||||
}
|
||||
|
||||
// EventStream headers with specific meaning to async API functionality.
|
||||
const (
|
||||
MessageTypeHeader = `:message-type` // Identifies type of message.
|
||||
EventMessageType = `event`
|
||||
ErrorMessageType = `error`
|
||||
ExceptionMessageType = `exception`
|
||||
|
||||
// Message Events
|
||||
EventTypeHeader = `:event-type` // Identifies message event type e.g. "Stats".
|
||||
|
||||
// Message Error
|
||||
ErrorCodeHeader = `:error-code`
|
||||
ErrorMessageHeader = `:error-message`
|
||||
|
||||
// Message Exception
|
||||
ExceptionTypeHeader = `:exception-type`
|
||||
)
|
||||
|
||||
// EventReader provides reading from the EventStream of an reader.
|
||||
type EventReader struct {
|
||||
reader io.ReadCloser
|
||||
decoder *eventstream.Decoder
|
||||
|
||||
unmarshalerForEventType func(string) (Unmarshaler, error)
|
||||
payloadUnmarshaler protocol.PayloadUnmarshaler
|
||||
|
||||
payloadBuf []byte
|
||||
}
|
||||
|
||||
// NewEventReader returns a EventReader built from the reader and unmarshaler
|
||||
// provided. Use ReadStream method to start reading from the EventStream.
|
||||
func NewEventReader(
|
||||
reader io.ReadCloser,
|
||||
payloadUnmarshaler protocol.PayloadUnmarshaler,
|
||||
unmarshalerForEventType func(string) (Unmarshaler, error),
|
||||
) *EventReader {
|
||||
return &EventReader{
|
||||
reader: reader,
|
||||
decoder: eventstream.NewDecoder(reader),
|
||||
payloadUnmarshaler: payloadUnmarshaler,
|
||||
unmarshalerForEventType: unmarshalerForEventType,
|
||||
payloadBuf: make([]byte, 10*1024),
|
||||
}
|
||||
}
|
||||
|
||||
// UseLogger instructs the EventReader to use the logger and log level
|
||||
// specified.
|
||||
func (r *EventReader) UseLogger(logger aws.Logger, logLevel aws.LogLevelType) {
|
||||
if logger != nil && logLevel.Matches(aws.LogDebugWithEventStreamBody) {
|
||||
r.decoder.UseLogger(logger)
|
||||
}
|
||||
}
|
||||
|
||||
// ReadEvent attempts to read a message from the EventStream and return the
|
||||
// unmarshaled event value that the message is for.
|
||||
//
|
||||
// For EventStream API errors check if the returned error satisfies the
|
||||
// awserr.Error interface to get the error's Code and Message components.
|
||||
//
|
||||
// EventUnmarshalers called with EventStream messages must take copies of the
|
||||
// message's Payload. The payload will is reused between events read.
|
||||
func (r *EventReader) ReadEvent() (event interface{}, err error) {
|
||||
msg, err := r.decoder.Decode(r.payloadBuf)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
// Reclaim payload buffer for next message read.
|
||||
r.payloadBuf = msg.Payload[0:0]
|
||||
}()
|
||||
|
||||
typ, err := GetHeaderString(msg, MessageTypeHeader)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
switch typ {
|
||||
case EventMessageType:
|
||||
return r.unmarshalEventMessage(msg)
|
||||
case ExceptionMessageType:
|
||||
err = r.unmarshalEventException(msg)
|
||||
return nil, err
|
||||
case ErrorMessageType:
|
||||
return nil, r.unmarshalErrorMessage(msg)
|
||||
default:
|
||||
return nil, fmt.Errorf("unknown eventstream message type, %v", typ)
|
||||
}
|
||||
}
|
||||
|
||||
func (r *EventReader) unmarshalEventMessage(
|
||||
msg eventstream.Message,
|
||||
) (event interface{}, err error) {
|
||||
eventType, err := GetHeaderString(msg, EventTypeHeader)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ev, err := r.unmarshalerForEventType(eventType)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
err = ev.UnmarshalEvent(r.payloadUnmarshaler, msg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return ev, nil
|
||||
}
|
||||
|
||||
func (r *EventReader) unmarshalEventException(
|
||||
msg eventstream.Message,
|
||||
) (err error) {
|
||||
eventType, err := GetHeaderString(msg, ExceptionTypeHeader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
ev, err := r.unmarshalerForEventType(eventType)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = ev.UnmarshalEvent(r.payloadUnmarshaler, msg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var ok bool
|
||||
err, ok = ev.(error)
|
||||
if !ok {
|
||||
err = messageError{
|
||||
code: "SerializationError",
|
||||
msg: fmt.Sprintf(
|
||||
"event stream exception %s mapped to non-error %T, %v",
|
||||
eventType, ev, ev,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *EventReader) unmarshalErrorMessage(msg eventstream.Message) (err error) {
|
||||
var msgErr messageError
|
||||
|
||||
msgErr.code, err = GetHeaderString(msg, ErrorCodeHeader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
msgErr.msg, err = GetHeaderString(msg, ErrorMessageHeader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return msgErr
|
||||
}
|
||||
|
||||
// Close closes the EventReader's EventStream reader.
|
||||
func (r *EventReader) Close() error {
|
||||
return r.reader.Close()
|
||||
}
|
||||
|
||||
// GetHeaderString returns the value of the header as a string. If the header
|
||||
// is not set or the value is not a string an error will be returned.
|
||||
func GetHeaderString(msg eventstream.Message, headerName string) (string, error) {
|
||||
headerVal := msg.Headers.Get(headerName)
|
||||
if headerVal == nil {
|
||||
return "", fmt.Errorf("error header %s not present", headerName)
|
||||
}
|
||||
|
||||
v, ok := headerVal.Get().(string)
|
||||
if !ok {
|
||||
return "", fmt.Errorf("error header value is not a string, %T", headerVal)
|
||||
}
|
||||
|
||||
return v, nil
|
||||
}
|
||||
Generated
Vendored
+253
@@ -0,0 +1,253 @@
|
||||
package eventstreamapi
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"testing"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws/request"
|
||||
"github.com/aws/aws-sdk-go/private/protocol"
|
||||
"github.com/aws/aws-sdk-go/private/protocol/eventstream"
|
||||
"github.com/aws/aws-sdk-go/private/protocol/restjson"
|
||||
)
|
||||
|
||||
var eventMessageTypeHeader = eventstream.Header{
|
||||
Name: MessageTypeHeader,
|
||||
Value: eventstream.StringValue(EventMessageType),
|
||||
}
|
||||
|
||||
func TestEventReader(t *testing.T) {
|
||||
stream := createStream(
|
||||
eventstream.Message{
|
||||
Headers: eventstream.Headers{
|
||||
eventMessageTypeHeader,
|
||||
eventstream.Header{
|
||||
Name: EventTypeHeader,
|
||||
Value: eventstream.StringValue("eventABC"),
|
||||
},
|
||||
},
|
||||
},
|
||||
eventstream.Message{
|
||||
Headers: eventstream.Headers{
|
||||
eventMessageTypeHeader,
|
||||
eventstream.Header{
|
||||
Name: EventTypeHeader,
|
||||
Value: eventstream.StringValue("eventEFG"),
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
var unmarshalers request.HandlerList
|
||||
unmarshalers.PushBackNamed(restjson.UnmarshalHandler)
|
||||
|
||||
eventReader := NewEventReader(stream,
|
||||
protocol.HandlerPayloadUnmarshal{
|
||||
Unmarshalers: unmarshalers,
|
||||
},
|
||||
unmarshalerForEventType,
|
||||
)
|
||||
|
||||
event, err := eventReader.ReadEvent()
|
||||
if err != nil {
|
||||
t.Fatalf("expect no error, got %v", err)
|
||||
}
|
||||
|
||||
if event == nil {
|
||||
t.Fatalf("expect event got none")
|
||||
}
|
||||
|
||||
event, err = eventReader.ReadEvent()
|
||||
if err == nil {
|
||||
t.Fatalf("expect error for unknown event, got none")
|
||||
}
|
||||
|
||||
if event != nil {
|
||||
t.Fatalf("expect no event, got %T, %v", event, event)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventReader_Error(t *testing.T) {
|
||||
stream := createStream(
|
||||
eventstream.Message{
|
||||
Headers: eventstream.Headers{
|
||||
eventstream.Header{
|
||||
Name: MessageTypeHeader,
|
||||
Value: eventstream.StringValue(ErrorMessageType),
|
||||
},
|
||||
eventstream.Header{
|
||||
Name: ErrorCodeHeader,
|
||||
Value: eventstream.StringValue("errorCode"),
|
||||
},
|
||||
eventstream.Header{
|
||||
Name: ErrorMessageHeader,
|
||||
Value: eventstream.StringValue("error message occur"),
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
var unmarshalers request.HandlerList
|
||||
unmarshalers.PushBackNamed(restjson.UnmarshalHandler)
|
||||
|
||||
eventReader := NewEventReader(stream,
|
||||
protocol.HandlerPayloadUnmarshal{
|
||||
Unmarshalers: unmarshalers,
|
||||
},
|
||||
unmarshalerForEventType,
|
||||
)
|
||||
|
||||
event, err := eventReader.ReadEvent()
|
||||
if err == nil {
|
||||
t.Fatalf("expect error got none")
|
||||
}
|
||||
|
||||
if event != nil {
|
||||
t.Fatalf("expect no event, got %v", event)
|
||||
}
|
||||
|
||||
if e, a := "errorCode: error message occur", err.Error(); e != a {
|
||||
t.Errorf("expect %v error, got %v", e, a)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventReader_Exception(t *testing.T) {
|
||||
eventMsgs := []eventstream.Message{
|
||||
{
|
||||
Headers: eventstream.Headers{
|
||||
eventstream.Header{
|
||||
Name: MessageTypeHeader,
|
||||
Value: eventstream.StringValue(ExceptionMessageType),
|
||||
},
|
||||
eventstream.Header{
|
||||
Name: ExceptionTypeHeader,
|
||||
Value: eventstream.StringValue("exception"),
|
||||
},
|
||||
},
|
||||
Payload: []byte(`{"message":"exception message"}`),
|
||||
},
|
||||
}
|
||||
stream := createStream(eventMsgs...)
|
||||
|
||||
var unmarshalers request.HandlerList
|
||||
unmarshalers.PushBackNamed(restjson.UnmarshalHandler)
|
||||
|
||||
eventReader := NewEventReader(stream,
|
||||
protocol.HandlerPayloadUnmarshal{
|
||||
Unmarshalers: unmarshalers,
|
||||
},
|
||||
unmarshalerForEventType,
|
||||
)
|
||||
|
||||
event, err := eventReader.ReadEvent()
|
||||
if err == nil {
|
||||
t.Fatalf("expect error got none")
|
||||
}
|
||||
|
||||
if event != nil {
|
||||
t.Fatalf("expect no event, got %v", event)
|
||||
}
|
||||
|
||||
et := err.(*exceptionType)
|
||||
if e, a := string(eventMsgs[0].Payload), string(et.Payload); e != a {
|
||||
t.Errorf("expect %v payload, got %v", e, a)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkEventReader(b *testing.B) {
|
||||
var buf bytes.Buffer
|
||||
encoder := eventstream.NewEncoder(&buf)
|
||||
msg := eventstream.Message{
|
||||
Headers: eventstream.Headers{
|
||||
eventMessageTypeHeader,
|
||||
eventstream.Header{
|
||||
Name: EventTypeHeader,
|
||||
Value: eventstream.StringValue("eventABC"),
|
||||
},
|
||||
},
|
||||
}
|
||||
if err := encoder.Encode(msg); err != nil {
|
||||
b.Fatalf("failed to encode message, %v", err)
|
||||
}
|
||||
stream := bytes.NewReader(buf.Bytes())
|
||||
|
||||
var unmarshalers request.HandlerList
|
||||
unmarshalers.PushBackNamed(restjson.UnmarshalHandler)
|
||||
|
||||
eventReader := NewEventReader(ioutil.NopCloser(stream),
|
||||
protocol.HandlerPayloadUnmarshal{
|
||||
Unmarshalers: unmarshalers,
|
||||
},
|
||||
unmarshalerForEventType,
|
||||
)
|
||||
b.ResetTimer()
|
||||
|
||||
for i := 0; i < b.N; i++ {
|
||||
stream.Seek(0, 0)
|
||||
|
||||
event, err := eventReader.ReadEvent()
|
||||
if err != nil {
|
||||
b.Fatalf("expect no error, got %v", err)
|
||||
}
|
||||
if event == nil {
|
||||
b.Fatalf("expect event got none")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func unmarshalerForEventType(eventType string) (Unmarshaler, error) {
|
||||
switch eventType {
|
||||
case "eventABC":
|
||||
return &eventABC{}, nil
|
||||
case "exception":
|
||||
return &exceptionType{}, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("unknown event type, %v", eventType)
|
||||
}
|
||||
}
|
||||
|
||||
type eventABC struct {
|
||||
_ struct{}
|
||||
|
||||
HeaderField string
|
||||
Payload []byte
|
||||
}
|
||||
|
||||
func (e *eventABC) UnmarshalEvent(
|
||||
unmarshaler protocol.PayloadUnmarshaler,
|
||||
msg eventstream.Message,
|
||||
) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func createStream(msgs ...eventstream.Message) io.ReadCloser {
|
||||
w := bytes.NewBuffer(nil)
|
||||
|
||||
encoder := eventstream.NewEncoder(w)
|
||||
|
||||
for _, msg := range msgs {
|
||||
if err := encoder.Encode(msg); err != nil {
|
||||
panic("createStream failed, " + err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
return ioutil.NopCloser(w)
|
||||
}
|
||||
|
||||
type exceptionType struct {
|
||||
Payload []byte
|
||||
}
|
||||
|
||||
func (e exceptionType) Error() string {
|
||||
return fmt.Sprintf("exception error message")
|
||||
}
|
||||
|
||||
func (e *exceptionType) UnmarshalEvent(
|
||||
unmarshaler protocol.PayloadUnmarshaler,
|
||||
msg eventstream.Message,
|
||||
) error {
|
||||
e.Payload = msg.Payload
|
||||
return nil
|
||||
}
|
||||
Generated
Vendored
+24
@@ -0,0 +1,24 @@
|
||||
package eventstreamapi
|
||||
|
||||
import "fmt"
|
||||
|
||||
type messageError struct {
|
||||
code string
|
||||
msg string
|
||||
}
|
||||
|
||||
func (e messageError) Code() string {
|
||||
return e.code
|
||||
}
|
||||
|
||||
func (e messageError) Message() string {
|
||||
return e.msg
|
||||
}
|
||||
|
||||
func (e messageError) Error() string {
|
||||
return fmt.Sprintf("%s: %s", e.code, e.msg)
|
||||
}
|
||||
|
||||
func (e messageError) OrigErr() error {
|
||||
return nil
|
||||
}
|
||||
Generated
Vendored
+17
@@ -0,0 +1,17 @@
|
||||
// +build !go1.10
|
||||
|
||||
package eventstreamtest
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
)
|
||||
|
||||
// /x/net/http2 is only available for the latest two versions of Go. Any Go
|
||||
// version older than that cannot use the utility to configure the http2
|
||||
// server.
|
||||
func setupServer(server *httptest.Server, useH2 bool) *http.Client {
|
||||
server.Start()
|
||||
|
||||
return nil
|
||||
}
|
||||
Generated
Vendored
+40
@@ -0,0 +1,40 @@
|
||||
// +build go1.10
|
||||
|
||||
package eventstreamtest
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
|
||||
"golang.org/x/net/http2"
|
||||
)
|
||||
|
||||
// /x/net/http2 is only available for the latest two versions of Go. Any Go
|
||||
// version older than that cannot use the utility to configure the http2
|
||||
// server.
|
||||
func setupServer(server *httptest.Server, useH2 bool) *http.Client {
|
||||
server.Config.TLSConfig = &tls.Config{
|
||||
InsecureSkipVerify: true,
|
||||
}
|
||||
|
||||
clientTrans := &http.Transport{
|
||||
TLSClientConfig: &tls.Config{
|
||||
InsecureSkipVerify: true,
|
||||
},
|
||||
}
|
||||
|
||||
if useH2 {
|
||||
http2.ConfigureServer(server.Config, nil)
|
||||
http2.ConfigureTransport(clientTrans)
|
||||
server.Config.TLSConfig.NextProtos = []string{http2.NextProtoTLS}
|
||||
clientTrans.TLSClientConfig.NextProtos = []string{http2.NextProtoTLS}
|
||||
}
|
||||
server.TLS = server.Config.TLSConfig
|
||||
|
||||
server.StartTLS()
|
||||
|
||||
return &http.Client{
|
||||
Transport: clientTrans,
|
||||
}
|
||||
}
|
||||
Generated
Vendored
+101
@@ -0,0 +1,101 @@
|
||||
package eventstreamtest
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/aws/session"
|
||||
"github.com/aws/aws-sdk-go/awstesting/unit"
|
||||
"github.com/aws/aws-sdk-go/private/protocol"
|
||||
"github.com/aws/aws-sdk-go/private/protocol/eventstream"
|
||||
"github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamapi"
|
||||
)
|
||||
|
||||
// ServeEventStream provides serving EventStream messages from a HTTP server to
|
||||
// the client. The events are sent sequentially to the client without delay.
|
||||
type ServeEventStream struct {
|
||||
T *testing.T
|
||||
Events []eventstream.Message
|
||||
}
|
||||
|
||||
func (s ServeEventStream) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
encoder := eventstream.NewEncoder(flushWriter{w})
|
||||
|
||||
for _, event := range s.Events {
|
||||
encoder.Encode(event)
|
||||
}
|
||||
}
|
||||
|
||||
// SetupEventStreamSession creates a HTTP server SDK session for communicating
|
||||
// with that server to be used for EventStream APIs. If HTTP/2 is enabled the
|
||||
// server/client will only attempt to use HTTP/2.
|
||||
func SetupEventStreamSession(
|
||||
t *testing.T, handler http.Handler, h2 bool,
|
||||
) (sess *session.Session, cleanupFn func(), err error) {
|
||||
server := httptest.NewUnstartedServer(handler)
|
||||
|
||||
client := setupServer(server, h2)
|
||||
|
||||
cleanupFn = func() {
|
||||
server.Close()
|
||||
}
|
||||
|
||||
sess, err = session.NewSession(unit.Session.Config, &aws.Config{
|
||||
Endpoint: &server.URL,
|
||||
DisableParamValidation: aws.Bool(true),
|
||||
HTTPClient: client,
|
||||
// LogLevel: aws.LogLevel(aws.LogDebugWithEventStreamBody),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
return sess, cleanupFn, nil
|
||||
}
|
||||
|
||||
type flushWriter struct {
|
||||
w io.Writer
|
||||
}
|
||||
|
||||
func (fw flushWriter) Write(p []byte) (n int, err error) {
|
||||
n, err = fw.w.Write(p)
|
||||
if f, ok := fw.w.(http.Flusher); ok {
|
||||
f.Flush()
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// MarshalEventPayload marshals a SDK API shape into its associated wire
|
||||
// protocol payload.
|
||||
func MarshalEventPayload(
|
||||
payloadMarshaler protocol.PayloadMarshaler,
|
||||
v interface{},
|
||||
) []byte {
|
||||
var w bytes.Buffer
|
||||
err := payloadMarshaler.MarshalPayload(&w, v)
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("failed to marshal event %T, %v", v, v))
|
||||
}
|
||||
|
||||
return w.Bytes()
|
||||
}
|
||||
|
||||
// EventMessageTypeHeader is an event message type header for specifying an
|
||||
// event is an message type.
|
||||
var EventMessageTypeHeader = eventstream.Header{
|
||||
Name: eventstreamapi.MessageTypeHeader,
|
||||
Value: eventstream.StringValue(eventstreamapi.EventMessageType),
|
||||
}
|
||||
|
||||
// EventExceptionTypeHeader is an event exception type header for specifying an
|
||||
// event is an exeption type.
|
||||
var EventExceptionTypeHeader = eventstream.Header{
|
||||
Name: eventstreamapi.MessageTypeHeader,
|
||||
Value: eventstream.StringValue(eventstreamapi.ExceptionMessageType),
|
||||
}
|
||||
+166
@@ -0,0 +1,166 @@
|
||||
package eventstream
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"io"
|
||||
)
|
||||
|
||||
// Headers are a collection of EventStream header values.
|
||||
type Headers []Header
|
||||
|
||||
// Header is a single EventStream Key Value header pair.
|
||||
type Header struct {
|
||||
Name string
|
||||
Value Value
|
||||
}
|
||||
|
||||
// Set associates the name with a value. If the header name already exists in
|
||||
// the Headers the value will be replaced with the new one.
|
||||
func (hs *Headers) Set(name string, value Value) {
|
||||
var i int
|
||||
for ; i < len(*hs); i++ {
|
||||
if (*hs)[i].Name == name {
|
||||
(*hs)[i].Value = value
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
*hs = append(*hs, Header{
|
||||
Name: name, Value: value,
|
||||
})
|
||||
}
|
||||
|
||||
// Get returns the Value associated with the header. Nil is returned if the
|
||||
// value does not exist.
|
||||
func (hs Headers) Get(name string) Value {
|
||||
for i := 0; i < len(hs); i++ {
|
||||
if h := hs[i]; h.Name == name {
|
||||
return h.Value
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Del deletes the value in the Headers if it exists.
|
||||
func (hs *Headers) Del(name string) {
|
||||
for i := 0; i < len(*hs); i++ {
|
||||
if (*hs)[i].Name == name {
|
||||
copy((*hs)[i:], (*hs)[i+1:])
|
||||
(*hs) = (*hs)[:len(*hs)-1]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func decodeHeaders(r io.Reader) (Headers, error) {
|
||||
hs := Headers{}
|
||||
|
||||
for {
|
||||
name, err := decodeHeaderName(r)
|
||||
if err != nil {
|
||||
if err == io.EOF {
|
||||
// EOF while getting header name means no more headers
|
||||
break
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
value, err := decodeHeaderValue(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
hs.Set(name, value)
|
||||
}
|
||||
|
||||
return hs, nil
|
||||
}
|
||||
|
||||
func decodeHeaderName(r io.Reader) (string, error) {
|
||||
var n headerName
|
||||
|
||||
var err error
|
||||
n.Len, err = decodeUint8(r)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
name := n.Name[:n.Len]
|
||||
if _, err := io.ReadFull(r, name); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return string(name), nil
|
||||
}
|
||||
|
||||
func decodeHeaderValue(r io.Reader) (Value, error) {
|
||||
var raw rawValue
|
||||
|
||||
typ, err := decodeUint8(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
raw.Type = valueType(typ)
|
||||
|
||||
var v Value
|
||||
|
||||
switch raw.Type {
|
||||
case trueValueType:
|
||||
v = BoolValue(true)
|
||||
case falseValueType:
|
||||
v = BoolValue(false)
|
||||
case int8ValueType:
|
||||
var tv Int8Value
|
||||
err = tv.decode(r)
|
||||
v = tv
|
||||
case int16ValueType:
|
||||
var tv Int16Value
|
||||
err = tv.decode(r)
|
||||
v = tv
|
||||
case int32ValueType:
|
||||
var tv Int32Value
|
||||
err = tv.decode(r)
|
||||
v = tv
|
||||
case int64ValueType:
|
||||
var tv Int64Value
|
||||
err = tv.decode(r)
|
||||
v = tv
|
||||
case bytesValueType:
|
||||
var tv BytesValue
|
||||
err = tv.decode(r)
|
||||
v = tv
|
||||
case stringValueType:
|
||||
var tv StringValue
|
||||
err = tv.decode(r)
|
||||
v = tv
|
||||
case timestampValueType:
|
||||
var tv TimestampValue
|
||||
err = tv.decode(r)
|
||||
v = tv
|
||||
case uuidValueType:
|
||||
var tv UUIDValue
|
||||
err = tv.decode(r)
|
||||
v = tv
|
||||
default:
|
||||
panic(fmt.Sprintf("unknown value type %d", raw.Type))
|
||||
}
|
||||
|
||||
// Error could be EOF, let caller deal with it
|
||||
return v, err
|
||||
}
|
||||
|
||||
const maxHeaderNameLen = 255
|
||||
|
||||
type headerName struct {
|
||||
Len uint8
|
||||
Name [maxHeaderNameLen]byte
|
||||
}
|
||||
|
||||
func (v headerName) encode(w io.Writer) error {
|
||||
if err := binary.Write(w, binary.BigEndian, v.Len); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
_, err := w.Write(v.Name[:v.Len])
|
||||
return err
|
||||
}
|
||||
+66
@@ -0,0 +1,66 @@
|
||||
package eventstream
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestHeaders_Set(t *testing.T) {
|
||||
expect := Headers{
|
||||
{Name: "ABC", Value: StringValue("123")},
|
||||
{Name: "EFG", Value: TimestampValue(time.Time{})},
|
||||
}
|
||||
|
||||
var actual Headers
|
||||
actual.Set("ABC", Int32Value(123))
|
||||
actual.Set("ABC", StringValue("123")) // replace case
|
||||
actual.Set("EFG", TimestampValue(time.Time{}))
|
||||
|
||||
if e, a := expect, actual; !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("expect %v headers, got %v", e, a)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHeaders_Get(t *testing.T) {
|
||||
headers := Headers{
|
||||
{Name: "ABC", Value: StringValue("123")},
|
||||
{Name: "EFG", Value: TimestampValue(time.Time{})},
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
Name string
|
||||
Value Value
|
||||
}{
|
||||
{Name: "ABC", Value: StringValue("123")},
|
||||
{Name: "EFG", Value: TimestampValue(time.Time{})},
|
||||
{Name: "NotFound"},
|
||||
}
|
||||
|
||||
for i, c := range cases {
|
||||
actual := headers.Get(c.Name)
|
||||
if e, a := c.Value, actual; !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("%d, expect %v value, got %v", i, e, a)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHeaders_Del(t *testing.T) {
|
||||
headers := Headers{
|
||||
{Name: "ABC", Value: StringValue("123")},
|
||||
{Name: "EFG", Value: TimestampValue(time.Time{})},
|
||||
{Name: "HIJ", Value: StringValue("123")},
|
||||
{Name: "KML", Value: TimestampValue(time.Time{})},
|
||||
}
|
||||
expectAfterDel := Headers{
|
||||
{Name: "EFG", Value: TimestampValue(time.Time{})},
|
||||
}
|
||||
|
||||
headers.Del("HIJ")
|
||||
headers.Del("ABC")
|
||||
headers.Del("KML")
|
||||
|
||||
if e, a := expectAfterDel, headers; !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("expect %v headers, got %v", e, a)
|
||||
}
|
||||
}
|
||||
+501
@@ -0,0 +1,501 @@
|
||||
package eventstream
|
||||
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"io"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
const maxHeaderValueLen = 1<<15 - 1 // 2^15-1 or 32KB - 1
|
||||
|
||||
// valueType is the EventStream header value type.
|
||||
type valueType uint8
|
||||
|
||||
// Header value types
|
||||
const (
|
||||
trueValueType valueType = iota
|
||||
falseValueType
|
||||
int8ValueType // Byte
|
||||
int16ValueType // Short
|
||||
int32ValueType // Integer
|
||||
int64ValueType // Long
|
||||
bytesValueType
|
||||
stringValueType
|
||||
timestampValueType
|
||||
uuidValueType
|
||||
)
|
||||
|
||||
func (t valueType) String() string {
|
||||
switch t {
|
||||
case trueValueType:
|
||||
return "bool"
|
||||
case falseValueType:
|
||||
return "bool"
|
||||
case int8ValueType:
|
||||
return "int8"
|
||||
case int16ValueType:
|
||||
return "int16"
|
||||
case int32ValueType:
|
||||
return "int32"
|
||||
case int64ValueType:
|
||||
return "int64"
|
||||
case bytesValueType:
|
||||
return "byte_array"
|
||||
case stringValueType:
|
||||
return "string"
|
||||
case timestampValueType:
|
||||
return "timestamp"
|
||||
case uuidValueType:
|
||||
return "uuid"
|
||||
default:
|
||||
return fmt.Sprintf("unknown value type %d", uint8(t))
|
||||
}
|
||||
}
|
||||
|
||||
type rawValue struct {
|
||||
Type valueType
|
||||
Len uint16 // Only set for variable length slices
|
||||
Value []byte // byte representation of value, BigEndian encoding.
|
||||
}
|
||||
|
||||
func (r rawValue) encodeScalar(w io.Writer, v interface{}) error {
|
||||
return binaryWriteFields(w, binary.BigEndian,
|
||||
r.Type,
|
||||
v,
|
||||
)
|
||||
}
|
||||
|
||||
func (r rawValue) encodeFixedSlice(w io.Writer, v []byte) error {
|
||||
binary.Write(w, binary.BigEndian, r.Type)
|
||||
|
||||
_, err := w.Write(v)
|
||||
return err
|
||||
}
|
||||
|
||||
func (r rawValue) encodeBytes(w io.Writer, v []byte) error {
|
||||
if len(v) > maxHeaderValueLen {
|
||||
return LengthError{
|
||||
Part: "header value",
|
||||
Want: maxHeaderValueLen, Have: len(v),
|
||||
Value: v,
|
||||
}
|
||||
}
|
||||
r.Len = uint16(len(v))
|
||||
|
||||
err := binaryWriteFields(w, binary.BigEndian,
|
||||
r.Type,
|
||||
r.Len,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
_, err = w.Write(v)
|
||||
return err
|
||||
}
|
||||
|
||||
func (r rawValue) encodeString(w io.Writer, v string) error {
|
||||
if len(v) > maxHeaderValueLen {
|
||||
return LengthError{
|
||||
Part: "header value",
|
||||
Want: maxHeaderValueLen, Have: len(v),
|
||||
Value: v,
|
||||
}
|
||||
}
|
||||
r.Len = uint16(len(v))
|
||||
|
||||
type stringWriter interface {
|
||||
WriteString(string) (int, error)
|
||||
}
|
||||
|
||||
err := binaryWriteFields(w, binary.BigEndian,
|
||||
r.Type,
|
||||
r.Len,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if sw, ok := w.(stringWriter); ok {
|
||||
_, err = sw.WriteString(v)
|
||||
} else {
|
||||
_, err = w.Write([]byte(v))
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func decodeFixedBytesValue(r io.Reader, buf []byte) error {
|
||||
_, err := io.ReadFull(r, buf)
|
||||
return err
|
||||
}
|
||||
|
||||
func decodeBytesValue(r io.Reader) ([]byte, error) {
|
||||
var raw rawValue
|
||||
var err error
|
||||
raw.Len, err = decodeUint16(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
buf := make([]byte, raw.Len)
|
||||
_, err = io.ReadFull(r, buf)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return buf, nil
|
||||
}
|
||||
|
||||
func decodeStringValue(r io.Reader) (string, error) {
|
||||
v, err := decodeBytesValue(r)
|
||||
return string(v), err
|
||||
}
|
||||
|
||||
// Value represents the abstract header value.
|
||||
type Value interface {
|
||||
Get() interface{}
|
||||
String() string
|
||||
valueType() valueType
|
||||
encode(io.Writer) error
|
||||
}
|
||||
|
||||
// An BoolValue provides eventstream encoding, and representation
|
||||
// of a Go bool value.
|
||||
type BoolValue bool
|
||||
|
||||
// Get returns the underlying type
|
||||
func (v BoolValue) Get() interface{} {
|
||||
return bool(v)
|
||||
}
|
||||
|
||||
// valueType returns the EventStream header value type value.
|
||||
func (v BoolValue) valueType() valueType {
|
||||
if v {
|
||||
return trueValueType
|
||||
}
|
||||
return falseValueType
|
||||
}
|
||||
|
||||
func (v BoolValue) String() string {
|
||||
return strconv.FormatBool(bool(v))
|
||||
}
|
||||
|
||||
// encode encodes the BoolValue into an eventstream binary value
|
||||
// representation.
|
||||
func (v BoolValue) encode(w io.Writer) error {
|
||||
return binary.Write(w, binary.BigEndian, v.valueType())
|
||||
}
|
||||
|
||||
// An Int8Value provides eventstream encoding, and representation of a Go
|
||||
// int8 value.
|
||||
type Int8Value int8
|
||||
|
||||
// Get returns the underlying value.
|
||||
func (v Int8Value) Get() interface{} {
|
||||
return int8(v)
|
||||
}
|
||||
|
||||
// valueType returns the EventStream header value type value.
|
||||
func (Int8Value) valueType() valueType {
|
||||
return int8ValueType
|
||||
}
|
||||
|
||||
func (v Int8Value) String() string {
|
||||
return fmt.Sprintf("0x%02x", int8(v))
|
||||
}
|
||||
|
||||
// encode encodes the Int8Value into an eventstream binary value
|
||||
// representation.
|
||||
func (v Int8Value) encode(w io.Writer) error {
|
||||
raw := rawValue{
|
||||
Type: v.valueType(),
|
||||
}
|
||||
|
||||
return raw.encodeScalar(w, v)
|
||||
}
|
||||
|
||||
func (v *Int8Value) decode(r io.Reader) error {
|
||||
n, err := decodeUint8(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
*v = Int8Value(n)
|
||||
return nil
|
||||
}
|
||||
|
||||
// An Int16Value provides eventstream encoding, and representation of a Go
|
||||
// int16 value.
|
||||
type Int16Value int16
|
||||
|
||||
// Get returns the underlying value.
|
||||
func (v Int16Value) Get() interface{} {
|
||||
return int16(v)
|
||||
}
|
||||
|
||||
// valueType returns the EventStream header value type value.
|
||||
func (Int16Value) valueType() valueType {
|
||||
return int16ValueType
|
||||
}
|
||||
|
||||
func (v Int16Value) String() string {
|
||||
return fmt.Sprintf("0x%04x", int16(v))
|
||||
}
|
||||
|
||||
// encode encodes the Int16Value into an eventstream binary value
|
||||
// representation.
|
||||
func (v Int16Value) encode(w io.Writer) error {
|
||||
raw := rawValue{
|
||||
Type: v.valueType(),
|
||||
}
|
||||
return raw.encodeScalar(w, v)
|
||||
}
|
||||
|
||||
func (v *Int16Value) decode(r io.Reader) error {
|
||||
n, err := decodeUint16(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
*v = Int16Value(n)
|
||||
return nil
|
||||
}
|
||||
|
||||
// An Int32Value provides eventstream encoding, and representation of a Go
|
||||
// int32 value.
|
||||
type Int32Value int32
|
||||
|
||||
// Get returns the underlying value.
|
||||
func (v Int32Value) Get() interface{} {
|
||||
return int32(v)
|
||||
}
|
||||
|
||||
// valueType returns the EventStream header value type value.
|
||||
func (Int32Value) valueType() valueType {
|
||||
return int32ValueType
|
||||
}
|
||||
|
||||
func (v Int32Value) String() string {
|
||||
return fmt.Sprintf("0x%08x", int32(v))
|
||||
}
|
||||
|
||||
// encode encodes the Int32Value into an eventstream binary value
|
||||
// representation.
|
||||
func (v Int32Value) encode(w io.Writer) error {
|
||||
raw := rawValue{
|
||||
Type: v.valueType(),
|
||||
}
|
||||
return raw.encodeScalar(w, v)
|
||||
}
|
||||
|
||||
func (v *Int32Value) decode(r io.Reader) error {
|
||||
n, err := decodeUint32(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
*v = Int32Value(n)
|
||||
return nil
|
||||
}
|
||||
|
||||
// An Int64Value provides eventstream encoding, and representation of a Go
|
||||
// int64 value.
|
||||
type Int64Value int64
|
||||
|
||||
// Get returns the underlying value.
|
||||
func (v Int64Value) Get() interface{} {
|
||||
return int64(v)
|
||||
}
|
||||
|
||||
// valueType returns the EventStream header value type value.
|
||||
func (Int64Value) valueType() valueType {
|
||||
return int64ValueType
|
||||
}
|
||||
|
||||
func (v Int64Value) String() string {
|
||||
return fmt.Sprintf("0x%016x", int64(v))
|
||||
}
|
||||
|
||||
// encode encodes the Int64Value into an eventstream binary value
|
||||
// representation.
|
||||
func (v Int64Value) encode(w io.Writer) error {
|
||||
raw := rawValue{
|
||||
Type: v.valueType(),
|
||||
}
|
||||
return raw.encodeScalar(w, v)
|
||||
}
|
||||
|
||||
func (v *Int64Value) decode(r io.Reader) error {
|
||||
n, err := decodeUint64(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
*v = Int64Value(n)
|
||||
return nil
|
||||
}
|
||||
|
||||
// An BytesValue provides eventstream encoding, and representation of a Go
|
||||
// byte slice.
|
||||
type BytesValue []byte
|
||||
|
||||
// Get returns the underlying value.
|
||||
func (v BytesValue) Get() interface{} {
|
||||
return []byte(v)
|
||||
}
|
||||
|
||||
// valueType returns the EventStream header value type value.
|
||||
func (BytesValue) valueType() valueType {
|
||||
return bytesValueType
|
||||
}
|
||||
|
||||
func (v BytesValue) String() string {
|
||||
return base64.StdEncoding.EncodeToString([]byte(v))
|
||||
}
|
||||
|
||||
// encode encodes the BytesValue into an eventstream binary value
|
||||
// representation.
|
||||
func (v BytesValue) encode(w io.Writer) error {
|
||||
raw := rawValue{
|
||||
Type: v.valueType(),
|
||||
}
|
||||
|
||||
return raw.encodeBytes(w, []byte(v))
|
||||
}
|
||||
|
||||
func (v *BytesValue) decode(r io.Reader) error {
|
||||
buf, err := decodeBytesValue(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
*v = BytesValue(buf)
|
||||
return nil
|
||||
}
|
||||
|
||||
// An StringValue provides eventstream encoding, and representation of a Go
|
||||
// string.
|
||||
type StringValue string
|
||||
|
||||
// Get returns the underlying value.
|
||||
func (v StringValue) Get() interface{} {
|
||||
return string(v)
|
||||
}
|
||||
|
||||
// valueType returns the EventStream header value type value.
|
||||
func (StringValue) valueType() valueType {
|
||||
return stringValueType
|
||||
}
|
||||
|
||||
func (v StringValue) String() string {
|
||||
return string(v)
|
||||
}
|
||||
|
||||
// encode encodes the StringValue into an eventstream binary value
|
||||
// representation.
|
||||
func (v StringValue) encode(w io.Writer) error {
|
||||
raw := rawValue{
|
||||
Type: v.valueType(),
|
||||
}
|
||||
|
||||
return raw.encodeString(w, string(v))
|
||||
}
|
||||
|
||||
func (v *StringValue) decode(r io.Reader) error {
|
||||
s, err := decodeStringValue(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
*v = StringValue(s)
|
||||
return nil
|
||||
}
|
||||
|
||||
// An TimestampValue provides eventstream encoding, and representation of a Go
|
||||
// timestamp.
|
||||
type TimestampValue time.Time
|
||||
|
||||
// Get returns the underlying value.
|
||||
func (v TimestampValue) Get() interface{} {
|
||||
return time.Time(v)
|
||||
}
|
||||
|
||||
// valueType returns the EventStream header value type value.
|
||||
func (TimestampValue) valueType() valueType {
|
||||
return timestampValueType
|
||||
}
|
||||
|
||||
func (v TimestampValue) epochMilli() int64 {
|
||||
nano := time.Time(v).UnixNano()
|
||||
msec := nano / int64(time.Millisecond)
|
||||
return msec
|
||||
}
|
||||
|
||||
func (v TimestampValue) String() string {
|
||||
msec := v.epochMilli()
|
||||
return strconv.FormatInt(msec, 10)
|
||||
}
|
||||
|
||||
// encode encodes the TimestampValue into an eventstream binary value
|
||||
// representation.
|
||||
func (v TimestampValue) encode(w io.Writer) error {
|
||||
raw := rawValue{
|
||||
Type: v.valueType(),
|
||||
}
|
||||
|
||||
msec := v.epochMilli()
|
||||
return raw.encodeScalar(w, msec)
|
||||
}
|
||||
|
||||
func (v *TimestampValue) decode(r io.Reader) error {
|
||||
n, err := decodeUint64(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
*v = TimestampValue(timeFromEpochMilli(int64(n)))
|
||||
return nil
|
||||
}
|
||||
|
||||
func timeFromEpochMilli(t int64) time.Time {
|
||||
secs := t / 1e3
|
||||
msec := t % 1e3
|
||||
return time.Unix(secs, msec*int64(time.Millisecond)).UTC()
|
||||
}
|
||||
|
||||
// An UUIDValue provides eventstream encoding, and representation of a UUID
|
||||
// value.
|
||||
type UUIDValue [16]byte
|
||||
|
||||
// Get returns the underlying value.
|
||||
func (v UUIDValue) Get() interface{} {
|
||||
return v[:]
|
||||
}
|
||||
|
||||
// valueType returns the EventStream header value type value.
|
||||
func (UUIDValue) valueType() valueType {
|
||||
return uuidValueType
|
||||
}
|
||||
|
||||
func (v UUIDValue) String() string {
|
||||
return fmt.Sprintf(`%X-%X-%X-%X-%X`, v[0:4], v[4:6], v[6:8], v[8:10], v[10:])
|
||||
}
|
||||
|
||||
// encode encodes the UUIDValue into an eventstream binary value
|
||||
// representation.
|
||||
func (v UUIDValue) encode(w io.Writer) error {
|
||||
raw := rawValue{
|
||||
Type: v.valueType(),
|
||||
}
|
||||
|
||||
return raw.encodeFixedSlice(w, v[:])
|
||||
}
|
||||
|
||||
func (v *UUIDValue) decode(r io.Reader) error {
|
||||
tv := (*v)[:]
|
||||
return decodeFixedBytesValue(r, tv)
|
||||
}
|
||||
Generated
Vendored
+203
@@ -0,0 +1,203 @@
|
||||
package eventstream
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"io"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func binWrite(v interface{}) []byte {
|
||||
var w bytes.Buffer
|
||||
binary.Write(&w, binary.BigEndian, v)
|
||||
return w.Bytes()
|
||||
}
|
||||
|
||||
var testValueEncodingCases = []struct {
|
||||
Val Value
|
||||
Expect []byte
|
||||
Decode func(io.Reader) (Value, error)
|
||||
}{
|
||||
{
|
||||
BoolValue(true),
|
||||
[]byte{byte(trueValueType)},
|
||||
nil,
|
||||
},
|
||||
{
|
||||
BoolValue(false),
|
||||
[]byte{byte(falseValueType)},
|
||||
nil,
|
||||
},
|
||||
{
|
||||
Int8Value(0x0f),
|
||||
[]byte{byte(int8ValueType), 0x0f},
|
||||
func(r io.Reader) (Value, error) {
|
||||
var v Int8Value
|
||||
err := v.decode(r)
|
||||
return v, err
|
||||
},
|
||||
},
|
||||
{
|
||||
Int16Value(0x0f),
|
||||
append([]byte{byte(int16ValueType)}, binWrite(int16(0x0f))...),
|
||||
func(r io.Reader) (Value, error) {
|
||||
var v Int16Value
|
||||
err := v.decode(r)
|
||||
return v, err
|
||||
},
|
||||
},
|
||||
{
|
||||
Int32Value(0x0f),
|
||||
append([]byte{byte(int32ValueType)}, binWrite(int32(0x0f))...),
|
||||
func(r io.Reader) (Value, error) {
|
||||
var v Int32Value
|
||||
err := v.decode(r)
|
||||
return v, err
|
||||
},
|
||||
},
|
||||
{
|
||||
Int64Value(0x0f),
|
||||
append([]byte{byte(int64ValueType)}, binWrite(int64(0x0f))...),
|
||||
func(r io.Reader) (Value, error) {
|
||||
var v Int64Value
|
||||
err := v.decode(r)
|
||||
return v, err
|
||||
},
|
||||
},
|
||||
{
|
||||
BytesValue([]byte{0, 1, 2, 3}),
|
||||
[]byte{byte(bytesValueType), 0x00, 0x04, 0, 1, 2, 3},
|
||||
func(r io.Reader) (Value, error) {
|
||||
var v BytesValue
|
||||
err := v.decode(r)
|
||||
return v, err
|
||||
},
|
||||
},
|
||||
{
|
||||
StringValue("abc123"),
|
||||
append([]byte{byte(stringValueType), 0, 6}, []byte("abc123")...),
|
||||
func(r io.Reader) (Value, error) {
|
||||
var v StringValue
|
||||
err := v.decode(r)
|
||||
return v, err
|
||||
},
|
||||
},
|
||||
{
|
||||
TimestampValue(
|
||||
time.Date(2014, 04, 04, 0, 1, 0, 0, time.FixedZone("PDT", -7)),
|
||||
),
|
||||
append([]byte{byte(timestampValueType)}, binWrite(int64(1396569667000))...),
|
||||
func(r io.Reader) (Value, error) {
|
||||
var v TimestampValue
|
||||
err := v.decode(r)
|
||||
return v, err
|
||||
},
|
||||
},
|
||||
{
|
||||
UUIDValue(
|
||||
[16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0xa, 0xb, 0xc, 0xd, 0xe, 0xf},
|
||||
),
|
||||
[]byte{byte(uuidValueType), 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0xa, 0xb, 0xc, 0xd, 0xe, 0xf},
|
||||
func(r io.Reader) (Value, error) {
|
||||
var v UUIDValue
|
||||
err := v.decode(r)
|
||||
return v, err
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
func TestValue_MarshalValue(t *testing.T) {
|
||||
for i, c := range testValueEncodingCases {
|
||||
var w bytes.Buffer
|
||||
|
||||
if err := c.Val.encode(&w); err != nil {
|
||||
t.Fatalf("%d, expect no error, got %v", i, err)
|
||||
}
|
||||
|
||||
if e, a := c.Expect, w.Bytes(); !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("%d, expect %v, got %v", i, e, a)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHeader_DecodeValues(t *testing.T) {
|
||||
for i, c := range testValueEncodingCases {
|
||||
r := bytes.NewBuffer(c.Expect)
|
||||
v, err := decodeHeaderValue(r)
|
||||
if err != nil {
|
||||
t.Fatalf("%d, expect no error, got %v", i, err)
|
||||
}
|
||||
|
||||
switch tv := v.(type) {
|
||||
case TimestampValue:
|
||||
exp := time.Time(c.Val.(TimestampValue))
|
||||
if e, a := exp, time.Time(tv); !e.Equal(a) {
|
||||
t.Errorf("%d, expect %v, got %v", i, e, a)
|
||||
}
|
||||
default:
|
||||
if e, a := c.Val, v; !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("%d, expect %v, got %v", i, e, a)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestValue_Decode(t *testing.T) {
|
||||
for i, c := range testValueEncodingCases {
|
||||
if c.Decode == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
r := bytes.NewBuffer(c.Expect)
|
||||
r.ReadByte() // strip off Type field
|
||||
|
||||
v, err := c.Decode(r)
|
||||
if err != nil {
|
||||
t.Fatalf("%d, expect no error, got %v", i, err)
|
||||
}
|
||||
|
||||
switch tv := v.(type) {
|
||||
case TimestampValue:
|
||||
exp := time.Time(c.Val.(TimestampValue))
|
||||
if e, a := exp, time.Time(tv); !e.Equal(a) {
|
||||
t.Errorf("%d, expect %v, got %v", i, e, a)
|
||||
}
|
||||
default:
|
||||
if e, a := c.Val, v; !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("%d, expect %v, got %v", i, e, a)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestValue_String(t *testing.T) {
|
||||
cases := []struct {
|
||||
Val Value
|
||||
Expect string
|
||||
}{
|
||||
{BoolValue(true), "true"},
|
||||
{BoolValue(false), "false"},
|
||||
{Int8Value(0x0f), "0x0f"},
|
||||
{Int16Value(0x0f), "0x000f"},
|
||||
{Int32Value(0x0f), "0x0000000f"},
|
||||
{Int64Value(0x0f), "0x000000000000000f"},
|
||||
{BytesValue([]byte{0, 1, 2, 3}), "AAECAw=="},
|
||||
{StringValue("abc123"), "abc123"},
|
||||
{TimestampValue(
|
||||
time.Date(2014, 04, 04, 0, 1, 0, 0, time.FixedZone("PDT", -7)),
|
||||
),
|
||||
"1396569667000",
|
||||
},
|
||||
{UUIDValue([16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0xa, 0xb, 0xc, 0xd, 0xe, 0xf}),
|
||||
"00010203-0405-0607-0809-0A0B0C0D0E0F",
|
||||
},
|
||||
}
|
||||
|
||||
for i, c := range cases {
|
||||
if e, a := c.Expect, c.Val.String(); e != a {
|
||||
t.Errorf("%d, expect %v, got %v", i, e, a)
|
||||
}
|
||||
}
|
||||
}
|
||||
+103
@@ -0,0 +1,103 @@
|
||||
package eventstream
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"hash/crc32"
|
||||
)
|
||||
|
||||
const preludeLen = 8
|
||||
const preludeCRCLen = 4
|
||||
const msgCRCLen = 4
|
||||
const minMsgLen = preludeLen + preludeCRCLen + msgCRCLen
|
||||
const maxPayloadLen = 1024 * 1024 * 16 // 16MB
|
||||
const maxHeadersLen = 1024 * 128 // 128KB
|
||||
const maxMsgLen = minMsgLen + maxHeadersLen + maxPayloadLen
|
||||
|
||||
var crc32IEEETable = crc32.MakeTable(crc32.IEEE)
|
||||
|
||||
// A Message provides the eventstream message representation.
|
||||
type Message struct {
|
||||
Headers Headers
|
||||
Payload []byte
|
||||
}
|
||||
|
||||
func (m *Message) rawMessage() (rawMessage, error) {
|
||||
var raw rawMessage
|
||||
|
||||
if len(m.Headers) > 0 {
|
||||
var headers bytes.Buffer
|
||||
if err := encodeHeaders(&headers, m.Headers); err != nil {
|
||||
return rawMessage{}, err
|
||||
}
|
||||
raw.Headers = headers.Bytes()
|
||||
raw.HeadersLen = uint32(len(raw.Headers))
|
||||
}
|
||||
|
||||
raw.Length = raw.HeadersLen + uint32(len(m.Payload)) + minMsgLen
|
||||
|
||||
hash := crc32.New(crc32IEEETable)
|
||||
binaryWriteFields(hash, binary.BigEndian, raw.Length, raw.HeadersLen)
|
||||
raw.PreludeCRC = hash.Sum32()
|
||||
|
||||
binaryWriteFields(hash, binary.BigEndian, raw.PreludeCRC)
|
||||
|
||||
if raw.HeadersLen > 0 {
|
||||
hash.Write(raw.Headers)
|
||||
}
|
||||
|
||||
// Read payload bytes and update hash for it as well.
|
||||
if len(m.Payload) > 0 {
|
||||
raw.Payload = m.Payload
|
||||
hash.Write(raw.Payload)
|
||||
}
|
||||
|
||||
raw.CRC = hash.Sum32()
|
||||
|
||||
return raw, nil
|
||||
}
|
||||
|
||||
type messagePrelude struct {
|
||||
Length uint32
|
||||
HeadersLen uint32
|
||||
PreludeCRC uint32
|
||||
}
|
||||
|
||||
func (p messagePrelude) PayloadLen() uint32 {
|
||||
return p.Length - p.HeadersLen - minMsgLen
|
||||
}
|
||||
|
||||
func (p messagePrelude) ValidateLens() error {
|
||||
if p.Length == 0 || p.Length > maxMsgLen {
|
||||
return LengthError{
|
||||
Part: "message prelude",
|
||||
Want: maxMsgLen,
|
||||
Have: int(p.Length),
|
||||
}
|
||||
}
|
||||
if p.HeadersLen > maxHeadersLen {
|
||||
return LengthError{
|
||||
Part: "message headers",
|
||||
Want: maxHeadersLen,
|
||||
Have: int(p.HeadersLen),
|
||||
}
|
||||
}
|
||||
if payloadLen := p.PayloadLen(); payloadLen > maxPayloadLen {
|
||||
return LengthError{
|
||||
Part: "message payload",
|
||||
Want: maxPayloadLen,
|
||||
Have: int(payloadLen),
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type rawMessage struct {
|
||||
messagePrelude
|
||||
|
||||
Headers []byte
|
||||
Payload []byte
|
||||
|
||||
CRC uint32
|
||||
}
|
||||
+152
@@ -0,0 +1,152 @@
|
||||
package eventstream
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
type testCase struct {
|
||||
Name string
|
||||
Encoded []byte
|
||||
Decoded decodedMessage
|
||||
}
|
||||
|
||||
type testErrorCase struct {
|
||||
Name string
|
||||
Encoded []byte
|
||||
Err string
|
||||
}
|
||||
|
||||
type rawTestCase struct {
|
||||
Name string
|
||||
Encoded, Decoded []byte
|
||||
}
|
||||
|
||||
func readRawTestCases(root, class string) (map[string]rawTestCase, error) {
|
||||
encoded, err := readTests(filepath.Join(root, "encoded", class))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
decoded, err := readTests(filepath.Join(root, "decoded", class))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if len(encoded) == 0 {
|
||||
return nil, fmt.Errorf("expect encoded cases, found none")
|
||||
}
|
||||
|
||||
if len(encoded) != len(decoded) {
|
||||
return nil, fmt.Errorf("encoded and decoded sets different")
|
||||
}
|
||||
|
||||
rawCases := map[string]rawTestCase{}
|
||||
for name, encData := range encoded {
|
||||
decData, ok := decoded[name]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("encoded %q case not found in decoded set", name)
|
||||
}
|
||||
|
||||
rawCases[name] = rawTestCase{
|
||||
Name: name,
|
||||
Encoded: encData,
|
||||
Decoded: decData,
|
||||
}
|
||||
}
|
||||
|
||||
return rawCases, nil
|
||||
}
|
||||
|
||||
func readNegativeTests(root string) ([]testErrorCase, error) {
|
||||
rawCases, err := readRawTestCases(root, "negative")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cases := make([]testErrorCase, 0, len(rawCases))
|
||||
for name, rawCase := range rawCases {
|
||||
cases = append(cases, testErrorCase{
|
||||
Name: name,
|
||||
Encoded: rawCase.Encoded,
|
||||
Err: string(rawCase.Decoded),
|
||||
})
|
||||
}
|
||||
|
||||
return cases, nil
|
||||
}
|
||||
|
||||
func readPositiveTests(root string) ([]testCase, error) {
|
||||
rawCases, err := readRawTestCases(root, "positive")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cases := make([]testCase, 0, len(rawCases))
|
||||
for name, rawCase := range rawCases {
|
||||
|
||||
var dec decodedMessage
|
||||
if err := json.Unmarshal(rawCase.Decoded, &dec); err != nil {
|
||||
return nil, fmt.Errorf("failed to decode %q, %v", name, err)
|
||||
}
|
||||
|
||||
cases = append(cases, testCase{
|
||||
Name: name,
|
||||
Encoded: rawCase.Encoded,
|
||||
Decoded: dec,
|
||||
})
|
||||
}
|
||||
|
||||
return cases, nil
|
||||
}
|
||||
|
||||
func readTests(root string) (map[string][]byte, error) {
|
||||
items, err := ioutil.ReadDir(root)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read test suite %q dirs, %v", root, err)
|
||||
}
|
||||
|
||||
cases := map[string][]byte{}
|
||||
for _, item := range items {
|
||||
if item.IsDir() {
|
||||
continue
|
||||
}
|
||||
|
||||
filename := filepath.Join(root, item.Name())
|
||||
data, err := ioutil.ReadFile(filename)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read test_data file %q, %v", filename, err)
|
||||
}
|
||||
|
||||
cases[item.Name()] = data
|
||||
}
|
||||
|
||||
return cases, nil
|
||||
}
|
||||
|
||||
func compareLines(t *testing.T, a, b []byte) bool {
|
||||
as := bufio.NewScanner(bytes.NewBuffer(a))
|
||||
bs := bufio.NewScanner(bytes.NewBuffer(b))
|
||||
|
||||
var failed bool
|
||||
for {
|
||||
if ab, bb := as.Scan(), bs.Scan(); ab != bb {
|
||||
t.Errorf("expect a & b to have same number of lines")
|
||||
return false
|
||||
} else if !ab {
|
||||
break
|
||||
}
|
||||
|
||||
if v1, v2 := as.Text(), bs.Text(); v1 != v2 {
|
||||
t.Errorf("expect %q to be %q", v1, v2)
|
||||
failed = true
|
||||
}
|
||||
}
|
||||
|
||||
return !failed
|
||||
}
|
||||
Generated
Vendored
+1
@@ -0,0 +1 @@
|
||||
Prelude checksum mismatch
|
||||
Generated
Vendored
+1
@@ -0,0 +1 @@
|
||||
Message checksum mismatch
|
||||
Generated
Vendored
+1
@@ -0,0 +1 @@
|
||||
Prelude checksum mismatch
|
||||
Generated
Vendored
+1
@@ -0,0 +1 @@
|
||||
Message checksum mismatch
|
||||
Generated
Vendored
+58
@@ -0,0 +1,58 @@
|
||||
{
|
||||
"total_length": 204,
|
||||
"headers_length": 175,
|
||||
"prelude_crc": 263087306,
|
||||
"headers": [ {
|
||||
"name": "event-type",
|
||||
"type": 4,
|
||||
"value": 40972
|
||||
},
|
||||
{
|
||||
"name": "content-type",
|
||||
"type": 7,
|
||||
"value": "YXBwbGljYXRpb24vanNvbg=="
|
||||
},
|
||||
{
|
||||
"name": "bool false",
|
||||
"type": 1,
|
||||
"value": false
|
||||
},
|
||||
{
|
||||
"name": "bool true",
|
||||
"type": 0,
|
||||
"value": true
|
||||
},
|
||||
{
|
||||
"name": "byte",
|
||||
"type": 2,
|
||||
"value": -49
|
||||
},
|
||||
{
|
||||
"name": "byte buf",
|
||||
"type": 6,
|
||||
"value": "SSdtIGEgbGl0dGxlIHRlYXBvdCE="
|
||||
},
|
||||
{
|
||||
"name": "timestamp",
|
||||
"type": 8,
|
||||
"value": 8675309
|
||||
},
|
||||
{
|
||||
"name": "int16",
|
||||
"type": 3,
|
||||
"value": 42
|
||||
},
|
||||
{
|
||||
"name": "int64",
|
||||
"type": 5,
|
||||
"value": 42424242
|
||||
},
|
||||
{
|
||||
"name": "uuid",
|
||||
"type": 9,
|
||||
"value": "AQIDBAUGBwgJCgsMDQ4PEA=="
|
||||
}
|
||||
],
|
||||
"payload": "eydmb28nOidiYXInfQ==",
|
||||
"message_crc": -1415188212
|
||||
}
|
||||
Generated
Vendored
+8
@@ -0,0 +1,8 @@
|
||||
{
|
||||
"total_length": 16,
|
||||
"headers_length": 0,
|
||||
"prelude_crc": 96618731,
|
||||
"headers": [ ],
|
||||
"payload": "",
|
||||
"message_crc": 2107164927
|
||||
}
|
||||
vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/testdata/decoded/positive/int32_header
Generated
Vendored
+13
@@ -0,0 +1,13 @@
|
||||
{
|
||||
"total_length": 45,
|
||||
"headers_length": 16,
|
||||
"prelude_crc": 1103373496,
|
||||
"headers": [ {
|
||||
"name": "event-type",
|
||||
"type": 4,
|
||||
"value": 40972
|
||||
}
|
||||
],
|
||||
"payload": "eydmb28nOidiYXInfQ==",
|
||||
"message_crc": 921993376
|
||||
}
|
||||
Generated
Vendored
+8
@@ -0,0 +1,8 @@
|
||||
{
|
||||
"total_length": 29,
|
||||
"headers_length": 0,
|
||||
"prelude_crc": -44921766,
|
||||
"headers": [ ],
|
||||
"payload": "eydmb28nOidiYXInfQ==",
|
||||
"message_crc": -1016776394
|
||||
}
|
||||
Generated
Vendored
+13
@@ -0,0 +1,13 @@
|
||||
{
|
||||
"total_length": 61,
|
||||
"headers_length": 32,
|
||||
"prelude_crc": 134054806,
|
||||
"headers": [ {
|
||||
"name": "content-type",
|
||||
"type": 7,
|
||||
"value": "YXBwbGljYXRpb24vanNvbg=="
|
||||
}
|
||||
],
|
||||
"payload": "eydmb28nOidiYXInfQ==",
|
||||
"message_crc": -1919153999
|
||||
}
|
||||
Generated
Vendored
BIN
Binary file not shown.
Generated
Vendored
BIN
Binary file not shown.
Generated
Vendored
BIN
Binary file not shown.
Generated
Vendored
BIN
Binary file not shown.
Generated
Vendored
BIN
Binary file not shown.
Generated
Vendored
BIN
Binary file not shown.
vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/testdata/encoded/positive/int32_header
Generated
Vendored
BIN
Binary file not shown.
Generated
Vendored
BIN
Binary file not shown.
Generated
Vendored
BIN
Binary file not shown.
Reference in New Issue
Block a user