Bump aws sdk to v1.15.11

This is the latest official release for this dependency

Signed-off-by: Ryan Abrams <rdabrams@gmail.com>
This commit is contained in:
Ryan Abrams 2018-08-14 12:13:40 -07:00
parent 5f37adaa41
commit 57212c909b
67 changed files with 5178 additions and 916 deletions

View file

@ -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))
}
}

View file

@ -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
}

View file

@ -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
}

View file

@ -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"
}

View file

@ -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
}

View file

@ -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
}

View file

@ -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
}

View file

@ -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)
}

View file

@ -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
}

View file

@ -0,0 +1,81 @@
package protocol
import (
"io"
"io/ioutil"
"net/http"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/client/metadata"
"github.com/aws/aws-sdk-go/aws/request"
)
// PayloadUnmarshaler provides the interface for unmarshaling a payload's
// reader into a SDK shape.
type PayloadUnmarshaler interface {
UnmarshalPayload(io.Reader, interface{}) error
}
// HandlerPayloadUnmarshal implements the PayloadUnmarshaler from a
// HandlerList. This provides the support for unmarshaling a payload reader to
// a shape without needing a SDK request first.
type HandlerPayloadUnmarshal struct {
Unmarshalers request.HandlerList
}
// UnmarshalPayload unmarshals the io.Reader payload into the SDK shape using
// the Unmarshalers HandlerList provided. Returns an error if unable
// unmarshaling fails.
func (h HandlerPayloadUnmarshal) UnmarshalPayload(r io.Reader, v interface{}) error {
req := &request.Request{
HTTPRequest: &http.Request{},
HTTPResponse: &http.Response{
StatusCode: 200,
Header: http.Header{},
Body: ioutil.NopCloser(r),
},
Data: v,
}
h.Unmarshalers.Run(req)
return req.Error
}
// PayloadMarshaler provides the interface for marshaling a SDK shape into and
// io.Writer.
type PayloadMarshaler interface {
MarshalPayload(io.Writer, interface{}) error
}
// HandlerPayloadMarshal implements the PayloadMarshaler from a HandlerList.
// This provides support for marshaling a SDK shape into an io.Writer without
// needing a SDK request first.
type HandlerPayloadMarshal struct {
Marshalers request.HandlerList
}
// MarshalPayload marshals the SDK shape into the io.Writer using the
// Marshalers HandlerList provided. Returns an error if unable if marshal
// fails.
func (h HandlerPayloadMarshal) MarshalPayload(w io.Writer, v interface{}) error {
req := request.New(
aws.Config{},
metadata.ClientInfo{},
request.Handlers{},
nil,
&request.Operation{HTTPMethod: "GET"},
v,
nil,
)
h.Marshalers.Run(req)
if req.Error != nil {
return req.Error
}
io.Copy(w, req.GetBody())
return nil
}

View file

@ -25,7 +25,7 @@ func Build(r *request.Request) {
return
}
if r.ExpireTime == 0 {
if !r.IsPresigned() {
r.HTTPRequest.Method = "POST"
r.HTTPRequest.Header.Set("Content-Type", "application/x-www-form-urlencoded; charset=utf-8")
r.SetBufferBody([]byte(body.Encode()))

View file

@ -233,7 +233,12 @@ func (q *queryParser) parseScalar(v url.Values, r reflect.Value, name string, ta
v.Set(name, strconv.FormatFloat(float64(value), 'f', -1, 32))
case time.Time:
const ISO8601UTC = "2006-01-02T15:04:05Z"
v.Set(name, value.UTC().Format(ISO8601UTC))
format := tag.Get("timestampFormat")
if len(format) == 0 {
format = protocol.ISO8601TimeFormatName
}
v.Set(name, protocol.FormatTime(format, value))
default:
return fmt.Errorf("unsupported value for param %s: %v (%s)", name, r.Interface(), r.Type().Name())
}

View file

@ -20,9 +20,6 @@ import (
"github.com/aws/aws-sdk-go/private/protocol"
)
// RFC822 returns an RFC822 formatted timestamp for AWS protocols
const RFC822 = "Mon, 2 Jan 2006 15:04:05 GMT"
// Whether the byte value can be sent without escaping in AWS URLs
var noEscape [256]bool
@ -270,7 +267,14 @@ func convertType(v reflect.Value, tag reflect.StructTag) (str string, err error)
case float64:
str = strconv.FormatFloat(value, 'f', -1, 64)
case time.Time:
str = value.UTC().Format(RFC822)
format := tag.Get("timestampFormat")
if len(format) == 0 {
format = protocol.RFC822TimeFormatName
if tag.Get("location") == "querystring" {
format = protocol.ISO8601TimeFormatName
}
}
str = protocol.FormatTime(format, value)
case aws.JSONValue:
if len(value) == 0 {
return "", errValueNotSet

View file

@ -198,7 +198,11 @@ func unmarshalHeader(v reflect.Value, header string, tag reflect.StructTag) erro
}
v.Set(reflect.ValueOf(&f))
case *time.Time:
t, err := time.Parse(RFC822, header)
format := tag.Get("timestampFormat")
if len(format) == 0 {
format = protocol.RFC822TimeFormatName
}
t, err := protocol.ParseTime(format, header)
if err != nil {
return err
}

View file

@ -0,0 +1,72 @@
package protocol
import (
"strconv"
"time"
)
// Names of time formats supported by the SDK
const (
RFC822TimeFormatName = "rfc822"
ISO8601TimeFormatName = "iso8601"
UnixTimeFormatName = "unixTimestamp"
)
// Time formats supported by the SDK
const (
// RFC 7231#section-7.1.1.1 timetamp format. e.g Tue, 29 Apr 2014 18:30:38 GMT
RFC822TimeFormat = "Mon, 2 Jan 2006 15:04:05 GMT"
// RFC3339 a subset of the ISO8601 timestamp format. e.g 2014-04-29T18:30:38Z
ISO8601TimeFormat = "2006-01-02T15:04:05Z"
)
// IsKnownTimestampFormat returns if the timestamp format name
// is know to the SDK's protocols.
func IsKnownTimestampFormat(name string) bool {
switch name {
case RFC822TimeFormatName:
fallthrough
case ISO8601TimeFormatName:
fallthrough
case UnixTimeFormatName:
return true
default:
return false
}
}
// FormatTime returns a string value of the time.
func FormatTime(name string, t time.Time) string {
t = t.UTC()
switch name {
case RFC822TimeFormatName:
return t.Format(RFC822TimeFormat)
case ISO8601TimeFormatName:
return t.Format(ISO8601TimeFormat)
case UnixTimeFormatName:
return strconv.FormatInt(t.Unix(), 10)
default:
panic("unknown timestamp format name, " + name)
}
}
// ParseTime attempts to parse the time given the format. Returns
// the time if it was able to be parsed, and fails otherwise.
func ParseTime(formatName, value string) (time.Time, error) {
switch formatName {
case RFC822TimeFormatName:
return time.Parse(RFC822TimeFormat, value)
case ISO8601TimeFormatName:
return time.Parse(ISO8601TimeFormat, value)
case UnixTimeFormatName:
v, err := strconv.ParseFloat(value, 64)
if err != nil {
return time.Time{}, err
}
return time.Unix(int64(v), 0), nil
default:
panic("unknown timestamp format name, " + formatName)
}
}

View file

@ -13,9 +13,13 @@ import (
"github.com/aws/aws-sdk-go/private/protocol"
)
// BuildXML will serialize params into an xml.Encoder.
// Error will be returned if the serialization of any of the params or nested values fails.
// BuildXML will serialize params into an xml.Encoder. Error will be returned
// if the serialization of any of the params or nested values fails.
func BuildXML(params interface{}, e *xml.Encoder) error {
return buildXML(params, e, false)
}
func buildXML(params interface{}, e *xml.Encoder, sorted bool) error {
b := xmlBuilder{encoder: e, namespaces: map[string]string{}}
root := NewXMLElement(xml.Name{})
if err := b.buildValue(reflect.ValueOf(params), root, ""); err != nil {
@ -23,7 +27,7 @@ func BuildXML(params interface{}, e *xml.Encoder) error {
}
for _, c := range root.Children {
for _, v := range c {
return StructToXML(e, v, false)
return StructToXML(e, v, sorted)
}
}
return nil
@ -90,8 +94,6 @@ func (b *xmlBuilder) buildStruct(value reflect.Value, current *XMLNode, tag refl
return nil
}
fieldAdded := false
// unwrap payloads
if payload := tag.Get("payload"); payload != "" {
field, _ := value.Type().FieldByName(payload)
@ -119,6 +121,8 @@ func (b *xmlBuilder) buildStruct(value reflect.Value, current *XMLNode, tag refl
child.Attr = append(child.Attr, ns)
}
var payloadFields, nonPayloadFields int
t := value.Type()
for i := 0; i < value.NumField(); i++ {
member := elemOf(value.Field(i))
@ -133,8 +137,10 @@ func (b *xmlBuilder) buildStruct(value reflect.Value, current *XMLNode, tag refl
mTag := field.Tag
if mTag.Get("location") != "" { // skip non-body members
nonPayloadFields++
continue
}
payloadFields++
if protocol.CanSetIdempotencyToken(value.Field(i), field) {
token := protocol.GetIdempotencyToken()
@ -149,11 +155,11 @@ func (b *xmlBuilder) buildStruct(value reflect.Value, current *XMLNode, tag refl
if err := b.buildValue(member, child, mTag); err != nil {
return err
}
fieldAdded = true
}
if fieldAdded { // only append this child if we have one ore more valid members
// Only case where the child shape is not added is if the shape only contains
// non-payload fields, e.g headers/query.
if !(payloadFields == 0 && nonPayloadFields > 0) {
current.AddChild(child)
}
@ -278,8 +284,12 @@ func (b *xmlBuilder) buildScalar(value reflect.Value, current *XMLNode, tag refl
case float32:
str = strconv.FormatFloat(float64(converted), 'f', -1, 32)
case time.Time:
const ISO8601UTC = "2006-01-02T15:04:05Z"
str = converted.UTC().Format(ISO8601UTC)
format := tag.Get("timestampFormat")
if len(format) == 0 {
format = protocol.ISO8601TimeFormatName
}
str = protocol.FormatTime(format, converted)
default:
return fmt.Errorf("unsupported value for param %s: %v (%s)",
tag.Get("locationName"), value.Interface(), value.Type().Name())

View file

@ -9,6 +9,8 @@ import (
"strconv"
"strings"
"time"
"github.com/aws/aws-sdk-go/private/protocol"
)
// UnmarshalXML deserializes an xml.Decoder into the container v. V
@ -52,9 +54,15 @@ func parse(r reflect.Value, node *XMLNode, tag reflect.StructTag) error {
if t == "" {
switch rtype.Kind() {
case reflect.Struct:
t = "structure"
// also it can't be a time object
if _, ok := r.Interface().(*time.Time); !ok {
t = "structure"
}
case reflect.Slice:
t = "list"
// also it can't be a byte slice
if _, ok := r.Interface().([]byte); !ok {
t = "list"
}
case reflect.Map:
t = "map"
}
@ -247,8 +255,12 @@ func parseScalar(r reflect.Value, node *XMLNode, tag reflect.StructTag) error {
}
r.Set(reflect.ValueOf(&v))
case *time.Time:
const ISO8601UTC = "2006-01-02T15:04:05Z"
t, err := time.Parse(ISO8601UTC, node.Text)
format := tag.Get("timestampFormat")
if len(format) == 0 {
format = protocol.ISO8601TimeFormatName
}
t, err := protocol.ParseTime(format, node.Text)
if err != nil {
return err
}

View file

@ -29,6 +29,7 @@ func NewXMLElement(name xml.Name) *XMLNode {
// AddChild adds child to the XMLNode.
func (n *XMLNode) AddChild(child *XMLNode) {
child.parent = n
if _, ok := n.Children[child.Name.Local]; !ok {
n.Children[child.Name.Local] = []*XMLNode{}
}