change github.com/aws/aws-sdk-go to be the same revision in kubernetes

vendor/github.com/aws/aws-sdk-go/private/model/api/eventstream.go

@@ -6,14 +6,12 @@ import (
 	"bytes"
 	"fmt"
 	"io"
-	"strings"
 	"text/template"
 )
 
 // EventStreamAPI provides details about the event stream async API and
 // associated EventStream shapes.
 type EventStreamAPI struct {
-	API       *API
 	Name      string
 	Operation *Operation
 	Shape     *Shape
@@ -24,10 +22,9 @@ type EventStreamAPI struct {
 // EventStream represents a single eventstream group (input/output) and the
 // modeled events that are known for the stream.
 type EventStream struct {
-	Name       string
-	Shape      *Shape
-	Events     []*Event
-	Exceptions []*Event
+	Name   string
+	Shape  *Shape
+	Events []*Event
 }
 
 // Event is a single EventStream event that can be sent or received in an
@@ -75,16 +72,6 @@ The events that can be sent are:
 	return commentify(w.String())
 }
 
-func hasEventStream(topShape *Shape) bool {
-	for _, ref := range topShape.MemberRefs {
-		if ref.Shape.IsEventStream {
-			return true
-		}
-	}
-
-	return false
-}
-
 func eventStreamAPIShapeRefDoc(refName string) string {
 	return commentify(fmt.Sprintf("Use %s to use the API's stream.", refName))
 }
@@ -112,8 +99,7 @@ func (a *API) setupEventStreams() {
 				a.Metadata.Protocol))
 		}
 
-		op.EventStreamAPI = &EventStreamAPI{
-			API:       a,
+		eventStreamAPI := &EventStreamAPI{
 			Name:      op.ExportedName + eventStreamMemberName,
 			Operation: op,
 			Outbound:  outbound,
@@ -122,18 +108,11 @@ func (a *API) setupEventStreams() {
 
 		streamShape := &Shape{
 			API:            a,
-			ShapeName:      op.EventStreamAPI.Name,
-			Documentation:  op.EventStreamAPI.ShapeDoc(),
+			ShapeName:      eventStreamAPI.Name,
+			Documentation:  eventStreamAPI.ShapeDoc(),
 			Type:           "structure",
-			EventStreamAPI: op.EventStreamAPI,
-			IsEventStream:  true,
-			MemberRefs: map[string]*ShapeRef{
-				"Inbound": &ShapeRef{
-					ShapeName: inbound.Shape.ShapeName,
-				},
-			},
+			EventStreamAPI: eventStreamAPI,
 		}
-		inbound.Shape.refs = append(inbound.Shape.refs, streamShape.MemberRefs["Inbound"])
 		streamShapeRef := &ShapeRef{
 			API:           a,
 			ShapeName:     streamShape.ShapeName,
@@ -141,7 +120,7 @@ func (a *API) setupEventStreams() {
 			Documentation: eventStreamAPIShapeRefDoc(eventStreamMemberName),
 		}
 		streamShape.refs = []*ShapeRef{streamShapeRef}
-		op.EventStreamAPI.Shape = streamShape
+		eventStreamAPI.Shape = streamShape
 
 		if _, ok := op.OutputRef.Shape.MemberRefs[eventStreamMemberName]; ok {
 			panic(fmt.Sprintf("shape ref already exists, %s.%s",
@@ -173,14 +152,9 @@ func setupEventStream(topShape *Shape) *EventStream {
 			Name:  ref.Shape.ShapeName,
 			Shape: ref.Shape,
 		}
-
-		if topShape.API.Metadata.Protocol == "json" {
-			topShape.EventFor = append(topShape.EventFor, eventStream)
-		}
-
 		for _, eventRefName := range ref.Shape.MemberNames() {
 			eventRef := ref.Shape.MemberRefs[eventRefName]
-			if !(eventRef.Shape.IsEvent || eventRef.Shape.Exception) {
+			if !eventRef.Shape.IsEvent {
 				panic(fmt.Sprintf("unexpected non-event member reference %s.%s",
 					ref.Shape.ShapeName, eventRefName))
 			}
@@ -188,19 +162,11 @@ func setupEventStream(topShape *Shape) *EventStream {
 			updateEventPayloadRef(eventRef.Shape)
 
 			eventRef.Shape.EventFor = append(eventRef.Shape.EventFor, eventStream)
-
-			// Exceptions and events are two different lists to allow the SDK
-			// to easly generate code with the two handled differently.
-			event := &Event{
+			eventStream.Events = append(eventStream.Events, &Event{
 				Name:  eventRefName,
 				Shape: eventRef.Shape,
 				For:   eventStream,
-			}
-			if eventRef.Shape.Exception {
-				eventStream.Exceptions = append(eventStream.Exceptions, event)
-			} else {
-				eventStream.Events = append(eventStream.Events, event)
-			}
+			})
 		}
 
 		// Remove the eventstream references as they will be added elsewhere.
@@ -233,18 +199,22 @@ func updateEventPayloadRef(parent *Shape) {
 
 func renderEventStreamAPIShape(w io.Writer, s *Shape) error {
 	// Imports needed by the EventStream APIs.
-	s.API.imports["fmt"] = true
 	s.API.imports["bytes"] = true
 	s.API.imports["io"] = true
 	s.API.imports["sync"] = true
 	s.API.imports["sync/atomic"] = true
-	s.API.imports["github.com/aws/aws-sdk-go/aws/awserr"] = true
 	s.API.imports["github.com/aws/aws-sdk-go/private/protocol/eventstream"] = true
 	s.API.imports["github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamapi"] = true
 
 	return eventStreamAPIShapeTmpl.Execute(w, s)
 }
 
+// EventStreamReaderInterfaceName returns the interface name for the
+// EventStream's reader interface.
+func EventStreamReaderInterfaceName(s *Shape) string {
+	return s.ShapeName + "Reader"
+}
+
 // Template for an EventStream API Shape that will provide read/writing events
 // across the EventStream. This is a special shape that's only public members
 // are the Events channel and a Close and Err method.
@@ -370,7 +340,7 @@ var eventStreamAPIReaderTmpl = template.Must(template.New("eventStreamAPIReaderT
 //     * {{ $event.Shape.ShapeName }}
 {{- end }}
 type {{ $.EventStreamAPI.Inbound.Name }}Event interface {
-	event{{ $.EventStreamAPI.Inbound.Name }}()
+	event{{ $.EventStreamAPI.Name }}()
 }
 
 // {{ $.ShapeName }}Reader provides the interface for reading EventStream
@@ -402,10 +372,6 @@ type read{{ $.ShapeName }} struct {
 
 	done      chan struct{}
 	closeOnce sync.Once
-
-	{{ if eq $.API.Metadata.Protocol "json" -}}
-		initResp eventstreamapi.Unmarshaler
-	{{ end -}}
 }
 
 func newRead{{ $.ShapeName }}(
@@ -413,16 +379,10 @@ func newRead{{ $.ShapeName }}(
 	unmarshalers request.HandlerList,
 	logger aws.Logger,
 	logLevel aws.LogLevelType,
-	{{ if eq $.API.Metadata.Protocol "json" -}}
-		initResp eventstreamapi.Unmarshaler,
-	{{ end -}}
 ) *read{{ $.ShapeName }} {
 	r := &read{{ $.ShapeName }}{
 		stream: make(chan {{ $.EventStreamAPI.Inbound.Name }}Event),
 		done: make(chan struct{}),
-		{{ if eq $.API.Metadata.Protocol "json" -}}
-			initResp: initResp,
-		{{ end -}}
 	}
 
 	r.eventReader = eventstreamapi.NewEventReader(
@@ -496,24 +456,13 @@ func (r *read{{ $.ShapeName }}) unmarshalerForEventType(
 	eventType string,
 ) (eventstreamapi.Unmarshaler, error) {
 	switch eventType {
-		{{- if eq $.API.Metadata.Protocol "json" }}
-			case "initial-response":
-				return r.initResp, nil
-		{{ end -}}
 		{{- range $_, $event := $.EventStreamAPI.Inbound.Events }}
 			case {{ printf "%q" $event.Name }}:
 				return &{{ $event.Shape.ShapeName }}{}, nil
 		{{ end -}}
-		{{- range $_, $event := $.EventStreamAPI.Inbound.Exceptions }}
-			case {{ printf "%q" $event.Name }}:
-				return &{{ $event.Shape.ShapeName }}{}, nil
-		{{ end -}}
 	default:
-		return nil, awserr.New(
-			request.ErrCodeSerialization,
-			fmt.Sprintf("unknown event type name, %s, for {{ $.ShapeName }}", eventType),
-			nil,
-		)
+		return nil, fmt.Errorf(
+			"unknown event type name, %s, for {{ $.ShapeName }}", eventType)
 	}
 }
 `))
@@ -536,9 +485,6 @@ func (s *{{ $.ShapeName }}) runEventStreamLoop(r *request.Request) {
 			r.Handlers.UnmarshalStream,
 			r.Config.Logger,
 			r.Config.LogLevel.Value(),
-			{{ if eq $.API.Metadata.Protocol "json" -}}
-				s,
-			{{ end -}}
 		)
 		go reader.readEventStream()
 
@@ -550,85 +496,13 @@ func (s *{{ $.ShapeName }}) runEventStreamLoop(r *request.Request) {
 
 	s.{{ $.EventStreamsMemberName }} = eventStream
 }
-
-{{ if eq $.API.Metadata.Protocol "json" -}}
-	func (s *{{ $.ShapeName }}) unmarshalInitialResponse(r *request.Request) {
-		// Wait for the initial response event, which must be the first event to be
-		// received from the API.
-		select {
-		case event, ok := <-s.EventStream.Events():
-			if !ok {
-				return
-			}
-			es := s.EventStream
-			v, ok := event.(*{{ $.ShapeName }})
-			if !ok || v == nil {
-				r.Error = awserr.New(
-					request.ErrCodeSerialization,
-					fmt.Sprintf("invalid event, %T, expect *SubscribeToShardOutput, %v", event, v),
-					nil,
-				)
-				return
-			}
-			*s = *v
-			s.EventStream = es
-		}
-	}
-{{ end -}}
 `))
 
-// EventStreamHeaderTypeMap provides the mapping of a EventStream Header's
-// Value type to the shape reference's member type.
-type EventStreamHeaderTypeMap struct {
-	Header string
-	Member string
-}
-
-var eventStreamEventShapeTmplFuncs = template.FuncMap{
-	"EventStreamHeaderTypeMap": func(ref *ShapeRef) EventStreamHeaderTypeMap {
-		switch ref.Shape.Type {
-		case "boolean":
-			return EventStreamHeaderTypeMap{Header: "bool", Member: "bool"}
-		case "byte":
-			return EventStreamHeaderTypeMap{Header: "int8", Member: "int64"}
-		case "short":
-			return EventStreamHeaderTypeMap{Header: "int16", Member: "int64"}
-		case "integer":
-			return EventStreamHeaderTypeMap{Header: "int32", Member: "int64"}
-		case "long":
-			return EventStreamHeaderTypeMap{Header: "int64", Member: "int64"}
-		case "timestamp":
-			return EventStreamHeaderTypeMap{Header: "time.Time", Member: "time.Time"}
-		case "blob":
-			return EventStreamHeaderTypeMap{Header: "[]byte", Member: "[]byte"}
-		case "string":
-			return EventStreamHeaderTypeMap{Header: "string", Member: "string"}
-		// TODO case "uuid"  what is modeled type
-		default:
-			panic("unsupported EventStream header type, " + ref.Shape.Type)
-		}
-	},
-	"HasNonBlobPayloadMembers": eventHasNonBlobPayloadMembers,
-}
-
-// Returns if the event has any members which are not the event's blob payload,
-// nor a header.
-func eventHasNonBlobPayloadMembers(s *Shape) bool {
-	num := len(s.MemberRefs)
-	for _, ref := range s.MemberRefs {
-		if ref.IsEventHeader || (ref.IsEventPayload && (ref.Shape.Type == "blob" || ref.Shape.Type == "string")) {
-			num--
-		}
-	}
-	return num > 0
-}
-
 // Template for an EventStream Event shape. This is a normal API shape that is
 // decorated as an EventStream Event.
 //
 // Executed in the context of a Shape.
-var eventStreamEventShapeTmpl = template.Must(template.New("eventStreamEventShapeTmpl").
-	Funcs(eventStreamEventShapeTmplFuncs).Parse(`
+var eventStreamEventShapeTmpl = template.Must(template.New("eventStreamEventShapeTmpl").Parse(`
 {{ range $_, $eventstream := $.EventFor }}
 	// The {{ $.ShapeName }} is and event in the {{ $eventstream.Name }} group of events.
 	func (s *{{ $.ShapeName }}) event{{ $eventstream.Name }}() {}
@@ -640,541 +514,24 @@ func (s *{{ $.ShapeName }}) UnmarshalEvent(
 	payloadUnmarshaler protocol.PayloadUnmarshaler,
 	msg eventstream.Message,
 ) error {
-	{{- range $memName, $memRef := $.MemberRefs }}
-		{{- if $memRef.IsEventHeader }}
-			if hv := msg.Headers.Get("{{ $memName }}"); hv != nil {
-				{{ $types := EventStreamHeaderTypeMap $memRef -}}
-				v := hv.Get().({{ $types.Header }})
-				{{- if ne $types.Header $types.Member }}
-					m := {{ $types.Member }}(v)
-					s.{{ $memName }} = {{ if $memRef.UseIndirection }}&{{ end }}m
-				{{- else }}
-					s.{{ $memName }} = {{ if $memRef.UseIndirection }}&{{ end }}v
-				{{- end }}
+	{{- range $fieldIdx, $fieldName := $.MemberNames }}
+		{{- $fieldRef := index $.MemberRefs $fieldName -}}
+		{{ if $fieldRef.IsEventHeader }}
+			// TODO handle event header, {{ $fieldName }}
+		{{- else if (and ($fieldRef.IsEventPayload) (eq $fieldRef.Shape.Type "blob")) }}
+			s.{{ $fieldName }} = make([]byte, len(msg.Payload))
+			copy(s.{{ $fieldName }}, msg.Payload)
+		{{- else }}
+			if err := payloadUnmarshaler.UnmarshalPayload(
+				bytes.NewReader(msg.Payload), s,
+			); err != nil {
+				return fmt.Errorf("failed to unmarshal payload, %v", err)
 			}
-		{{- else if (and ($memRef.IsEventPayload) (eq $memRef.Shape.Type "blob")) }}
-			s.{{ $memName }} = make([]byte, len(msg.Payload))
-			copy(s.{{ $memName }}, msg.Payload)
-		{{- else if (and ($memRef.IsEventPayload) (eq $memRef.Shape.Type "string")) }}
-			s.{{ $memName }} = aws.String(string(msg.Payload))
 		{{- end }}
 	{{- end }}
-	{{- if HasNonBlobPayloadMembers $ }}
-		if err := payloadUnmarshaler.UnmarshalPayload(
-			bytes.NewReader(msg.Payload), s,
-		); err != nil {
-			return err
-		}
-	{{- end }}
 	return nil
 }
 `))
 
-var eventStreamExceptionEventShapeTmpl = template.Must(
-	template.New("eventStreamExceptionEventShapeTmpl").Parse(`
-// Code returns the exception type name.
-func (s {{ $.ShapeName }}) Code() string {
-	{{- if $.ErrorInfo.Code }}
-		return "{{ $.ErrorInfo.Code }}"
-	{{- else }}
-		return "{{ $.ShapeName }}"
-	{{ end -}}
-}
-
-// Message returns the exception's message.
-func (s {{ $.ShapeName }}) Message() string {
-	{{- if index $.MemberRefs "Message_" }}
-		return *s.Message_
-	{{- else }}
-		return ""
-	{{ end -}}
-}
-
-// OrigErr always returns nil, satisfies awserr.Error interface.
-func (s {{ $.ShapeName }}) OrigErr() error {
-	return nil
-}
-
-func (s {{ $.ShapeName }}) Error() string {
-	return fmt.Sprintf("%s: %s", s.Code(), s.Message())
-}
-`))
-
-// APIEventStreamTestGoCode generates Go code for EventStream operation tests.
-func (a *API) APIEventStreamTestGoCode() string {
-	var buf bytes.Buffer
-
-	a.resetImports()
-	a.imports["bytes"] = true
-	a.imports["io/ioutil"] = true
-	a.imports["net/http"] = true
-	a.imports["reflect"] = true
-	a.imports["testing"] = true
-	a.imports["time"] = true
-	a.imports["github.com/aws/aws-sdk-go/aws/corehandlers"] = true
-	a.imports["github.com/aws/aws-sdk-go/aws/request"] = true
-	a.imports["github.com/aws/aws-sdk-go/aws/awserr"] = true
-	a.imports["github.com/aws/aws-sdk-go/awstesting/unit"] = true
-	a.imports["github.com/aws/aws-sdk-go/private/protocol"] = true
-	a.imports["github.com/aws/aws-sdk-go/private/protocol/"+a.ProtocolPackage()] = true
-	a.imports["github.com/aws/aws-sdk-go/private/protocol/eventstream"] = true
-	a.imports["github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamapi"] = true
-	a.imports["github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamtest"] = true
-
-	unused := `
-	var _ time.Time
-	var _ awserr.Error
-	`
-
-	if err := eventStreamTestTmpl.Execute(&buf, a); err != nil {
-		panic(err)
-	}
-
-	return a.importsGoCode() + unused + strings.TrimSpace(buf.String())
-}
-
-func valueForType(s *Shape, visited []string) string {
-	for _, v := range visited {
-		if v == s.ShapeName {
-			return "nil"
-		}
-	}
-
-	visited = append(visited, s.ShapeName)
-
-	switch s.Type {
-	case "blob":
-		return `[]byte("blob value goes here")`
-	case "string":
-		return `aws.String("string value goes here")`
-	case "boolean":
-		return `aws.Bool(true)`
-	case "byte":
-		return `aws.Int64(1)`
-	case "short":
-		return `aws.Int64(12)`
-	case "integer":
-		return `aws.Int64(123)`
-	case "long":
-		return `aws.Int64(1234)`
-	case "float":
-		return `aws.Float64(123.4)`
-	case "double":
-		return `aws.Float64(123.45)`
-	case "timestamp":
-		return `aws.Time(time.Unix(1396594860, 0).UTC())`
-	case "structure":
-		w := bytes.NewBuffer(nil)
-		fmt.Fprintf(w, "&%s{\n", s.ShapeName)
-		for _, refName := range s.MemberNames() {
-			fmt.Fprintf(w, "%s: %s,\n", refName, valueForType(s.MemberRefs[refName].Shape, visited))
-		}
-		fmt.Fprintf(w, "}")
-		return w.String()
-	case "list":
-		w := bytes.NewBuffer(nil)
-		fmt.Fprintf(w, "%s{\n", s.GoType())
-		for i := 0; i < 3; i++ {
-			fmt.Fprintf(w, "%s,\n", valueForType(s.MemberRef.Shape, visited))
-		}
-		fmt.Fprintf(w, "}")
-		return w.String()
-
-	case "map":
-		w := bytes.NewBuffer(nil)
-		fmt.Fprintf(w, "%s{\n", s.GoType())
-		for _, k := range []string{"a", "b", "c"} {
-			fmt.Fprintf(w, "%q: %s,\n", k, valueForType(s.ValueRef.Shape, visited))
-		}
-		fmt.Fprintf(w, "}")
-		return w.String()
-
-	default:
-		panic(fmt.Sprintf("valueForType does not support %s, %s", s.ShapeName, s.Type))
-	}
-}
-
-func setEventHeaderValueForType(s *Shape, memVar string) string {
-	switch s.Type {
-	case "blob":
-		return fmt.Sprintf("eventstream.BytesValue(%s)", memVar)
-	case "string":
-		return fmt.Sprintf("eventstream.StringValue(*%s)", memVar)
-	case "boolean":
-		return fmt.Sprintf("eventstream.BoolValue(*%s)", memVar)
-	case "byte":
-		return fmt.Sprintf("eventstream.Int8Value(int8(*%s))", memVar)
-	case "short":
-		return fmt.Sprintf("eventstream.Int16Value(int16(*%s))", memVar)
-	case "integer":
-		return fmt.Sprintf("eventstream.Int32Value(int32(*%s))", memVar)
-	case "long":
-		return fmt.Sprintf("eventstream.Int64Value(*%s)", memVar)
-	case "float":
-		return fmt.Sprintf("eventstream.Float32Value(float32(*%s))", memVar)
-	case "double":
-		return fmt.Sprintf("eventstream.Float64Value(*%s)", memVar)
-	case "timestamp":
-		return fmt.Sprintf("eventstream.TimestampValue(*%s)", memVar)
-	default:
-		panic(fmt.Sprintf("value type %s not supported for event headers, %s", s.Type, s.ShapeName))
-	}
-}
-
-func templateMap(args ...interface{}) map[string]interface{} {
-	if len(args)%2 != 0 {
-		panic(fmt.Sprintf("invalid map call, non-even args %v", args))
-	}
-
-	m := map[string]interface{}{}
-	for i := 0; i < len(args); i += 2 {
-		k, ok := args[i].(string)
-		if !ok {
-			panic(fmt.Sprintf("invalid map call, arg is not string, %T, %v", args[i], args[i]))
-		}
-		m[k] = args[i+1]
-	}
-
-	return m
-}
-
-var eventStreamTestTmpl = template.Must(
-	template.New("eventStreamTestTmpl").Funcs(template.FuncMap{
-		"ValueForType":               valueForType,
-		"HasNonBlobPayloadMembers":   eventHasNonBlobPayloadMembers,
-		"SetEventHeaderValueForType": setEventHeaderValueForType,
-		"Map": templateMap,
-		"OptionalAddInt": func(do bool, a, b int) int {
-			if !do {
-				return a
-			}
-			return a + b
-		},
-		"HasNonEventStreamMember": func(s *Shape) bool {
-			for _, ref := range s.MemberRefs {
-				if !ref.Shape.IsEventStream {
-					return true
-				}
-			}
-			return false
-		},
-	}).Parse(`
-{{ range $opName, $op := $.Operations }}
-	{{ if $op.EventStreamAPI }}
-		{{ if $op.EventStreamAPI.Inbound }}
-			{{ template "event stream inbound tests" $op.EventStreamAPI }}
-		{{ end }}
-	{{ end }}
-{{ end }}
-
-type loopReader struct {
-	source *bytes.Reader
-}
-
-func (c *loopReader) Read(p []byte) (int, error) {
-	if c.source.Len() == 0 {
-		c.source.Seek(0, 0)
-	}
-
-	return c.source.Read(p)
-}
-
-{{ define "event stream inbound tests" }}
-	func Test{{ $.Operation.ExportedName }}_Read(t *testing.T) {
-		expectEvents, eventMsgs := mock{{ $.Operation.ExportedName }}ReadEvents()
-		sess, cleanupFn, err := eventstreamtest.SetupEventStreamSession(t,
-			eventstreamtest.ServeEventStream{
-				T:      t,
-				Events: eventMsgs,
-			},
-			true,
-		)
-		if err != nil {
-			t.Fatalf("expect no error, %v", err)
-		}
-		defer cleanupFn()
-
-		svc := New(sess)
-		resp, err := svc.{{ $.Operation.ExportedName }}(nil)
-		if err != nil {
-			t.Fatalf("expect no error got, %v", err)
-		}
-		defer resp.EventStream.Close()
-
-		{{- if and (eq $.Operation.API.Metadata.Protocol "json") (HasNonEventStreamMember $.Operation.OutputRef.Shape) }}
-			expectResp := expectEvents[0].(*{{ $.Operation.OutputRef.Shape.ShapeName }})
-			{{- range $name, $ref := $.Operation.OutputRef.Shape.MemberRefs }}
-				{{- if not $ref.Shape.IsEventStream }}
-					if e, a := expectResp.{{ $name }}, resp.{{ $name }}; !reflect.DeepEqual(e,a) {
-						t.Errorf("expect %v, got %v", e, a)
-					}
-				{{- end }}
-			{{- end }}
-			// Trim off response output type pseudo event so only event messages remain.
-			expectEvents = expectEvents[1:]
-		{{ end }}
-
-		var i int
-		for event := range resp.EventStream.Events() {
-			if event == nil {
-				t.Errorf("%d, expect event, got nil", i)
-			}
-			if e, a := expectEvents[i], event; !reflect.DeepEqual(e, a) {
-				t.Errorf("%d, expect %T %v, got %T %v", i, e, e, a, a)
-			}
-			i++
-		}
-
-		if err := resp.EventStream.Err(); err != nil {
-			t.Errorf("expect no error, %v", err)
-		}
-	}
-
-	func Test{{ $.Operation.ExportedName }}_ReadClose(t *testing.T) {
-		_, eventMsgs := mock{{ $.Operation.ExportedName }}ReadEvents()
-		sess, cleanupFn, err := eventstreamtest.SetupEventStreamSession(t,
-			eventstreamtest.ServeEventStream{
-				T:      t,
-				Events: eventMsgs,
-			},
-			true,
-		)
-		if err != nil {
-			t.Fatalf("expect no error, %v", err)
-		}
-		defer cleanupFn()
-
-		svc := New(sess)
-		resp, err := svc.{{ $.Operation.ExportedName }}(nil)
-		if err != nil {
-			t.Fatalf("expect no error got, %v", err)
-		}
-
-		resp.EventStream.Close()
-		<-resp.EventStream.Events()
-
-		if err := resp.EventStream.Err(); err != nil {
-			t.Errorf("expect no error, %v", err)
-		}
-	}
-
-	func Benchmark{{ $.Operation.ExportedName }}_Read(b *testing.B) {
-		_, eventMsgs := mock{{ $.Operation.ExportedName }}ReadEvents()
-		var buf bytes.Buffer
-		encoder := eventstream.NewEncoder(&buf)
-		for _, msg := range eventMsgs {
-			if err := encoder.Encode(msg); err != nil {
-				b.Fatalf("failed to encode message, %v", err)
-			}
-		}
-		stream := &loopReader{source: bytes.NewReader(buf.Bytes())}
-
-		sess := unit.Session
-		svc := New(sess, &aws.Config{
-			Endpoint:               aws.String("https://example.com"),
-			DisableParamValidation: aws.Bool(true),
-		})
-		svc.Handlers.Send.Swap(corehandlers.SendHandler.Name,
-			request.NamedHandler{Name: "mockSend",
-				Fn: func(r *request.Request) {
-					r.HTTPResponse = &http.Response{
-						Status:     "200 OK",
-						StatusCode: 200,
-						Header:     http.Header{},
-						Body:       ioutil.NopCloser(stream),
-					}
-				},
-			},
-		)
-
-		resp, err := svc.{{ $.Operation.ExportedName }}(nil)
-		if err != nil {
-			b.Fatalf("failed to create request, %v", err)
-		}
-		defer resp.EventStream.Close()
-		b.ResetTimer()
-
-		for i := 0; i < b.N; i++ {
-			if err = resp.EventStream.Err(); err != nil {
-				b.Fatalf("expect no error, got %v", err)
-			}
-			event := <-resp.EventStream.Events()
-			if event == nil {
-				b.Fatalf("expect event, got nil, %v, %d", resp.EventStream.Err(), i)
-			}
-		}
-	}
-
-	func mock{{ $.Operation.ExportedName }}ReadEvents() (
-		[]{{ $.Inbound.Name }}Event,
-		[]eventstream.Message,
-	) {
-		expectEvents := []{{ $.Inbound.Name }}Event {
-			{{- if eq $.Operation.API.Metadata.Protocol "json" }}
-				{{- template "set event type" $.Operation.OutputRef.Shape }}
-			{{- end }}
-			{{- range $_, $event := $.Inbound.Events }}
-				{{- template "set event type" $event.Shape }}
-			{{- end }}
-		}
-
-		var marshalers request.HandlerList
-		marshalers.PushBackNamed({{ $.API.ProtocolPackage }}.BuildHandler)
-		payloadMarshaler := protocol.HandlerPayloadMarshal{
-			Marshalers: marshalers,
-		}
-		_ = payloadMarshaler
-
-		eventMsgs := []eventstream.Message{
-			{{- if eq $.Operation.API.Metadata.Protocol "json" }}
-				{{- template "set event message" Map "idx" 0 "parentShape" $.Operation.OutputRef.Shape "eventName" "initial-response" }}
-			{{- end }}
-			{{- range $idx, $event := $.Inbound.Events }}
-				{{- $offsetIdx := OptionalAddInt (eq $.Operation.API.Metadata.Protocol "json") $idx 1 }}
-				{{- template "set event message" Map "idx" $offsetIdx "parentShape" $event.Shape "eventName" $event.Name }}
-			{{- end }}
-		}
-
-		return expectEvents, eventMsgs
-	}
-
-	{{- if $.Inbound.Exceptions }}
-		func Test{{ $.Operation.ExportedName }}_ReadException(t *testing.T) {
-			expectEvents := []{{ $.Inbound.Name }}Event {
-				{{- if eq $.Operation.API.Metadata.Protocol "json" }}
-					{{- template "set event type" $.Operation.OutputRef.Shape }}
-				{{- end }}
-
-				{{- $exception := index $.Inbound.Exceptions 0 }}
-				{{- template "set event type" $exception.Shape }}
-			}
-
-			var marshalers request.HandlerList
-			marshalers.PushBackNamed({{ $.API.ProtocolPackage }}.BuildHandler)
-			payloadMarshaler := protocol.HandlerPayloadMarshal{
-				Marshalers: marshalers,
-			}
-
-			eventMsgs := []eventstream.Message{
-				{{- if eq $.Operation.API.Metadata.Protocol "json" }}
-					{{- template "set event message" Map "idx" 0 "parentShape" $.Operation.OutputRef.Shape "eventName" "initial-response" }}
-				{{- end }}
-
-				{{- $offsetIdx := OptionalAddInt (eq $.Operation.API.Metadata.Protocol "json") 0 1 }}
-				{{- $exception := index $.Inbound.Exceptions 0 }}
-				{{- template "set event message" Map "idx" $offsetIdx "parentShape" $exception.Shape "eventName" $exception.Name }}
-			}
-
-			sess, cleanupFn, err := eventstreamtest.SetupEventStreamSession(t,
-				eventstreamtest.ServeEventStream{
-					T:      t,
-					Events: eventMsgs,
-				},
-				true,
-			)
-			if err != nil {
-				t.Fatalf("expect no error, %v", err)
-			}
-			defer cleanupFn()
-
-			svc := New(sess)
-			resp, err := svc.{{ $.Operation.ExportedName }}(nil)
-			if err != nil {
-				t.Fatalf("expect no error got, %v", err)
-			}
-
-			defer resp.EventStream.Close()
-
-			<-resp.EventStream.Events()
-
-			err = resp.EventStream.Err()
-			if err == nil {
-				t.Fatalf("expect err, got none")
-			}
-
-			expectErr := {{ ValueForType $exception.Shape nil }}
-			aerr, ok := err.(awserr.Error)
-			if !ok {
-				t.Errorf("expect exception, got %T, %#v", err, err)
-			}
-			if e, a := expectErr.Code(), aerr.Code(); e != a {
-				t.Errorf("expect %v, got %v", e, a)
-			}
-			if e, a := expectErr.Message(), aerr.Message(); e != a {
-				t.Errorf("expect %v, got %v", e, a)
-			}
-
-			if e, a := expectErr, aerr; !reflect.DeepEqual(e, a) {
-				t.Errorf("expect %#v, got %#v", e, a)
-			}
-		}
-
-		{{- range $_, $exception := $.Inbound.Exceptions }}
-			var _ awserr.Error = (*{{ $exception.Shape.ShapeName }})(nil)
-		{{- end }}
-
-	{{ end }}
-{{ end }}
-
-{{/* Params: *Shape */}}
-{{ define "set event type" }}
-	&{{ $.ShapeName }}{
-		{{- range $memName, $memRef := $.MemberRefs }}
-			{{- if not $memRef.Shape.IsEventStream }}
-				{{ $memName }}: {{ ValueForType $memRef.Shape nil }},
-			{{- end }}
-		{{- end }}
-	},
-{{- end }}
-
-{{/* Params: idx:int, parentShape:*Shape, eventName:string */}}
-{{ define "set event message" }}
-	{
-		Headers: eventstream.Headers{
-			{{- if $.parentShape.Exception }}
-				eventstreamtest.EventExceptionTypeHeader,
-				{
-					Name:  eventstreamapi.ExceptionTypeHeader,
-					Value: eventstream.StringValue("{{ $.eventName }}"),
-				},
-			{{- else }}
-				eventstreamtest.EventMessageTypeHeader,
-				{
-					Name:  eventstreamapi.EventTypeHeader,
-					Value: eventstream.StringValue("{{ $.eventName }}"),
-				},
-			{{- end }}
-			{{- range $memName, $memRef := $.parentShape.MemberRefs }}
-				{{- template "set event message header" Map "idx" $.idx "parentShape" $.parentShape "memName" $memName "memRef" $memRef }}
-			{{- end }}
-		},
-		{{- template "set event message payload" Map "idx" $.idx "parentShape" $.parentShape }}
-	},
-{{- end }}
-
-{{/* Params: idx:int, parentShape:*Shape, memName:string, memRef:*ShapeRef */}}
-{{ define "set event message header" }}
-	{{- if $.memRef.IsEventHeader }}
-		{
-			Name: "{{ $.memName }}",
-			{{- $shapeValueVar := printf "expectEvents[%d].(%s).%s" $.idx $.parentShape.GoType $.memName }}
-			Value: {{ SetEventHeaderValueForType $.memRef.Shape $shapeValueVar }},
-		},
-	{{- end }}
-{{- end }}
-
-{{/* Params: idx:int, parentShape:*Shape, memName:string, memRef:*ShapeRef */}}
-{{ define "set event message payload" }}
-	{{- $payloadMemName := $.parentShape.PayloadRefName }}
-	{{- if HasNonBlobPayloadMembers $.parentShape }}
-		Payload: eventstreamtest.MarshalEventPayload(payloadMarshaler, expectEvents[{{ $.idx }}]),
-	{{- else if $payloadMemName }} 
-		{{- $shapeType := (index $.parentShape.MemberRefs $payloadMemName).Shape.Type }}
-		{{- if eq $shapeType "blob" }}
-			Payload: expectEvents[{{ $.idx }}].({{ $.parentShape.GoType }}).{{ $payloadMemName }},
-		{{- else if eq $shapeType "string" }}
-			Payload: []byte(*expectEvents[{{ $.idx }}].({{ $.parentShape.GoType }}).{{ $payloadMemName }}),
-		{{- end }}
-	{{- end }}
-{{- end }}
+var eventStreamTestTmpl = template.Must(template.New("eventStreamTestTmpl").Parse(`
 `))