Bump AWS SDK

Fixes https://github.com/docker/distribution/issues/3097

Signed-off-by: Yong Wen Chua <lawliet89@users.noreply.github.com>

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/debug.go

@@ -101,7 +101,7 @@ func (hs *decodedHeaders) UnmarshalJSON(b []byte) error {
 		}
 		headers.Set(h.Name, value)
 	}
-	(*hs) = decodedHeaders(headers)
+	*hs = decodedHeaders(headers)
 
 	return nil
 }

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/decode.go

@@ -21,10 +21,24 @@ type Decoder struct {
 
 // NewDecoder initializes and returns a Decoder for decoding event
 // stream messages from the reader provided.
-func NewDecoder(r io.Reader) *Decoder {
-	return &Decoder{
+func NewDecoder(r io.Reader, opts ...func(*Decoder)) *Decoder {
+	d := &Decoder{
 		r: r,
 	}
+
+	for _, opt := range opts {
+		opt(d)
+	}
+
+	return d
+}
+
+// DecodeWithLogger adds a logger to be used by the decoder when decoding
+// stream events.
+func DecodeWithLogger(logger aws.Logger) func(*Decoder) {
+	return func(d *Decoder) {
+		d.logger = logger
+	}
 }
 
 // Decode attempts to decode a single message from the event stream reader.
@@ -40,6 +54,15 @@ func (d *Decoder) Decode(payloadBuf []byte) (m Message, err error) {
 		}()
 	}
 
+	m, err = Decode(reader, payloadBuf)
+
+	return m, err
+}
+
+// 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 reader.
+func Decode(reader io.Reader, payloadBuf []byte) (m Message, err error) {
 	crc := crc32.New(crc32IEEETable)
 	hashReader := io.TeeReader(reader, crc)
 
@@ -72,12 +95,6 @@ func (d *Decoder) Decode(payloadBuf []byte) (m Message, err error) {
 	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()) }()

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/encode.go

@@ -3,61 +3,107 @@ package eventstream
 import (
 	"bytes"
 	"encoding/binary"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
 	"hash"
 	"hash/crc32"
 	"io"
+
+	"github.com/aws/aws-sdk-go/aws"
 )
 
 // Encoder provides EventStream message encoding.
 type Encoder struct {
-	w io.Writer
+	w      io.Writer
+	logger aws.Logger
 
 	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{
+func NewEncoder(w io.Writer, opts ...func(*Encoder)) *Encoder {
+	e := &Encoder{
 		w:          w,
 		headersBuf: bytes.NewBuffer(nil),
 	}
+
+	for _, opt := range opts {
+		opt(e)
+	}
+
+	return e
+}
+
+// EncodeWithLogger adds a logger to be used by the encode when decoding
+// stream events.
+func EncodeWithLogger(logger aws.Logger) func(*Encoder) {
+	return func(d *Encoder) {
+		d.logger = logger
+	}
 }
 
 // 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 {
+func (e *Encoder) Encode(msg Message) (err error) {
 	e.headersBuf.Reset()
 
-	err := encodeHeaders(e.headersBuf, msg.Headers)
-	if err != nil {
+	writer := e.w
+	if e.logger != nil {
+		encodeMsgBuf := bytes.NewBuffer(nil)
+		writer = io.MultiWriter(writer, encodeMsgBuf)
+		defer func() {
+			logMessageEncode(e.logger, encodeMsgBuf, msg, err)
+		}()
+	}
+
+	if err = EncodeHeaders(e.headersBuf, msg.Headers); err != nil {
 		return err
 	}
 
 	crc := crc32.New(crc32IEEETable)
-	hashWriter := io.MultiWriter(e.w, crc)
+	hashWriter := io.MultiWriter(writer, crc)
 
 	headersLen := uint32(e.headersBuf.Len())
 	payloadLen := uint32(len(msg.Payload))
 
-	if err := encodePrelude(hashWriter, crc, headersLen, payloadLen); err != nil {
+	if err = encodePrelude(hashWriter, crc, headersLen, payloadLen); err != nil {
 		return err
 	}
 
 	if headersLen > 0 {
-		if _, err := io.Copy(hashWriter, e.headersBuf); err != nil {
+		if _, err = io.Copy(hashWriter, e.headersBuf); err != nil {
 			return err
 		}
 	}
 
 	if payloadLen > 0 {
-		if _, err := hashWriter.Write(msg.Payload); err != nil {
+		if _, err = hashWriter.Write(msg.Payload); err != nil {
 			return err
 		}
 	}
 
 	msgCRC := crc.Sum32()
-	return binary.Write(e.w, binary.BigEndian, msgCRC)
+	return binary.Write(writer, binary.BigEndian, msgCRC)
+}
+
+func logMessageEncode(logger aws.Logger, msgBuf *bytes.Buffer, msg Message, encodeErr error) {
+	w := bytes.NewBuffer(nil)
+	defer func() { logger.Log(w.String()) }()
+
+	fmt.Fprintf(w, "Message to encode:\n")
+	encoder := json.NewEncoder(w)
+	if err := encoder.Encode(msg); err != nil {
+		fmt.Fprintf(w, "Failed to get encoded message, %v\n", err)
+	}
+
+	if encodeErr != nil {
+		fmt.Fprintf(w, "Encode error: %v\n", encodeErr)
+		return
+	}
+
+	fmt.Fprintf(w, "Raw message:\n%s\n", hex.Dump(msgBuf.Bytes()))
 }
 
 func encodePrelude(w io.Writer, crc hash.Hash32, headersLen, payloadLen uint32) error {
@@ -86,7 +132,9 @@ func encodePrelude(w io.Writer, crc hash.Hash32, headersLen, payloadLen uint32)
 	return nil
 }
 
-func encodeHeaders(w io.Writer, headers Headers) error {
+// EncodeHeaders writes the header values to the writer encoded in the event
+// stream format. Returns an error if a header fails to encode.
+func EncodeHeaders(w io.Writer, headers Headers) error {
 	for _, h := range headers {
 		hn := headerName{
 			Len: uint8(len(h.Name)),

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamapi/error.go

@@ -1,6 +1,9 @@
 package eventstreamapi
 
-import "fmt"
+import (
+	"fmt"
+	"sync"
+)
 
 type messageError struct {
 	code string
@@ -22,3 +25,53 @@ func (e messageError) Error() string {
 func (e messageError) OrigErr() error {
 	return nil
 }
+
+// OnceError wraps the behavior of recording an error
+// once and signal on a channel when this has occurred.
+// Signaling is done by closing of the channel.
+//
+// Type is safe for concurrent usage.
+type OnceError struct {
+	mu  sync.RWMutex
+	err error
+	ch  chan struct{}
+}
+
+// NewOnceError return a new OnceError
+func NewOnceError() *OnceError {
+	return &OnceError{
+		ch: make(chan struct{}, 1),
+	}
+}
+
+// Err acquires a read-lock and returns an
+// error if one has been set.
+func (e *OnceError) Err() error {
+	e.mu.RLock()
+	err := e.err
+	e.mu.RUnlock()
+
+	return err
+}
+
+// SetError acquires a write-lock and will set
+// the underlying error value if one has not been set.
+func (e *OnceError) SetError(err error) {
+	if err == nil {
+		return
+	}
+
+	e.mu.Lock()
+	if e.err == nil {
+		e.err = err
+		close(e.ch)
+	}
+	e.mu.Unlock()
+}
+
+// ErrorSet returns a channel that will be used to signal
+// that an error has been set. This channel will be closed
+// when the error value has been set for OnceError.
+func (e *OnceError) ErrorSet() <-chan struct{} {
+	return e.ch
+}

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamapi/reader.go

@@ -2,9 +2,7 @@ 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"
 )
@@ -15,27 +13,8 @@ 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)
@@ -47,27 +26,18 @@ type EventReader struct {
 // 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,
+	decoder *eventstream.Decoder,
 	payloadUnmarshaler protocol.PayloadUnmarshaler,
 	unmarshalerForEventType func(string) (Unmarshaler, error),
 ) *EventReader {
 	return &EventReader{
-		reader:                  reader,
-		decoder:                 eventstream.NewDecoder(reader),
+		decoder:                 decoder,
 		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.
 //
@@ -95,15 +65,27 @@ func (r *EventReader) ReadEvent() (event interface{}, err error) {
 	case EventMessageType:
 		return r.unmarshalEventMessage(msg)
 	case ExceptionMessageType:
-		err = r.unmarshalEventException(msg)
-		return nil, err
+		return nil, r.unmarshalEventException(msg)
 	case ErrorMessageType:
 		return nil, r.unmarshalErrorMessage(msg)
 	default:
-		return nil, fmt.Errorf("unknown eventstream message type, %v", typ)
+		return nil, &UnknownMessageTypeError{
+			Type: typ, Message: msg.Clone(),
+		}
 	}
 }
 
+// UnknownMessageTypeError provides an error when a message is received from
+// the stream, but the reader is unable to determine what kind of message it is.
+type UnknownMessageTypeError struct {
+	Type    string
+	Message eventstream.Message
+}
+
+func (e *UnknownMessageTypeError) Error() string {
+	return "unknown eventstream message type, " + e.Type
+}
+
 func (r *EventReader) unmarshalEventMessage(
 	msg eventstream.Message,
 ) (event interface{}, err error) {
@@ -174,11 +156,6 @@ func (r *EventReader) unmarshalErrorMessage(msg eventstream.Message) (err error)
 	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) {

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamapi/shared.go

@@ -0,0 +1,23 @@
+package eventstreamapi
+
+// EventStream headers with specific meaning to async API functionality.
+const (
+	ChunkSignatureHeader = `:chunk-signature` // chunk signature for message
+	DateHeader           = `:date`            // Date header for signature
+
+	// Message header and values
+	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`
+)

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamapi/signer.go

@@ -0,0 +1,123 @@
+package eventstreamapi
+
+import (
+	"bytes"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/private/protocol/eventstream"
+)
+
+var timeNow = time.Now
+
+// StreamSigner defines an interface for the implementation of signing of event stream payloads
+type StreamSigner interface {
+	GetSignature(headers, payload []byte, date time.Time) ([]byte, error)
+}
+
+// SignEncoder envelopes event stream messages
+// into an event stream message payload with included
+// signature headers using the provided signer and encoder.
+type SignEncoder struct {
+	signer     StreamSigner
+	encoder    Encoder
+	bufEncoder *BufferEncoder
+
+	closeErr error
+	closed   bool
+}
+
+// NewSignEncoder returns a new SignEncoder using the provided stream signer and
+// event stream encoder.
+func NewSignEncoder(signer StreamSigner, encoder Encoder) *SignEncoder {
+	// TODO: Need to pass down logging
+
+	return &SignEncoder{
+		signer:     signer,
+		encoder:    encoder,
+		bufEncoder: NewBufferEncoder(),
+	}
+}
+
+// Close encodes a final event stream signing envelope with an empty event stream
+// payload. This final end-frame is used to mark the conclusion of the stream.
+func (s *SignEncoder) Close() error {
+	if s.closed {
+		return s.closeErr
+	}
+
+	if err := s.encode([]byte{}); err != nil {
+		if strings.Contains(err.Error(), "on closed pipe") {
+			return nil
+		}
+
+		s.closeErr = err
+		s.closed = true
+		return s.closeErr
+	}
+
+	return nil
+}
+
+// Encode takes the provided message and add envelopes the message
+// with the required signature.
+func (s *SignEncoder) Encode(msg eventstream.Message) error {
+	payload, err := s.bufEncoder.Encode(msg)
+	if err != nil {
+		return err
+	}
+
+	return s.encode(payload)
+}
+
+func (s SignEncoder) encode(payload []byte) error {
+	date := timeNow()
+
+	var msg eventstream.Message
+	msg.Headers.Set(DateHeader, eventstream.TimestampValue(date))
+	msg.Payload = payload
+
+	var headers bytes.Buffer
+	if err := eventstream.EncodeHeaders(&headers, msg.Headers); err != nil {
+		return err
+	}
+
+	sig, err := s.signer.GetSignature(headers.Bytes(), msg.Payload, date)
+	if err != nil {
+		return err
+	}
+
+	msg.Headers.Set(ChunkSignatureHeader, eventstream.BytesValue(sig))
+
+	return s.encoder.Encode(msg)
+}
+
+// BufferEncoder is a utility that provides a buffered
+// event stream encoder
+type BufferEncoder struct {
+	encoder Encoder
+	buffer  *bytes.Buffer
+}
+
+// NewBufferEncoder returns a new BufferEncoder initialized
+// with a 1024 byte buffer.
+func NewBufferEncoder() *BufferEncoder {
+	buf := bytes.NewBuffer(make([]byte, 1024))
+	return &BufferEncoder{
+		encoder: eventstream.NewEncoder(buf),
+		buffer:  buf,
+	}
+}
+
+// Encode returns the encoded message as a byte slice.
+// The returned byte slice will be modified on the next encode call
+// and should not be held onto.
+func (e *BufferEncoder) Encode(msg eventstream.Message) ([]byte, error) {
+	e.buffer.Reset()
+
+	if err := e.encoder.Encode(msg); err != nil {
+		return nil, err
+	}
+
+	return e.buffer.Bytes(), nil
+}

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamapi/stream_writer.go

@@ -0,0 +1,129 @@
+package eventstreamapi
+
+import (
+	"fmt"
+	"io"
+	"sync"
+
+	"github.com/aws/aws-sdk-go/aws"
+)
+
+// StreamWriter provides concurrent safe writing to an event stream.
+type StreamWriter struct {
+	eventWriter *EventWriter
+	stream      chan eventWriteAsyncReport
+
+	done      chan struct{}
+	closeOnce sync.Once
+	err       *OnceError
+
+	streamCloser io.Closer
+}
+
+// NewStreamWriter returns a StreamWriter for the event writer, and stream
+// closer provided.
+func NewStreamWriter(eventWriter *EventWriter, streamCloser io.Closer) *StreamWriter {
+	w := &StreamWriter{
+		eventWriter:  eventWriter,
+		streamCloser: streamCloser,
+		stream:       make(chan eventWriteAsyncReport),
+		done:         make(chan struct{}),
+		err:          NewOnceError(),
+	}
+	go w.writeStream()
+
+	return w
+}
+
+// Close terminates the writers ability to write new events to the stream. Any
+// future call to Send will fail with an error.
+func (w *StreamWriter) Close() error {
+	w.closeOnce.Do(w.safeClose)
+	return w.Err()
+}
+
+func (w *StreamWriter) safeClose() {
+	close(w.done)
+}
+
+// ErrorSet returns a channel which will be closed
+// if an error occurs.
+func (w *StreamWriter) ErrorSet() <-chan struct{} {
+	return w.err.ErrorSet()
+}
+
+// Err returns any error that occurred while attempting to write an event to the
+// stream.
+func (w *StreamWriter) Err() error {
+	return w.err.Err()
+}
+
+// Send writes a single event to the stream returning an error if the write
+// failed.
+//
+// Send may be called concurrently. Events will be written to the stream
+// safely.
+func (w *StreamWriter) Send(ctx aws.Context, event Marshaler) error {
+	if err := w.Err(); err != nil {
+		return err
+	}
+
+	resultCh := make(chan error)
+	wrapped := eventWriteAsyncReport{
+		Event:  event,
+		Result: resultCh,
+	}
+
+	select {
+	case w.stream <- wrapped:
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-w.done:
+		return fmt.Errorf("stream closed, unable to send event")
+	}
+
+	select {
+	case err := <-resultCh:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-w.done:
+		return fmt.Errorf("stream closed, unable to send event")
+	}
+}
+
+func (w *StreamWriter) writeStream() {
+	defer w.Close()
+
+	for {
+		select {
+		case wrapper := <-w.stream:
+			err := w.eventWriter.WriteEvent(wrapper.Event)
+			wrapper.ReportResult(w.done, err)
+			if err != nil {
+				w.err.SetError(err)
+				return
+			}
+
+		case <-w.done:
+			if err := w.streamCloser.Close(); err != nil {
+				w.err.SetError(err)
+			}
+			return
+		}
+	}
+}
+
+type eventWriteAsyncReport struct {
+	Event  Marshaler
+	Result chan<- error
+}
+
+func (e eventWriteAsyncReport) ReportResult(cancel <-chan struct{}, err error) bool {
+	select {
+	case e.Result <- err:
+		return true
+	case <-cancel:
+		return false
+	}
+}

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamapi/writer.go

@@ -0,0 +1,109 @@
+package eventstreamapi
+
+import (
+	"github.com/aws/aws-sdk-go/private/protocol"
+	"github.com/aws/aws-sdk-go/private/protocol/eventstream"
+)
+
+// Marshaler provides a marshaling interface for event types to event stream
+// messages.
+type Marshaler interface {
+	MarshalEvent(protocol.PayloadMarshaler) (eventstream.Message, error)
+}
+
+// Encoder is an stream encoder that will encode an event stream message for
+// the transport.
+type Encoder interface {
+	Encode(eventstream.Message) error
+}
+
+// EventWriter provides a wrapper around the underlying event stream encoder
+// for an io.WriteCloser.
+type EventWriter struct {
+	encoder          Encoder
+	payloadMarshaler protocol.PayloadMarshaler
+	eventTypeFor     func(Marshaler) (string, error)
+}
+
+// NewEventWriter returns a new event stream writer, that will write to the
+// writer provided. Use the WriteEvent method to write an event to the stream.
+func NewEventWriter(encoder Encoder, pm protocol.PayloadMarshaler, eventTypeFor func(Marshaler) (string, error),
+) *EventWriter {
+	return &EventWriter{
+		encoder:          encoder,
+		payloadMarshaler: pm,
+		eventTypeFor:     eventTypeFor,
+	}
+}
+
+// WriteEvent writes an event to the stream. Returns an error if the event
+// fails to marshal into a message, or writing to the underlying writer fails.
+func (w *EventWriter) WriteEvent(event Marshaler) error {
+	msg, err := w.marshal(event)
+	if err != nil {
+		return err
+	}
+
+	return w.encoder.Encode(msg)
+}
+
+func (w *EventWriter) marshal(event Marshaler) (eventstream.Message, error) {
+	eventType, err := w.eventTypeFor(event)
+	if err != nil {
+		return eventstream.Message{}, err
+	}
+
+	msg, err := event.MarshalEvent(w.payloadMarshaler)
+	if err != nil {
+		return eventstream.Message{}, err
+	}
+
+	msg.Headers.Set(EventTypeHeader, eventstream.StringValue(eventType))
+	return msg, nil
+}
+
+//type EventEncoder struct {
+//	encoder           Encoder
+//	ppayloadMarshaler protocol.PayloadMarshaler
+//	eventTypeFor      func(Marshaler) (string, error)
+//}
+//
+//func (e EventEncoder) Encode(event Marshaler) error {
+//	msg, err := e.marshal(event)
+//	if err != nil {
+//		return err
+//	}
+//
+//	return w.encoder.Encode(msg)
+//}
+//
+//func (e EventEncoder) marshal(event Marshaler) (eventstream.Message, error) {
+//	eventType, err := w.eventTypeFor(event)
+//	if err != nil {
+//		return eventstream.Message{}, err
+//	}
+//
+//	msg, err := event.MarshalEvent(w.payloadMarshaler)
+//	if err != nil {
+//		return eventstream.Message{}, err
+//	}
+//
+//	msg.Headers.Set(EventTypeHeader, eventstream.StringValue(eventType))
+//	return msg, nil
+//}
+//
+//func (w *EventWriter) marshal(event Marshaler) (eventstream.Message, error) {
+//	eventType, err := w.eventTypeFor(event)
+//	if err != nil {
+//		return eventstream.Message{}, err
+//	}
+//
+//	msg, err := event.MarshalEvent(w.payloadMarshaler)
+//	if err != nil {
+//		return eventstream.Message{}, err
+//	}
+//
+//	msg.Headers.Set(EventTypeHeader, eventstream.StringValue(eventType))
+//	return msg, nil
+//}
+//

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/header.go

@@ -52,6 +52,15 @@ func (hs *Headers) Del(name string) {
 	}
 }
 
+// Clone returns a deep copy of the headers
+func (hs Headers) Clone() Headers {
+	o := make(Headers, 0, len(hs))
+	for _, h := range hs {
+		o.Set(h.Name, h.Value)
+	}
+	return o
+}
+
 func decodeHeaders(r io.Reader) (Headers, error) {
 	hs := Headers{}
 

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/header_value.go

@@ -461,6 +461,11 @@ func (v *TimestampValue) decode(r io.Reader) error {
 	return nil
 }
 
+// MarshalJSON implements the json.Marshaler interface
+func (v TimestampValue) MarshalJSON() ([]byte, error) {
+	return []byte(v.String()), nil
+}
+
 func timeFromEpochMilli(t int64) time.Time {
 	secs := t / 1e3
 	msec := t % 1e3

vendor/github.com/aws/aws-sdk-go/private/protocol/eventstream/message.go

@@ -27,7 +27,7 @@ func (m *Message) rawMessage() (rawMessage, error) {
 
 	if len(m.Headers) > 0 {
 		var headers bytes.Buffer
-		if err := encodeHeaders(&headers, m.Headers); err != nil {
+		if err := EncodeHeaders(&headers, m.Headers); err != nil {
 			return rawMessage{}, err
 		}
 		raw.Headers = headers.Bytes()
@@ -57,6 +57,20 @@ func (m *Message) rawMessage() (rawMessage, error) {
 	return raw, nil
 }
 
+// Clone returns a deep copy of the message.
+func (m Message) Clone() Message {
+	var payload []byte
+	if m.Payload != nil {
+		payload = make([]byte, len(m.Payload))
+		copy(payload, m.Payload)
+	}
+
+	return Message{
+		Headers: m.Headers.Clone(),
+		Payload: payload,
+	}
+}
+
 type messagePrelude struct {
 	Length     uint32
 	HeadersLen uint32