// Code generated by smithy-go-codegen DO NOT EDIT.

package bedrockruntime

import (
	"bytes"
	"context"
	"fmt"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream"
	"github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream/eventstreamapi"
	"github.com/aws/aws-sdk-go-v2/aws/signer/v4"
	internalauthsmithy "github.com/aws/aws-sdk-go-v2/internal/auth/smithy"
	"github.com/aws/aws-sdk-go-v2/service/bedrockruntime/types"
	smithy "github.com/aws/smithy-go"
	"github.com/aws/smithy-go/middleware"
	smithysync "github.com/aws/smithy-go/sync"
	smithyhttp "github.com/aws/smithy-go/transport/http"
	"io"
	"io/ioutil"
	"sync"
	"time"
)

// InvokeModelWithBidirectionalStreamInputWriter provides the interface for
// writing events to a stream.
//
// The writer's Close method must allow multiple concurrent calls.
type InvokeModelWithBidirectionalStreamInputWriter interface {
	Send(context.Context, types.InvokeModelWithBidirectionalStreamInput) error
	Close() error
	Err() error
}

// ConverseStreamOutputReader provides the interface for reading events from a
// stream.
//
// The writer's Close method must allow multiple concurrent calls.
type ConverseStreamOutputReader interface {
	Events() <-chan types.ConverseStreamOutput
	Close() error
	Err() error
}

// InvokeModelWithBidirectionalStreamOutputReader provides the interface for
// reading events from a stream.
//
// The writer's Close method must allow multiple concurrent calls.
type InvokeModelWithBidirectionalStreamOutputReader interface {
	Events() <-chan types.InvokeModelWithBidirectionalStreamOutput
	Close() error
	Err() error
}

// ResponseStreamReader provides the interface for reading events from a stream.
//
// The writer's Close method must allow multiple concurrent calls.
type ResponseStreamReader interface {
	Events() <-chan types.ResponseStream
	Close() error
	Err() error
}

type eventStreamSigner interface {
	GetSignature(ctx context.Context, headers, payload []byte, signingTime time.Time, optFns ...func(*v4.StreamSignerOptions)) ([]byte, error)
}

type asyncInvokeModelWithBidirectionalStreamInput struct {
	Event  types.InvokeModelWithBidirectionalStreamInput
	Result chan<- error
}

func (e asyncInvokeModelWithBidirectionalStreamInput) ReportResult(cancel <-chan struct{}, err error) bool {
	select {
	case e.Result <- err:
		return true

	case <-cancel:
		return false

	}
}

type invokeModelWithBidirectionalStreamInputWriter struct {
	encoder             *eventstream.Encoder
	signer              eventStreamSigner
	stream              chan asyncInvokeModelWithBidirectionalStreamInput
	serializationBuffer *bytes.Buffer
	signingBuffer       *bytes.Buffer
	eventStream         io.WriteCloser
	done                chan struct{}
	closeOnce           sync.Once
	err                 *smithysync.OnceErr
}

func newInvokeModelWithBidirectionalStreamInputWriter(stream io.WriteCloser, encoder *eventstream.Encoder, signer eventStreamSigner) *invokeModelWithBidirectionalStreamInputWriter {
	w := &invokeModelWithBidirectionalStreamInputWriter{
		encoder:             encoder,
		signer:              signer,
		stream:              make(chan asyncInvokeModelWithBidirectionalStreamInput),
		eventStream:         stream,
		done:                make(chan struct{}),
		err:                 smithysync.NewOnceErr(),
		serializationBuffer: bytes.NewBuffer(nil),
		signingBuffer:       bytes.NewBuffer(nil),
	}

	go w.writeStream()

	return w

}

func (w *invokeModelWithBidirectionalStreamInputWriter) Send(ctx context.Context, event types.InvokeModelWithBidirectionalStreamInput) error {
	return w.send(ctx, event)
}

func (w *invokeModelWithBidirectionalStreamInputWriter) send(ctx context.Context, event types.InvokeModelWithBidirectionalStreamInput) error {
	if err := w.err.Err(); err != nil {
		return err
	}

	resultCh := make(chan error)

	wrapped := asyncInvokeModelWithBidirectionalStreamInput{
		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 *invokeModelWithBidirectionalStreamInputWriter) writeStream() {
	defer w.Close()

	for {
		select {
		case wrapper := <-w.stream:
			err := w.writeEvent(wrapper.Event)
			wrapper.ReportResult(w.done, err)
			if err != nil {
				w.err.SetError(err)
				return
			}

		case <-w.done:
			if err := w.closeStream(); err != nil {
				w.err.SetError(err)
			}
			return

		}
	}
}

func (w *invokeModelWithBidirectionalStreamInputWriter) writeEvent(event types.InvokeModelWithBidirectionalStreamInput) error {
	// serializedEvent returned bytes refers to an underlying byte buffer and must not
	// escape this writeEvent scope without first copying. Any previous bytes stored in
	// the buffer are cleared by this call.
	serializedEvent, err := w.serializeEvent(event)
	if err != nil {
		return err
	}

	// signedEvent returned bytes refers to an underlying byte buffer and must not
	// escape this writeEvent scope without first copying. Any previous bytes stored in
	// the buffer are cleared by this call.
	signedEvent, err := w.signEvent(serializedEvent)
	if err != nil {
		return err
	}

	// bytes are now copied to the underlying stream writer
	_, err = io.Copy(w.eventStream, bytes.NewReader(signedEvent))
	return err
}

func (w *invokeModelWithBidirectionalStreamInputWriter) serializeEvent(event types.InvokeModelWithBidirectionalStreamInput) ([]byte, error) {
	w.serializationBuffer.Reset()

	eventMessage := eventstream.Message{}

	if err := awsRestjson1_serializeEventStreamInvokeModelWithBidirectionalStreamInput(event, &eventMessage); err != nil {
		return nil, err
	}

	if err := w.encoder.Encode(w.serializationBuffer, eventMessage); err != nil {
		return nil, err
	}

	return w.serializationBuffer.Bytes(), nil
}

func (w *invokeModelWithBidirectionalStreamInputWriter) signEvent(payload []byte) ([]byte, error) {
	w.signingBuffer.Reset()

	date := time.Now().UTC()

	var msg eventstream.Message
	msg.Headers.Set(eventstreamapi.DateHeader, eventstream.TimestampValue(date))
	msg.Payload = payload

	var headers bytes.Buffer
	if err := eventstream.EncodeHeaders(&headers, msg.Headers); err != nil {
		return nil, err
	}

	sig, err := w.signer.GetSignature(context.Background(), headers.Bytes(), msg.Payload, date)
	if err != nil {
		return nil, err
	}

	msg.Headers.Set(eventstreamapi.ChunkSignatureHeader, eventstream.BytesValue(sig))

	if err := w.encoder.Encode(w.signingBuffer, msg); err != nil {
		return nil, err
	}

	return w.signingBuffer.Bytes(), nil
}

func (w *invokeModelWithBidirectionalStreamInputWriter) closeStream() (err error) {
	defer func() {
		if cErr := w.eventStream.Close(); cErr != nil && err == nil {
			err = cErr
		}
	}()

	// Per the protocol, a signed empty message is used to indicate the end of the stream,
	// and that no subsequent events will be sent.
	signedEvent, err := w.signEvent([]byte{})
	if err != nil {
		return err
	}

	_, err = io.Copy(w.eventStream, bytes.NewReader(signedEvent))
	return err
}

func (w *invokeModelWithBidirectionalStreamInputWriter) ErrorSet() <-chan struct{} {
	return w.err.ErrorSet()
}

func (w *invokeModelWithBidirectionalStreamInputWriter) Close() error {
	w.closeOnce.Do(w.safeClose)
	return w.Err()
}

func (w *invokeModelWithBidirectionalStreamInputWriter) safeClose() {
	close(w.done)
}

func (w *invokeModelWithBidirectionalStreamInputWriter) Err() error {
	return w.err.Err()
}

type invokeModelWithBidirectionalStreamOutputReader struct {
	stream      chan types.InvokeModelWithBidirectionalStreamOutput
	decoder     *eventstream.Decoder
	eventStream io.ReadCloser
	err         *smithysync.OnceErr
	payloadBuf  []byte
	done        chan struct{}
	closeOnce   sync.Once
}

func newInvokeModelWithBidirectionalStreamOutputReader(readCloser io.ReadCloser, decoder *eventstream.Decoder) *invokeModelWithBidirectionalStreamOutputReader {
	w := &invokeModelWithBidirectionalStreamOutputReader{
		stream:      make(chan types.InvokeModelWithBidirectionalStreamOutput),
		decoder:     decoder,
		eventStream: readCloser,
		err:         smithysync.NewOnceErr(),
		done:        make(chan struct{}),
		payloadBuf:  make([]byte, 10*1024),
	}

	go w.readEventStream()

	return w
}

func (r *invokeModelWithBidirectionalStreamOutputReader) Events() <-chan types.InvokeModelWithBidirectionalStreamOutput {
	return r.stream
}

func (r *invokeModelWithBidirectionalStreamOutputReader) readEventStream() {
	defer r.Close()
	defer close(r.stream)

	for {
		r.payloadBuf = r.payloadBuf[0:0]
		decodedMessage, err := r.decoder.Decode(r.eventStream, r.payloadBuf)
		if err != nil {
			if err == io.EOF {
				return
			}
			select {
			case <-r.done:
				return
			default:
				r.err.SetError(err)
				return
			}
		}

		event, err := r.deserializeEventMessage(&decodedMessage)
		if err != nil {
			r.err.SetError(err)
			return
		}

		select {
		case r.stream <- event:
		case <-r.done:
			return
		}

	}
}

func (r *invokeModelWithBidirectionalStreamOutputReader) deserializeEventMessage(msg *eventstream.Message) (types.InvokeModelWithBidirectionalStreamOutput, error) {
	messageType := msg.Headers.Get(eventstreamapi.MessageTypeHeader)
	if messageType == nil {
		return nil, fmt.Errorf("%s event header not present", eventstreamapi.MessageTypeHeader)
	}

	switch messageType.String() {
	case eventstreamapi.EventMessageType:
		var v types.InvokeModelWithBidirectionalStreamOutput
		if err := awsRestjson1_deserializeEventStreamInvokeModelWithBidirectionalStreamOutput(&v, msg); err != nil {
			return nil, err
		}
		return v, nil

	case eventstreamapi.ExceptionMessageType:
		return nil, awsRestjson1_deserializeEventStreamExceptionInvokeModelWithBidirectionalStreamOutput(msg)

	case eventstreamapi.ErrorMessageType:
		errorCode := "UnknownError"
		errorMessage := errorCode
		if header := msg.Headers.Get(eventstreamapi.ErrorCodeHeader); header != nil {
			errorCode = header.String()
		}
		if header := msg.Headers.Get(eventstreamapi.ErrorMessageHeader); header != nil {
			errorMessage = header.String()
		}
		return nil, &smithy.GenericAPIError{
			Code:    errorCode,
			Message: errorMessage,
		}

	default:
		mc := msg.Clone()
		return nil, &UnknownEventMessageError{
			Type:    messageType.String(),
			Message: &mc,
		}

	}
}

func (r *invokeModelWithBidirectionalStreamOutputReader) ErrorSet() <-chan struct{} {
	return r.err.ErrorSet()
}

func (r *invokeModelWithBidirectionalStreamOutputReader) Close() error {
	r.closeOnce.Do(r.safeClose)
	return r.Err()
}

func (r *invokeModelWithBidirectionalStreamOutputReader) safeClose() {
	close(r.done)
	r.eventStream.Close()

}

func (r *invokeModelWithBidirectionalStreamOutputReader) Err() error {
	return r.err.Err()
}

func (r *invokeModelWithBidirectionalStreamOutputReader) Closed() <-chan struct{} {
	return r.done
}

type responseStreamReader struct {
	stream      chan types.ResponseStream
	decoder     *eventstream.Decoder
	eventStream io.ReadCloser
	err         *smithysync.OnceErr
	payloadBuf  []byte
	done        chan struct{}
	closeOnce   sync.Once
}

func newResponseStreamReader(readCloser io.ReadCloser, decoder *eventstream.Decoder) *responseStreamReader {
	w := &responseStreamReader{
		stream:      make(chan types.ResponseStream),
		decoder:     decoder,
		eventStream: readCloser,
		err:         smithysync.NewOnceErr(),
		done:        make(chan struct{}),
		payloadBuf:  make([]byte, 10*1024),
	}

	go w.readEventStream()

	return w
}

func (r *responseStreamReader) Events() <-chan types.ResponseStream {
	return r.stream
}

func (r *responseStreamReader) readEventStream() {
	defer r.Close()
	defer close(r.stream)

	for {
		r.payloadBuf = r.payloadBuf[0:0]
		decodedMessage, err := r.decoder.Decode(r.eventStream, r.payloadBuf)
		if err != nil {
			if err == io.EOF {
				return
			}
			select {
			case <-r.done:
				return
			default:
				r.err.SetError(err)
				return
			}
		}

		event, err := r.deserializeEventMessage(&decodedMessage)
		if err != nil {
			r.err.SetError(err)
			return
		}

		select {
		case r.stream <- event:
		case <-r.done:
			return
		}

	}
}

func (r *responseStreamReader) deserializeEventMessage(msg *eventstream.Message) (types.ResponseStream, error) {
	messageType := msg.Headers.Get(eventstreamapi.MessageTypeHeader)
	if messageType == nil {
		return nil, fmt.Errorf("%s event header not present", eventstreamapi.MessageTypeHeader)
	}

	switch messageType.String() {
	case eventstreamapi.EventMessageType:
		var v types.ResponseStream
		if err := awsRestjson1_deserializeEventStreamResponseStream(&v, msg); err != nil {
			return nil, err
		}
		return v, nil

	case eventstreamapi.ExceptionMessageType:
		return nil, awsRestjson1_deserializeEventStreamExceptionResponseStream(msg)

	case eventstreamapi.ErrorMessageType:
		errorCode := "UnknownError"
		errorMessage := errorCode
		if header := msg.Headers.Get(eventstreamapi.ErrorCodeHeader); header != nil {
			errorCode = header.String()
		}
		if header := msg.Headers.Get(eventstreamapi.ErrorMessageHeader); header != nil {
			errorMessage = header.String()
		}
		return nil, &smithy.GenericAPIError{
			Code:    errorCode,
			Message: errorMessage,
		}

	default:
		mc := msg.Clone()
		return nil, &UnknownEventMessageError{
			Type:    messageType.String(),
			Message: &mc,
		}

	}
}

func (r *responseStreamReader) ErrorSet() <-chan struct{} {
	return r.err.ErrorSet()
}

func (r *responseStreamReader) Close() error {
	r.closeOnce.Do(r.safeClose)
	return r.Err()
}

func (r *responseStreamReader) safeClose() {
	close(r.done)
	r.eventStream.Close()

}

func (r *responseStreamReader) Err() error {
	return r.err.Err()
}

func (r *responseStreamReader) Closed() <-chan struct{} {
	return r.done
}

type converseStreamOutputReader struct {
	stream      chan types.ConverseStreamOutput
	decoder     *eventstream.Decoder
	eventStream io.ReadCloser
	err         *smithysync.OnceErr
	payloadBuf  []byte
	done        chan struct{}
	closeOnce   sync.Once
}

func newConverseStreamOutputReader(readCloser io.ReadCloser, decoder *eventstream.Decoder) *converseStreamOutputReader {
	w := &converseStreamOutputReader{
		stream:      make(chan types.ConverseStreamOutput),
		decoder:     decoder,
		eventStream: readCloser,
		err:         smithysync.NewOnceErr(),
		done:        make(chan struct{}),
		payloadBuf:  make([]byte, 10*1024),
	}

	go w.readEventStream()

	return w
}

func (r *converseStreamOutputReader) Events() <-chan types.ConverseStreamOutput {
	return r.stream
}

func (r *converseStreamOutputReader) readEventStream() {
	defer r.Close()
	defer close(r.stream)

	for {
		r.payloadBuf = r.payloadBuf[0:0]
		decodedMessage, err := r.decoder.Decode(r.eventStream, r.payloadBuf)
		if err != nil {
			if err == io.EOF {
				return
			}
			select {
			case <-r.done:
				return
			default:
				r.err.SetError(err)
				return
			}
		}

		event, err := r.deserializeEventMessage(&decodedMessage)
		if err != nil {
			r.err.SetError(err)
			return
		}

		select {
		case r.stream <- event:
		case <-r.done:
			return
		}

	}
}

func (r *converseStreamOutputReader) deserializeEventMessage(msg *eventstream.Message) (types.ConverseStreamOutput, error) {
	messageType := msg.Headers.Get(eventstreamapi.MessageTypeHeader)
	if messageType == nil {
		return nil, fmt.Errorf("%s event header not present", eventstreamapi.MessageTypeHeader)
	}

	switch messageType.String() {
	case eventstreamapi.EventMessageType:
		var v types.ConverseStreamOutput
		if err := awsRestjson1_deserializeEventStreamConverseStreamOutput(&v, msg); err != nil {
			return nil, err
		}
		return v, nil

	case eventstreamapi.ExceptionMessageType:
		return nil, awsRestjson1_deserializeEventStreamExceptionConverseStreamOutput(msg)

	case eventstreamapi.ErrorMessageType:
		errorCode := "UnknownError"
		errorMessage := errorCode
		if header := msg.Headers.Get(eventstreamapi.ErrorCodeHeader); header != nil {
			errorCode = header.String()
		}
		if header := msg.Headers.Get(eventstreamapi.ErrorMessageHeader); header != nil {
			errorMessage = header.String()
		}
		return nil, &smithy.GenericAPIError{
			Code:    errorCode,
			Message: errorMessage,
		}

	default:
		mc := msg.Clone()
		return nil, &UnknownEventMessageError{
			Type:    messageType.String(),
			Message: &mc,
		}

	}
}

func (r *converseStreamOutputReader) ErrorSet() <-chan struct{} {
	return r.err.ErrorSet()
}

func (r *converseStreamOutputReader) Close() error {
	r.closeOnce.Do(r.safeClose)
	return r.Err()
}

func (r *converseStreamOutputReader) safeClose() {
	close(r.done)
	r.eventStream.Close()

}

func (r *converseStreamOutputReader) Err() error {
	return r.err.Err()
}

func (r *converseStreamOutputReader) Closed() <-chan struct{} {
	return r.done
}

type awsRestjson1_deserializeOpEventStreamConverseStream struct {
	LogEventStreamWrites bool
	LogEventStreamReads  bool
}

func (*awsRestjson1_deserializeOpEventStreamConverseStream) ID() string {
	return "OperationEventStreamDeserializer"
}

func (m *awsRestjson1_deserializeOpEventStreamConverseStream) HandleDeserialize(ctx context.Context, in middleware.DeserializeInput, next middleware.DeserializeHandler) (
	out middleware.DeserializeOutput, metadata middleware.Metadata, err error,
) {
	defer func() {
		if err == nil {
			return
		}
		m.closeResponseBody(out)
	}()

	logger := middleware.GetLogger(ctx)

	request, ok := in.Request.(*smithyhttp.Request)
	if !ok {
		return out, metadata, fmt.Errorf("unknown transport type: %T", in.Request)
	}
	_ = request

	out, metadata, err = next.HandleDeserialize(ctx, in)
	if err != nil {
		return out, metadata, err
	}

	deserializeOutput, ok := out.RawResponse.(*smithyhttp.Response)
	if !ok {
		return out, metadata, fmt.Errorf("unknown transport type: %T", out.RawResponse)
	}
	_ = deserializeOutput

	output, ok := out.Result.(*ConverseStreamOutput)
	if out.Result != nil && !ok {
		return out, metadata, fmt.Errorf("unexpected output result type: %T", out.Result)
	} else if out.Result == nil {
		output = &ConverseStreamOutput{}
		out.Result = output
	}

	eventReader := newConverseStreamOutputReader(
		deserializeOutput.Body,
		eventstream.NewDecoder(func(options *eventstream.DecoderOptions) {
			options.Logger = logger
			options.LogMessages = m.LogEventStreamReads

		}),
	)
	defer func() {
		if err == nil {
			return
		}
		_ = eventReader.Close()
	}()

	output.eventStream = NewConverseStreamEventStream(func(stream *ConverseStreamEventStream) {
		stream.Reader = eventReader
	})

	go output.eventStream.waitStreamClose()

	return out, metadata, nil
}

func (*awsRestjson1_deserializeOpEventStreamConverseStream) closeResponseBody(out middleware.DeserializeOutput) {
	if resp, ok := out.RawResponse.(*smithyhttp.Response); ok && resp != nil && resp.Body != nil {
		_, _ = io.Copy(ioutil.Discard, resp.Body)
		_ = resp.Body.Close()
	}
}

func addEventStreamConverseStreamMiddleware(stack *middleware.Stack, options Options) error {
	if err := stack.Deserialize.Insert(&awsRestjson1_deserializeOpEventStreamConverseStream{
		LogEventStreamWrites: options.ClientLogMode.IsRequestEventMessage(),
		LogEventStreamReads:  options.ClientLogMode.IsResponseEventMessage(),
	}, "OperationDeserializer", middleware.Before); err != nil {
		return err
	}
	return nil

}

type awsRestjson1_deserializeOpEventStreamInvokeModelWithBidirectionalStream struct {
	LogEventStreamWrites bool
	LogEventStreamReads  bool

	existingResult *InvokeModelWithBidirectionalStreamOutput
	asyncResult    chan deserializeResult
}

func (*awsRestjson1_deserializeOpEventStreamInvokeModelWithBidirectionalStream) ID() string {
	return "OperationEventStreamDeserializer"
}

func (m *awsRestjson1_deserializeOpEventStreamInvokeModelWithBidirectionalStream) HandleDeserialize(ctx context.Context, in middleware.DeserializeInput, next middleware.DeserializeHandler) (
	out middleware.DeserializeOutput, metadata middleware.Metadata, err error,
) {
	defer func() {
		if err == nil {
			return
		}
		m.closeResponseBody(out)
	}()

	logger := middleware.GetLogger(ctx)

	request, ok := in.Request.(*smithyhttp.Request)
	if !ok {
		return out, metadata, fmt.Errorf("unknown transport type: %T", in.Request)
	}
	_ = request

	if err := eventstreamapi.ApplyHTTPTransportFixes(request); err != nil {
		return out, metadata, err
	}

	requestSignature, err := v4.GetSignedRequestSignature(request.Request)
	if err != nil {
		return out, metadata, fmt.Errorf("failed to get event stream seed signature: %v", err)
	}

	identity := getIdentity(ctx)
	if identity == nil {
		return out, metadata, fmt.Errorf("no identity")
	}

	creds, ok := identity.(*internalauthsmithy.CredentialsAdapter)
	if !ok {
		return out, metadata, fmt.Errorf("identity is not sigv4 credentials")
	}

	rscheme := getResolvedAuthScheme(ctx)
	if rscheme == nil {
		return out, metadata, fmt.Errorf("no resolved auth scheme")
	}

	name, ok := smithyhttp.GetSigV4SigningName(&rscheme.SignerProperties)
	if !ok {
		return out, metadata, fmt.Errorf("no sigv4 signing name")
	}

	region, ok := smithyhttp.GetSigV4SigningRegion(&rscheme.SignerProperties)
	if !ok {
		return out, metadata, fmt.Errorf("no sigv4 signing region")
	}

	signer := v4.NewStreamSigner(creds.Credentials, name, region, requestSignature)

	eventWriter := newInvokeModelWithBidirectionalStreamInputWriter(
		eventstreamapi.GetInputStreamWriter(ctx),
		eventstream.NewEncoder(func(options *eventstream.EncoderOptions) {
			options.Logger = logger
			options.LogMessages = m.LogEventStreamWrites

		}),
		signer,
	)
	defer func() {
		if err == nil {
			return
		}
		_ = eventWriter.Close()
	}()

	//  storing existing output instead of creating a new one
	var output *InvokeModelWithBidirectionalStreamOutput
	var asyncResult chan deserializeResult

	existingResult := m.existingResult
	asyncResult = m.asyncResult
	if existingResult == nil {
		// Create async result channel
		asyncResult = make(chan deserializeResult, 1)
		asyncReader := newAsyncEventStreamReader(asyncResult)
		eventReader := newInvokeModelWithBidirectionalStreamOutputReader(
			asyncReader.pipeReader,
			eventstream.NewDecoder(func(options *eventstream.DecoderOptions) {
				options.Logger = logger
				options.Logger = logger
				options.LogMessages = m.LogEventStreamReads
			}),
		)
		output = &InvokeModelWithBidirectionalStreamOutput{}
		output.eventStream = NewInvokeModelWithBidirectionalStreamEventStream(func(stream *InvokeModelWithBidirectionalStreamEventStream) {
			stream.Writer = eventWriter
			stream.Reader = eventReader
		})
		output.initialReply = make(chan InvokeModelWithBidirectionalStreamInitialReply, 1)

		go output.eventStream.waitStreamClose()

		m.existingResult = output
		m.asyncResult = asyncResult
	}

	ctxCh := ctx.Value(partialResultChan{})
	if ctxCh == nil {
		return out, metadata, fmt.Errorf("Expected a result channel to be stored in the contex, got none")
	}

	prc, ok := ctxCh.(chan PartialResult)
	if !ok {
		return out, metadata, fmt.Errorf("async channel expected to be of type `chan partialResult`, got: %T", ctxCh)
	}
	// Drain existing results from the channel in case this is a retry
	select {
	case <-prc:
	default:
	}
	partial := PartialResult{
		Output:   output,
		Metadata: middleware.Metadata{},
		Error:    nil,
	}
	prc <- partial

	out, metadata, err = next.HandleDeserialize(ctx, in)

	if err == nil {
		// Extract actual response and create real reader
		resp := out.RawResponse.(*smithyhttp.Response)
		asyncResult <- deserializeResult{reader: resp.Body, err: nil}
	} else {
		asyncResult <- deserializeResult{reader: nil, err: err}
	}
	middleware.AddEventStreamOutputToMetadata(&metadata, m.existingResult)
	return out, metadata, err
}

func (*awsRestjson1_deserializeOpEventStreamInvokeModelWithBidirectionalStream) closeResponseBody(out middleware.DeserializeOutput) {
	if resp, ok := out.RawResponse.(*smithyhttp.Response); ok && resp != nil && resp.Body != nil {
		_, _ = io.Copy(ioutil.Discard, resp.Body)
		_ = resp.Body.Close()
	}
}

func addEventStreamInvokeModelWithBidirectionalStreamMiddleware(stack *middleware.Stack, options Options) error {
	if err := stack.Deserialize.Insert(&awsRestjson1_deserializeOpEventStreamInvokeModelWithBidirectionalStream{
		LogEventStreamWrites: options.ClientLogMode.IsRequestEventMessage(),
		LogEventStreamReads:  options.ClientLogMode.IsResponseEventMessage(),
	}, "OperationDeserializer", middleware.Before); err != nil {
		return err
	}
	return nil

}

type awsRestjson1_deserializeOpEventStreamInvokeModelWithResponseStream struct {
	LogEventStreamWrites bool
	LogEventStreamReads  bool
}

func (*awsRestjson1_deserializeOpEventStreamInvokeModelWithResponseStream) ID() string {
	return "OperationEventStreamDeserializer"
}

func (m *awsRestjson1_deserializeOpEventStreamInvokeModelWithResponseStream) HandleDeserialize(ctx context.Context, in middleware.DeserializeInput, next middleware.DeserializeHandler) (
	out middleware.DeserializeOutput, metadata middleware.Metadata, err error,
) {
	defer func() {
		if err == nil {
			return
		}
		m.closeResponseBody(out)
	}()

	logger := middleware.GetLogger(ctx)

	request, ok := in.Request.(*smithyhttp.Request)
	if !ok {
		return out, metadata, fmt.Errorf("unknown transport type: %T", in.Request)
	}
	_ = request

	out, metadata, err = next.HandleDeserialize(ctx, in)
	if err != nil {
		return out, metadata, err
	}

	deserializeOutput, ok := out.RawResponse.(*smithyhttp.Response)
	if !ok {
		return out, metadata, fmt.Errorf("unknown transport type: %T", out.RawResponse)
	}
	_ = deserializeOutput

	output, ok := out.Result.(*InvokeModelWithResponseStreamOutput)
	if out.Result != nil && !ok {
		return out, metadata, fmt.Errorf("unexpected output result type: %T", out.Result)
	} else if out.Result == nil {
		output = &InvokeModelWithResponseStreamOutput{}
		out.Result = output
	}

	eventReader := newResponseStreamReader(
		deserializeOutput.Body,
		eventstream.NewDecoder(func(options *eventstream.DecoderOptions) {
			options.Logger = logger
			options.LogMessages = m.LogEventStreamReads

		}),
	)
	defer func() {
		if err == nil {
			return
		}
		_ = eventReader.Close()
	}()

	output.eventStream = NewInvokeModelWithResponseStreamEventStream(func(stream *InvokeModelWithResponseStreamEventStream) {
		stream.Reader = eventReader
	})

	go output.eventStream.waitStreamClose()

	return out, metadata, nil
}

func (*awsRestjson1_deserializeOpEventStreamInvokeModelWithResponseStream) closeResponseBody(out middleware.DeserializeOutput) {
	if resp, ok := out.RawResponse.(*smithyhttp.Response); ok && resp != nil && resp.Body != nil {
		_, _ = io.Copy(ioutil.Discard, resp.Body)
		_ = resp.Body.Close()
	}
}

func addEventStreamInvokeModelWithResponseStreamMiddleware(stack *middleware.Stack, options Options) error {
	if err := stack.Deserialize.Insert(&awsRestjson1_deserializeOpEventStreamInvokeModelWithResponseStream{
		LogEventStreamWrites: options.ClientLogMode.IsRequestEventMessage(),
		LogEventStreamReads:  options.ClientLogMode.IsResponseEventMessage(),
	}, "OperationDeserializer", middleware.Before); err != nil {
		return err
	}
	return nil

}

// UnknownEventMessageError 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 UnknownEventMessageError struct {
	Type    string
	Message *eventstream.Message
}

// Error retruns the error message string.
func (e *UnknownEventMessageError) Error() string {
	return "unknown event stream message type, " + e.Type
}

func setSafeEventStreamClientLogMode(o *Options, operation string) {
	switch operation {
	case "ConverseStream":
		toggleEventStreamClientLogMode(o, false, true)
		return

	case "InvokeModelWithBidirectionalStream":
		toggleEventStreamClientLogMode(o, true, true)
		return

	case "InvokeModelWithResponseStream":
		toggleEventStreamClientLogMode(o, false, true)
		return

	default:
		return

	}
}
func toggleEventStreamClientLogMode(o *Options, request, response bool) {
	mode := o.ClientLogMode

	if request && mode.IsRequestWithBody() {
		mode.ClearRequestWithBody()
		mode |= aws.LogRequest
	}

	if response && mode.IsResponseWithBody() {
		mode.ClearResponseWithBody()
		mode |= aws.LogResponse
	}

	o.ClientLogMode = mode

}