diff --git a/generated/src/aws-cpp-sdk-bedrock-runtime/include/aws/bedrock-runtime/model/InvokeModelWithBidirectionalStreamInput.h b/generated/src/aws-cpp-sdk-bedrock-runtime/include/aws/bedrock-runtime/model/InvokeModelWithBidirectionalStreamInput.h index cac2f3ad194..5f81a9acb5b 100644 --- a/generated/src/aws-cpp-sdk-bedrock-runtime/include/aws/bedrock-runtime/model/InvokeModelWithBidirectionalStreamInput.h +++ b/generated/src/aws-cpp-sdk-bedrock-runtime/include/aws/bedrock-runtime/model/InvokeModelWithBidirectionalStreamInput.h @@ -7,6 +7,7 @@ #include #include #include +#include #include @@ -22,6 +23,9 @@ namespace Model { */ class AWS_BEDROCKRUNTIME_API InvokeModelWithBidirectionalStreamInput : public Aws::Utils::Event::EventEncoderStream { public: + InvokeModelWithBidirectionalStreamInput() = default; + explicit InvokeModelWithBidirectionalStreamInput(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} InvokeModelWithBidirectionalStreamInput& WriteBidirectionalInputPayloadPart(const BidirectionalInputPayloadPart& value) { Aws::Utils::Event::Message msg; msg.InsertEventHeader(":message-type", Aws::String("event")); diff --git a/generated/src/aws-cpp-sdk-bedrock-runtime/source/BedrockRuntimeClient.cpp b/generated/src/aws-cpp-sdk-bedrock-runtime/source/BedrockRuntimeClient.cpp index 46a0495f910..f69c499bea2 100644 --- a/generated/src/aws-cpp-sdk-bedrock-runtime/source/BedrockRuntimeClient.cpp +++ b/generated/src/aws-cpp-sdk-bedrock-runtime/source/BedrockRuntimeClient.cpp @@ -30,6 +30,7 @@ #include #include #include +#include #include #include #include @@ -387,6 +388,30 @@ void BedrockRuntimeClient::InvokeModelWithBidirectionalStreamAsync( resolvedEndpoint.AddPathSegments("/invoke-with-bidirectional-stream"); }; +#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, m_httpClient); + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + request.SetBody(eventEncoderStream); + + auto requestCopy = Aws::MakeShared(ALLOCATION_TAG, request); + + auto authCallback = [&](std::shared_ptr ctx) -> void { + eventEncoderStream->SetSigningCallback([this, ctx, eventEncoderStream](Aws::Utils::Event::Message& message, Aws::String& seed) -> bool { + auto outcome = SignEventMessage(message, seed, ctx); + return outcome.IsSuccess(); + }); + }; + + auto asyncTask = smithy::client::CreateSmithyBidirectionalWriteDataTask( + this, requestCopy, handler, handlerContext, eventEncoderStream, writeDataStreamBuf, std::move(endpointCallback), + std::move(authCallback)); + auto sem = asyncTask.GetSemaphore(); + m_clientConfiguration.executor->Submit(std::move(asyncTask)); + sem->WaitOne(); + streamReadyHandler(*eventEncoderStream); +#else + // Pull-based path auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); auto authCallback = [&](std::shared_ptr ctx) -> void { eventEncoderStream->SetSigningCallback([this, ctx, eventEncoderStream](Aws::Utils::Event::Message& message, Aws::String& seed) -> bool { @@ -395,8 +420,8 @@ void BedrockRuntimeClient::InvokeModelWithBidirectionalStreamAsync( }); }; auto requestCopy = Aws::MakeShared("InvokeModelWithBidirectionalStream", request); - requestCopy->SetBody(eventEncoderStream); // this becomes the body of the request - request.SetBody(eventEncoderStream); // this becomes the body of the request + requestCopy->SetBody(eventEncoderStream); + request.SetBody(eventEncoderStream); auto asyncTask = smithy::client::CreateSmithyBidirectionalEventStreamTask( this, requestCopy, handler, handlerContext, eventEncoderStream, endpointCallback, authCallback); @@ -404,6 +429,7 @@ void BedrockRuntimeClient::InvokeModelWithBidirectionalStreamAsync( m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } InvokeModelWithResponseStreamOutcome BedrockRuntimeClient::InvokeModelWithResponseStream( InvokeModelWithResponseStreamRequest& request) const { diff --git a/generated/src/aws-cpp-sdk-connecthealth/include/aws/connecthealth/model/MedicalScribeInputStream.h b/generated/src/aws-cpp-sdk-connecthealth/include/aws/connecthealth/model/MedicalScribeInputStream.h index fcad785da60..b97e3268041 100644 --- a/generated/src/aws-cpp-sdk-connecthealth/include/aws/connecthealth/model/MedicalScribeInputStream.h +++ b/generated/src/aws-cpp-sdk-connecthealth/include/aws/connecthealth/model/MedicalScribeInputStream.h @@ -9,6 +9,7 @@ #include #include #include +#include #include @@ -24,6 +25,9 @@ namespace Model { */ class AWS_CONNECTHEALTH_API MedicalScribeInputStream : public Aws::Utils::Event::EventEncoderStream { public: + MedicalScribeInputStream() = default; + explicit MedicalScribeInputStream(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} MedicalScribeInputStream& WriteMedicalScribeAudioEvent(const MedicalScribeAudioEvent& value) { Aws::Utils::Event::Message msg; msg.InsertEventHeader(":message-type", Aws::String("event")); diff --git a/generated/src/aws-cpp-sdk-connecthealth/source/ConnectHealthClient.cpp b/generated/src/aws-cpp-sdk-connecthealth/source/ConnectHealthClient.cpp index 89a3c59ceb5..36d6c860208 100644 --- a/generated/src/aws-cpp-sdk-connecthealth/source/ConnectHealthClient.cpp +++ b/generated/src/aws-cpp-sdk-connecthealth/source/ConnectHealthClient.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -517,10 +518,43 @@ void ConnectHealthClient::StartMedicalScribeListeningSessionAsync( } endpointResolutionOutcome.GetResult().AddPathSegments("/medical-scribe-stream/"); +#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); + + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + eventEncoderStream->SetSigner(signer); + + auto requestCopy = Aws::MakeShared(ALLOCATION_TAG, request); + request.SetInputStream(eventEncoderStream); + + auto& endpoint = endpointResolutionOutcome.GetResult(); + auto httpRequest = + CreateHttpRequest(endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_POST, Aws::Utils::Stream::DefaultResponseStreamFactoryMethod); + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + BuildHttpRequest(*requestCopy, httpRequest); + + if (!signer->SignRequest(*httpRequest, nullptr, nullptr, true)) { + handler(this, request, + StartMedicalScribeListeningSessionOutcome( + Aws::Client::AWSError(CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + handlerContext); + return; + } + eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + Aws::Client::SubmitBidirectionalStreamingRequest( + this, request, requestCopy, eventEncoderStream, writeDataStreamBuf, httpRequest, m_clientConfiguration.executor.get(), + streamReadyHandler, handler, handlerContext); +#else + // Pull-based path (curl/WinHTTP) auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER)); auto requestCopy = Aws::MakeShared("StartMedicalScribeListeningSession", request); - requestCopy->SetInputStream(eventEncoderStream); // this becomes the body of the request + requestCopy->SetInputStream(eventEncoderStream); request.SetInputStream(eventEncoderStream); auto asyncTask = CreateBidirectionalEventStreamTask( @@ -529,6 +563,7 @@ void ConnectHealthClient::StartMedicalScribeListeningSessionAsync( m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } StartPatientInsightsJobOutcome ConnectHealthClient::StartPatientInsightsJob(const StartPatientInsightsJobRequest& request) const { if (!request.DomainIdHasBeenSet()) { diff --git a/generated/src/aws-cpp-sdk-lexv2-runtime/include/aws/lexv2-runtime/model/StartConversationRequestEventStream.h b/generated/src/aws-cpp-sdk-lexv2-runtime/include/aws/lexv2-runtime/model/StartConversationRequestEventStream.h index 296f9f41f6a..f8bbabb7c41 100644 --- a/generated/src/aws-cpp-sdk-lexv2-runtime/include/aws/lexv2-runtime/model/StartConversationRequestEventStream.h +++ b/generated/src/aws-cpp-sdk-lexv2-runtime/include/aws/lexv2-runtime/model/StartConversationRequestEventStream.h @@ -5,6 +5,7 @@ #pragma once #include +#include #include #include #include @@ -27,6 +28,9 @@ namespace Model { */ class AWS_LEXRUNTIMEV2_API StartConversationRequestEventStream : public Aws::Utils::Event::EventEncoderStream { public: + StartConversationRequestEventStream() = default; + explicit StartConversationRequestEventStream(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} StartConversationRequestEventStream& WriteConfigurationEvent(const ConfigurationEvent& value) { Aws::Utils::Event::Message msg; msg.InsertEventHeader(":message-type", Aws::String("event")); diff --git a/generated/src/aws-cpp-sdk-lexv2-runtime/source/LexRuntimeV2Client.cpp b/generated/src/aws-cpp-sdk-lexv2-runtime/source/LexRuntimeV2Client.cpp index 0330dda83d5..38344975c9e 100644 --- a/generated/src/aws-cpp-sdk-lexv2-runtime/source/LexRuntimeV2Client.cpp +++ b/generated/src/aws-cpp-sdk-lexv2-runtime/source/LexRuntimeV2Client.cpp @@ -5,6 +5,7 @@ #include #include +#include #include #include #include @@ -489,10 +490,43 @@ void LexRuntimeV2Client::StartConversationAsync(Model::StartConversationRequest& endpointResolutionOutcome.GetResult().AddPathSegment(request.GetSessionId()); endpointResolutionOutcome.GetResult().AddPathSegments("/conversation"); +#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); + + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + eventEncoderStream->SetSigner(signer); + + auto requestCopy = Aws::MakeShared(ALLOCATION_TAG, request); + request.SetRequestEventStream(eventEncoderStream); + + auto& endpoint = endpointResolutionOutcome.GetResult(); + auto httpRequest = + CreateHttpRequest(endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_POST, Aws::Utils::Stream::DefaultResponseStreamFactoryMethod); + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + BuildHttpRequest(*requestCopy, httpRequest); + + if (!signer->SignRequest(*httpRequest, nullptr, nullptr, true)) { + handler(this, request, + StartConversationOutcome( + Aws::Client::AWSError(CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + handlerContext); + return; + } + eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + Aws::Client::SubmitBidirectionalStreamingRequest( + this, request, requestCopy, eventEncoderStream, writeDataStreamBuf, httpRequest, m_clientConfiguration.executor.get(), + streamReadyHandler, handler, handlerContext); +#else + // Pull-based path (curl/WinHTTP) auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER)); auto requestCopy = Aws::MakeShared("StartConversation", request); - requestCopy->SetRequestEventStream(eventEncoderStream); // this becomes the body of the request + requestCopy->SetRequestEventStream(eventEncoderStream); request.SetRequestEventStream(eventEncoderStream); auto asyncTask = CreateBidirectionalEventStreamTask(this, endpointResolutionOutcome.GetResultWithOwnership(), @@ -501,4 +535,5 @@ void LexRuntimeV2Client::StartConversationAsync(Model::StartConversationRequest& m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } diff --git a/generated/src/aws-cpp-sdk-polly/include/aws/polly/model/StartSpeechSynthesisStreamActionStream.h b/generated/src/aws-cpp-sdk-polly/include/aws/polly/model/StartSpeechSynthesisStreamActionStream.h index 265cf3e27ca..27961f991f3 100644 --- a/generated/src/aws-cpp-sdk-polly/include/aws/polly/model/StartSpeechSynthesisStreamActionStream.h +++ b/generated/src/aws-cpp-sdk-polly/include/aws/polly/model/StartSpeechSynthesisStreamActionStream.h @@ -5,6 +5,7 @@ #pragma once #include +#include #include #include #include @@ -23,6 +24,9 @@ namespace Model { */ class AWS_POLLY_API StartSpeechSynthesisStreamActionStream : public Aws::Utils::Event::EventEncoderStream { public: + StartSpeechSynthesisStreamActionStream() = default; + explicit StartSpeechSynthesisStreamActionStream(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} StartSpeechSynthesisStreamActionStream& WriteTextEvent(const TextEvent& value) { Aws::Utils::Event::Message msg; msg.InsertEventHeader(":message-type", Aws::String("event")); diff --git a/generated/src/aws-cpp-sdk-polly/source/PollyClient.cpp b/generated/src/aws-cpp-sdk-polly/source/PollyClient.cpp index 20b26d79989..bd64449838b 100644 --- a/generated/src/aws-cpp-sdk-polly/source/PollyClient.cpp +++ b/generated/src/aws-cpp-sdk-polly/source/PollyClient.cpp @@ -5,6 +5,7 @@ #include #include +#include #include #include #include @@ -340,10 +341,43 @@ void PollyClient::StartSpeechSynthesisStreamAsync(Model::StartSpeechSynthesisStr } endpointResolutionOutcome.GetResult().AddPathSegments("/v1/synthesisStream"); +#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); + + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + eventEncoderStream->SetSigner(signer); + + auto requestCopy = Aws::MakeShared(ALLOCATION_TAG, request); + request.SetActionStream(eventEncoderStream); + + auto& endpoint = endpointResolutionOutcome.GetResult(); + auto httpRequest = + CreateHttpRequest(endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_POST, Aws::Utils::Stream::DefaultResponseStreamFactoryMethod); + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + BuildHttpRequest(*requestCopy, httpRequest); + + if (!signer->SignRequest(*httpRequest, nullptr, nullptr, true)) { + handler(this, request, + StartSpeechSynthesisStreamOutcome( + Aws::Client::AWSError(CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + handlerContext); + return; + } + eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + Aws::Client::SubmitBidirectionalStreamingRequest( + this, request, requestCopy, eventEncoderStream, writeDataStreamBuf, httpRequest, m_clientConfiguration.executor.get(), + streamReadyHandler, handler, handlerContext); +#else + // Pull-based path (curl/WinHTTP) auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER)); auto requestCopy = Aws::MakeShared("StartSpeechSynthesisStream", request); - requestCopy->SetActionStream(eventEncoderStream); // this becomes the body of the request + requestCopy->SetActionStream(eventEncoderStream); request.SetActionStream(eventEncoderStream); auto asyncTask = CreateBidirectionalEventStreamTask( @@ -352,6 +386,7 @@ void PollyClient::StartSpeechSynthesisStreamAsync(Model::StartSpeechSynthesisStr m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } StartSpeechSynthesisTaskOutcome PollyClient::StartSpeechSynthesisTask(const StartSpeechSynthesisTaskRequest& request) const { auto uriResolver = [&](Aws::Endpoint::ResolveEndpointOutcome& endpointResolutionOutcome) { diff --git a/generated/src/aws-cpp-sdk-qbusiness/include/aws/qbusiness/model/ChatInputStream.h b/generated/src/aws-cpp-sdk-qbusiness/include/aws/qbusiness/model/ChatInputStream.h index df7b5cf4c5e..84dfe3a708b 100644 --- a/generated/src/aws-cpp-sdk-qbusiness/include/aws/qbusiness/model/ChatInputStream.h +++ b/generated/src/aws-cpp-sdk-qbusiness/include/aws/qbusiness/model/ChatInputStream.h @@ -6,6 +6,7 @@ #pragma once #include #include +#include #include #include #include @@ -28,6 +29,9 @@ namespace Model { */ class AWS_QBUSINESS_API ChatInputStream : public Aws::Utils::Event::EventEncoderStream { public: + ChatInputStream() = default; + explicit ChatInputStream(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} ChatInputStream& WriteConfigurationEvent(const ConfigurationEvent& value) { Aws::Utils::Event::Message msg; msg.InsertEventHeader(":message-type", Aws::String("event")); diff --git a/generated/src/aws-cpp-sdk-qbusiness/source/QBusinessClient.cpp b/generated/src/aws-cpp-sdk-qbusiness/source/QBusinessClient.cpp index 56cb4cdf5de..f9428b5305d 100644 --- a/generated/src/aws-cpp-sdk-qbusiness/source/QBusinessClient.cpp +++ b/generated/src/aws-cpp-sdk-qbusiness/source/QBusinessClient.cpp @@ -5,6 +5,7 @@ #include #include +#include #include #include #include @@ -393,10 +394,41 @@ void QBusinessClient::ChatAsync(Model::ChatRequest& request, const ChatStreamRea endpointResolutionOutcome.GetResult().AddPathSegment(request.GetApplicationId()); endpointResolutionOutcome.GetResult().AddPathSegments("/conversations"); +#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); + + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + eventEncoderStream->SetSigner(signer); + + auto requestCopy = Aws::MakeShared(ALLOCATION_TAG, request); + request.SetInputStream(eventEncoderStream); + + auto& endpoint = endpointResolutionOutcome.GetResult(); + auto httpRequest = + CreateHttpRequest(endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_POST, Aws::Utils::Stream::DefaultResponseStreamFactoryMethod); + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + BuildHttpRequest(*requestCopy, httpRequest); + + if (!signer->SignRequest(*httpRequest, nullptr, nullptr, true)) { + handler(this, request, + ChatOutcome(Aws::Client::AWSError(CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + handlerContext); + return; + } + eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + Aws::Client::SubmitBidirectionalStreamingRequest( + this, request, requestCopy, eventEncoderStream, writeDataStreamBuf, httpRequest, m_clientConfiguration.executor.get(), + streamReadyHandler, handler, handlerContext); +#else + // Pull-based path (curl/WinHTTP) auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER)); auto requestCopy = Aws::MakeShared("Chat", request); - requestCopy->SetInputStream(eventEncoderStream); // this becomes the body of the request + requestCopy->SetInputStream(eventEncoderStream); request.SetInputStream(eventEncoderStream); auto asyncTask = CreateBidirectionalEventStreamTask(this, endpointResolutionOutcome.GetResultWithOwnership(), requestCopy, @@ -405,6 +437,7 @@ void QBusinessClient::ChatAsync(Model::ChatRequest& request, const ChatStreamRea m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } ChatSyncOutcome QBusinessClient::ChatSync(const ChatSyncRequest& request) const { if (!request.ApplicationIdHasBeenSet()) { diff --git a/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/include/aws/sagemaker-runtime-http2/model/RequestStreamEvent.h b/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/include/aws/sagemaker-runtime-http2/model/RequestStreamEvent.h index 9555a057468..451ebedc8c3 100644 --- a/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/include/aws/sagemaker-runtime-http2/model/RequestStreamEvent.h +++ b/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/include/aws/sagemaker-runtime-http2/model/RequestStreamEvent.h @@ -5,6 +5,7 @@ #pragma once #include +#include #include #include @@ -21,6 +22,9 @@ namespace Model { */ class AWS_SAGEMAKERRUNTIMEHTTP2_API RequestStreamEvent : public Aws::Utils::Event::EventEncoderStream { public: + RequestStreamEvent() = default; + explicit RequestStreamEvent(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} RequestStreamEvent& WriteRequestPayloadPart(const RequestPayloadPart& value) { Aws::Utils::Event::Message msg; if (!value.GetBytes().empty()) { diff --git a/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/source/SageMakerRuntimeHTTP2Client.cpp b/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/source/SageMakerRuntimeHTTP2Client.cpp index 0cb4034ca22..658e3e28a80 100644 --- a/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/source/SageMakerRuntimeHTTP2Client.cpp +++ b/generated/src/aws-cpp-sdk-sagemaker-runtime-http2/source/SageMakerRuntimeHTTP2Client.cpp @@ -5,6 +5,7 @@ #include #include +#include #include #include #include @@ -227,10 +228,43 @@ void SageMakerRuntimeHTTP2Client::InvokeEndpointWithBidirectionalStreamAsync( endpointResolutionOutcome.GetResult().AddPathSegment(request.GetEndpointName()); endpointResolutionOutcome.GetResult().AddPathSegments("/invocations-bidirectional-stream"); +#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); + + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + eventEncoderStream->SetSigner(signer); + + auto requestCopy = Aws::MakeShared(ALLOCATION_TAG, request); + request.SetBody(eventEncoderStream); + + auto& endpoint = endpointResolutionOutcome.GetResult(); + auto httpRequest = + CreateHttpRequest(endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_POST, Aws::Utils::Stream::DefaultResponseStreamFactoryMethod); + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + BuildHttpRequest(*requestCopy, httpRequest); + + if (!signer->SignRequest(*httpRequest, nullptr, nullptr, true)) { + handler(this, request, + InvokeEndpointWithBidirectionalStreamOutcome( + Aws::Client::AWSError(CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + handlerContext); + return; + } + eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + Aws::Client::SubmitBidirectionalStreamingRequest( + this, request, requestCopy, eventEncoderStream, writeDataStreamBuf, httpRequest, m_clientConfiguration.executor.get(), + streamReadyHandler, handler, handlerContext); +#else + // Pull-based path (curl/WinHTTP) auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER)); auto requestCopy = Aws::MakeShared("InvokeEndpointWithBidirectionalStream", request); - requestCopy->SetBody(eventEncoderStream); // this becomes the body of the request + requestCopy->SetBody(eventEncoderStream); request.SetBody(eventEncoderStream); auto asyncTask = CreateBidirectionalEventStreamTask( @@ -239,4 +273,5 @@ void SageMakerRuntimeHTTP2Client::InvokeEndpointWithBidirectionalStreamAsync( m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } diff --git a/generated/src/aws-cpp-sdk-transcribestreaming/include/aws/transcribestreaming/model/AudioStream.h b/generated/src/aws-cpp-sdk-transcribestreaming/include/aws/transcribestreaming/model/AudioStream.h index 74cb2ce4701..9b2eb52b7fa 100644 --- a/generated/src/aws-cpp-sdk-transcribestreaming/include/aws/transcribestreaming/model/AudioStream.h +++ b/generated/src/aws-cpp-sdk-transcribestreaming/include/aws/transcribestreaming/model/AudioStream.h @@ -5,6 +5,7 @@ #pragma once #include +#include #include #include #include @@ -25,6 +26,9 @@ namespace Model { */ class AWS_TRANSCRIBESTREAMINGSERVICE_API AudioStream : public Aws::Utils::Event::EventEncoderStream { public: + AudioStream() = default; + explicit AudioStream(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} AudioStream& WriteAudioEvent(const AudioEvent& value) { Aws::Utils::Event::Message msg; if (!value.GetAudioChunk().empty()) { diff --git a/generated/src/aws-cpp-sdk-transcribestreaming/include/aws/transcribestreaming/model/MedicalScribeInputStream.h b/generated/src/aws-cpp-sdk-transcribestreaming/include/aws/transcribestreaming/model/MedicalScribeInputStream.h index 7b54ddc6a6e..06d002ce419 100644 --- a/generated/src/aws-cpp-sdk-transcribestreaming/include/aws/transcribestreaming/model/MedicalScribeInputStream.h +++ b/generated/src/aws-cpp-sdk-transcribestreaming/include/aws/transcribestreaming/model/MedicalScribeInputStream.h @@ -5,6 +5,7 @@ #pragma once #include +#include #include #include #include @@ -29,6 +30,9 @@ namespace Model { */ class AWS_TRANSCRIBESTREAMINGSERVICE_API MedicalScribeInputStream : public Aws::Utils::Event::EventEncoderStream { public: + MedicalScribeInputStream() = default; + explicit MedicalScribeInputStream(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} MedicalScribeInputStream& WriteMedicalScribeAudioEvent(const MedicalScribeAudioEvent& value) { Aws::Utils::Event::Message msg; if (!value.GetAudioChunk().empty()) { diff --git a/generated/src/aws-cpp-sdk-transcribestreaming/source/TranscribeStreamingServiceClient.cpp b/generated/src/aws-cpp-sdk-transcribestreaming/source/TranscribeStreamingServiceClient.cpp index 31e1186517b..109d7be3165 100644 --- a/generated/src/aws-cpp-sdk-transcribestreaming/source/TranscribeStreamingServiceClient.cpp +++ b/generated/src/aws-cpp-sdk-transcribestreaming/source/TranscribeStreamingServiceClient.cpp @@ -5,6 +5,7 @@ #include #include +#include #include #include #include @@ -256,10 +257,43 @@ void TranscribeStreamingServiceClient::StartCallAnalyticsStreamTranscriptionAsyn } endpointResolutionOutcome.GetResult().AddPathSegments("/call-analytics-stream-transcription"); +#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); + + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + eventEncoderStream->SetSigner(signer); + + auto requestCopy = Aws::MakeShared(ALLOCATION_TAG, request); + request.SetAudioStream(eventEncoderStream); + + auto& endpoint = endpointResolutionOutcome.GetResult(); + auto httpRequest = + CreateHttpRequest(endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_POST, Aws::Utils::Stream::DefaultResponseStreamFactoryMethod); + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + BuildHttpRequest(*requestCopy, httpRequest); + + if (!signer->SignRequest(*httpRequest, nullptr, nullptr, true)) { + handler(this, request, + StartCallAnalyticsStreamTranscriptionOutcome( + Aws::Client::AWSError(CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + handlerContext); + return; + } + eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + Aws::Client::SubmitBidirectionalStreamingRequest( + this, request, requestCopy, eventEncoderStream, writeDataStreamBuf, httpRequest, m_clientConfiguration.executor.get(), + streamReadyHandler, handler, handlerContext); +#else + // Pull-based path (curl/WinHTTP) auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER)); auto requestCopy = Aws::MakeShared("StartCallAnalyticsStreamTranscription", request); - requestCopy->SetAudioStream(eventEncoderStream); // this becomes the body of the request + requestCopy->SetAudioStream(eventEncoderStream); request.SetAudioStream(eventEncoderStream); auto asyncTask = CreateBidirectionalEventStreamTask( @@ -268,6 +302,7 @@ void TranscribeStreamingServiceClient::StartCallAnalyticsStreamTranscriptionAsyn m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } void TranscribeStreamingServiceClient::StartMedicalScribeStreamAsync( Model::StartMedicalScribeStreamRequest& request, const StartMedicalScribeStreamStreamReadyHandler& streamReadyHandler, @@ -322,10 +357,43 @@ void TranscribeStreamingServiceClient::StartMedicalScribeStreamAsync( } endpointResolutionOutcome.GetResult().AddPathSegments("/medical-scribe-stream"); +#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); + + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + eventEncoderStream->SetSigner(signer); + + auto requestCopy = Aws::MakeShared(ALLOCATION_TAG, request); + request.SetInputStream(eventEncoderStream); + + auto& endpoint = endpointResolutionOutcome.GetResult(); + auto httpRequest = + CreateHttpRequest(endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_POST, Aws::Utils::Stream::DefaultResponseStreamFactoryMethod); + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + BuildHttpRequest(*requestCopy, httpRequest); + + if (!signer->SignRequest(*httpRequest, nullptr, nullptr, true)) { + handler(this, request, + StartMedicalScribeStreamOutcome( + Aws::Client::AWSError(CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + handlerContext); + return; + } + eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + Aws::Client::SubmitBidirectionalStreamingRequest( + this, request, requestCopy, eventEncoderStream, writeDataStreamBuf, httpRequest, m_clientConfiguration.executor.get(), + streamReadyHandler, handler, handlerContext); +#else + // Pull-based path (curl/WinHTTP) auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER)); auto requestCopy = Aws::MakeShared("StartMedicalScribeStream", request); - requestCopy->SetInputStream(eventEncoderStream); // this becomes the body of the request + requestCopy->SetInputStream(eventEncoderStream); request.SetInputStream(eventEncoderStream); auto asyncTask = CreateBidirectionalEventStreamTask( @@ -334,6 +402,7 @@ void TranscribeStreamingServiceClient::StartMedicalScribeStreamAsync( m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } void TranscribeStreamingServiceClient::StartMedicalStreamTranscriptionAsync( Model::StartMedicalStreamTranscriptionRequest& request, const StartMedicalStreamTranscriptionStreamReadyHandler& streamReadyHandler, @@ -404,10 +473,43 @@ void TranscribeStreamingServiceClient::StartMedicalStreamTranscriptionAsync( } endpointResolutionOutcome.GetResult().AddPathSegments("/medical-stream-transcription"); +#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); + + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + eventEncoderStream->SetSigner(signer); + + auto requestCopy = Aws::MakeShared(ALLOCATION_TAG, request); + request.SetAudioStream(eventEncoderStream); + + auto& endpoint = endpointResolutionOutcome.GetResult(); + auto httpRequest = + CreateHttpRequest(endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_POST, Aws::Utils::Stream::DefaultResponseStreamFactoryMethod); + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + BuildHttpRequest(*requestCopy, httpRequest); + + if (!signer->SignRequest(*httpRequest, nullptr, nullptr, true)) { + handler(this, request, + StartMedicalStreamTranscriptionOutcome( + Aws::Client::AWSError(CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + handlerContext); + return; + } + eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + Aws::Client::SubmitBidirectionalStreamingRequest( + this, request, requestCopy, eventEncoderStream, writeDataStreamBuf, httpRequest, m_clientConfiguration.executor.get(), + streamReadyHandler, handler, handlerContext); +#else + // Pull-based path (curl/WinHTTP) auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER)); auto requestCopy = Aws::MakeShared("StartMedicalStreamTranscription", request); - requestCopy->SetAudioStream(eventEncoderStream); // this becomes the body of the request + requestCopy->SetAudioStream(eventEncoderStream); request.SetAudioStream(eventEncoderStream); auto asyncTask = CreateBidirectionalEventStreamTask( @@ -416,6 +518,7 @@ void TranscribeStreamingServiceClient::StartMedicalStreamTranscriptionAsync( m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } void TranscribeStreamingServiceClient::StartStreamTranscriptionAsync( Model::StartStreamTranscriptionRequest& request, const StartStreamTranscriptionStreamReadyHandler& streamReadyHandler, @@ -462,10 +565,43 @@ void TranscribeStreamingServiceClient::StartStreamTranscriptionAsync( } endpointResolutionOutcome.GetResult().AddPathSegments("/stream-transcription"); +#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); + + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + eventEncoderStream->SetSigner(signer); + + auto requestCopy = Aws::MakeShared(ALLOCATION_TAG, request); + request.SetAudioStream(eventEncoderStream); + + auto& endpoint = endpointResolutionOutcome.GetResult(); + auto httpRequest = + CreateHttpRequest(endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_POST, Aws::Utils::Stream::DefaultResponseStreamFactoryMethod); + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + BuildHttpRequest(*requestCopy, httpRequest); + + if (!signer->SignRequest(*httpRequest, nullptr, nullptr, true)) { + handler(this, request, + StartStreamTranscriptionOutcome( + Aws::Client::AWSError(CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + handlerContext); + return; + } + eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + Aws::Client::SubmitBidirectionalStreamingRequest( + this, request, requestCopy, eventEncoderStream, writeDataStreamBuf, httpRequest, m_clientConfiguration.executor.get(), + streamReadyHandler, handler, handlerContext); +#else + // Pull-based path (curl/WinHTTP) auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER)); auto requestCopy = Aws::MakeShared("StartStreamTranscription", request); - requestCopy->SetAudioStream(eventEncoderStream); // this becomes the body of the request + requestCopy->SetAudioStream(eventEncoderStream); request.SetAudioStream(eventEncoderStream); auto asyncTask = CreateBidirectionalEventStreamTask( @@ -474,4 +610,5 @@ void TranscribeStreamingServiceClient::StartStreamTranscriptionAsync( m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } diff --git a/src/aws-cpp-sdk-core/CMakeLists.txt b/src/aws-cpp-sdk-core/CMakeLists.txt index 08ffeb00a8a..14ae520433e 100644 --- a/src/aws-cpp-sdk-core/CMakeLists.txt +++ b/src/aws-cpp-sdk-core/CMakeLists.txt @@ -64,6 +64,7 @@ file(GLOB UTILS_CHECKSUM_HEADERS "include/aws/core/utils/checksum/*.h") file(GLOB UTILS_CRYPTO_HEADERS "include/aws/core/utils/crypto/*.h") file(GLOB UTILS_JSON_HEADERS "include/aws/core/utils/json/*.h") file(GLOB UTILS_LOCAL_HEADERS "include/aws/core/utils/local/*.h") +file(GLOB UTILS_LOCAL_STREAM_HEADERS "include/aws/core/utils/local/stream/*.h") file(GLOB UTILS_CBOR_HEADERS "include/aws/core/utils/cbor/*.h") file(GLOB UTILS_THREADING_HEADERS "include/aws/core/utils/threading/*.h") file(GLOB UTILS_XML_HEADERS "include/aws/core/utils/xml/*.h") @@ -121,6 +122,7 @@ file(GLOB UTILS_CHECKSUM_SOURCE "${CMAKE_CURRENT_SOURCE_DIR}/source/utils/checks file(GLOB UTILS_CRYPTO_SOURCE "${CMAKE_CURRENT_SOURCE_DIR}/source/utils/crypto/*.cpp") file(GLOB UTILS_JSON_SOURCE "${CMAKE_CURRENT_SOURCE_DIR}/source/utils/json/*.cpp") file(GLOB UTILS_LOCAL_SOURCE "${CMAKE_CURRENT_SOURCE_DIR}/source/utils/local/*.cpp") +file(GLOB UTILS_LOCAL_STREAM_SOURCE "${CMAKE_CURRENT_SOURCE_DIR}/source/utils/local/stream/*.cpp") file(GLOB UTILS_CBOR_SOURCE "${CMAKE_CURRENT_SOURCE_DIR}/source/utils/cbor/*.cpp") file(GLOB UTILS_THREADING_SOURCE "${CMAKE_CURRENT_SOURCE_DIR}/source/utils/threading/*.cpp") file(GLOB UTILS_XML_SOURCE "${CMAKE_CURRENT_SOURCE_DIR}/source/utils/xml/*.cpp") @@ -277,6 +279,7 @@ file(GLOB AWS_NATIVE_SDK_COMMON_HEADERS ${UTILS_CRYPTO_HEADERS} ${UTILS_JSON_HEADERS} ${UTILS_LOCAL_HEADERS} + ${UTILS_LOCAL_STREAM_HEADERS} ${UTILS_CBOR_HEADERS} ${UTILS_THREADING_HEADERS} ${UTILS_RETRY_HEADERS} @@ -364,6 +367,7 @@ file(GLOB AWS_NATIVE_SDK_NON_UNITY_SRC ${UTILS_CRYPTO_FACTORY_SOURCE} ${UTILS_JSON_SOURCE} ${UTILS_LOCAL_SOURCE} + ${UTILS_LOCAL_STREAM_SOURCE} ${UTILS_CBOR_SOURCE} ${UTILS_EVENT_SOURCE} ${UTILS_SOURCE} @@ -426,6 +430,7 @@ if(MSVC) source_group("Header Files\\aws\\core\\utils\\exceptions" FILES ${UTILS_EXCEPTIONS_HEADERS}) source_group("Header Files\\aws\\core\\utils\\json" FILES ${UTILS_JSON_HEADERS}) source_group("Header Files\\aws\\core\\utils\\local" FILES ${UTILS_LOCAL_HEADERS}) + source_group("Header Files\\aws\\core\\utils\\local" FILES ${UTILS_LOCAL_STREAM_HEADERS}) source_group("Header Files\\aws\\core\\utils\\cbor" FILES ${UTILS_CBOR_HEADERS}) source_group("Header Files\\aws\\core\\utils\\threading" FILES ${UTILS_THREADING_HEADERS}) source_group("Header Files\\aws\\core\\utils\\xml" FILES ${UTILS_XML_HEADERS}) @@ -497,6 +502,7 @@ if(MSVC) source_group("Source Files\\utils\\exceptions" FILES ${UTILS_EXCEPTIONS_SOURCE}) source_group("Source Files\\utils\\json" FILES ${UTILS_JSON_SOURCE}) source_group("Source Files\\utils\\local" FILES ${UTILS_LOCAL_SOURCE}) + source_group("Source Files\\utils\\local" FILES ${UTILS_LOCAL_STREAM_SOURCE}) source_group("Source Files\\utils\\cbor" FILES ${UTILS_CBOR_SOURCE}) source_group("Source Files\\utils\\threading" FILES ${UTILS_THREADING_SOURCE}) source_group("Source Files\\utils\\xml" FILES ${UTILS_XML_SOURCE}) diff --git a/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientAsyncCRTP.h b/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientAsyncCRTP.h index 3ec8969caa3..5321ee272ec 100644 --- a/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientAsyncCRTP.h +++ b/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientAsyncCRTP.h @@ -11,6 +11,11 @@ namespace Aws { +namespace Http { class HttpRequest; } +namespace Utils { + namespace Stream { class HttpWriteDataStreamBuf; } + namespace Threading { class Executor; } +} namespace Client { class AsyncCallerContext; @@ -209,6 +214,14 @@ namespace Client template friend class BidirectionalEventStreamingTask; // allow BidirectionalEventStreamingTask to access m_isInitialized + template + friend void SubmitBidirectionalStreamingRequest( + const ClientT*, RequestT&, std::shared_ptr, std::shared_ptr, + std::shared_ptr, const std::shared_ptr&, + Aws::Utils::Threading::Executor*, + const StreamReadyHandlerT&, const HandlerT&, const std::shared_ptr&); + std::atomic m_isInitialized; mutable std::atomic m_operationsProcessed; mutable std::condition_variable m_shutdownSignal; diff --git a/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientBidirectionalStreaming.h b/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientBidirectionalStreaming.h new file mode 100644 index 00000000000..b6096ff4f6d --- /dev/null +++ b/src/aws-cpp-sdk-core/include/aws/core/client/AWSClientBidirectionalStreaming.h @@ -0,0 +1,137 @@ +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. + */ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace Aws { +namespace Client { + +/** + * Initializes an HTTP/2 connection and submits an executor task that waits for + * bidirectional stream completion using the push-based WriteData API. + * + * The caller is responsible for: + * - Validating request parameters and resolving the endpoint + * - Creating the encoder stream (with HttpWriteDataStreamBuf) and setting it on the request + * - Building and signing the HTTP request + * - Seeding event signing on the encoder stream + * + * This function handles: + * - Wiring the initial response handler and response stream factory + * - Initializing the HTTP/2 connection via HttpWriteDataStreamBuf + * - Submitting the executor task with shutdown safety (m_isInitialized, RAIICounter) + * - Calling streamReadyHandler once the connection is live + */ +template < + typename ClientT, + typename OutcomeT, + typename RequestT, + typename EncoderStreamT, + typename HandlerT, + typename StreamReadyHandlerT> +void SubmitBidirectionalStreamingRequest( + const ClientT* client, + RequestT& request, + std::shared_ptr requestCopy, + std::shared_ptr eventEncoderStream, + std::shared_ptr writeDataStreamBuf, + const std::shared_ptr& httpRequest, + Aws::Utils::Threading::Executor* executor, + const StreamReadyHandlerT& streamReadyHandler, + const HandlerT& handler, + const std::shared_ptr& handlerContext) +{ + const char* allocationTag = ClientT::GetAllocationTag(); + + // Validate host name + if (!Aws::Utils::IsValidHost(httpRequest->GetUri().GetHost())) { + handler(client, *requestCopy, + OutcomeT(Aws::Client::AWSError(Aws::Client::CoreErrors::VALIDATION, "", + "Invalid DNS Label found in URI host", false)), + handlerContext); + return; + } + + // Fix decoder pointer after copy construction + std::weak_ptr wReq = requestCopy; + requestCopy->SetEventStreamHandler(requestCopy->GetEventStreamHandler()); + + // Wire initial response handler on httpRequest (CRT reads it from there) + httpRequest->SetHeadersReceivedEventHandler( + [wReq](const Aws::Http::HttpRequest*, Aws::Http::HttpResponse* response) { + auto req = wReq.lock(); + if (!req || !response) return; + auto& cb = req->GetEventStreamHandler().GetInitialResponseCallbackEx(); + if (cb) { + cb({response->GetHeaders()}, Aws::Utils::Event::InitialResponseType::ON_RESPONSE); + } + }); + + // Wire response stream factory (weak_ptr breaks reference cycle) + auto responseStreamFactory = [wReq, allocationTag]() -> Aws::IOStream* { + auto req = wReq.lock(); + if (!req) return nullptr; + req->GetEventStreamDecoder().Reset(); + return Aws::New(allocationTag, req->GetEventStreamDecoder()); + }; + requestCopy->SetResponseStreamFactory(responseStreamFactory); + httpRequest->SetResponseStreamFactory(responseStreamFactory); + + // Initialize the HTTP/2 connection + auto initError = writeDataStreamBuf->Initialize(httpRequest); + if (initError.has_value()) { + handler(client, request, OutcomeT(initError.value()), handlerContext); + return; + } + + // Submit executor task — waits for stream completion, then invokes handler + executor->Submit( + [client, requestCopy, eventEncoderStream, handler, handlerContext, writeDataStreamBuf, allocationTag]() { + if (!client->m_isInitialized) { + AWS_LOGSTREAM_ERROR(allocationTag, "Client is not initialized or already terminated"); + handler(client, *requestCopy, + OutcomeT(Aws::Client::AWSError( + CoreErrors::NOT_INITIALIZED, "NOT_INITIALIZED", + "Client is not initialized or already terminated", false)), + handlerContext); + return; + } + Aws::Utils::RAIICounter raiiGuard(client->m_operationsProcessed, &client->m_shutdownSignal); + + writeDataStreamBuf->WaitForStreamComplete(); + auto response = writeDataStreamBuf->GetResponse(); + + // Flush any remaining buffered response data through the EventDecoderStream + if (response) { + response->GetResponseBody().flush(); + } + + if (response && response->HasClientError()) { + eventEncoderStream->Close(); + handler(client, *requestCopy, + OutcomeT(Aws::Client::AWSError( + response->GetClientErrorType(), "", response->GetClientErrorMessage(), false)), + handlerContext); + } else { + handler(client, *requestCopy, OutcomeT(Aws::NoResult()), handlerContext); + } + }); + + streamReadyHandler(*eventEncoderStream); +} + +} // namespace Client +} // namespace Aws diff --git a/src/aws-cpp-sdk-core/include/aws/core/utils/event/EventEncoderStream.h b/src/aws-cpp-sdk-core/include/aws/core/utils/event/EventEncoderStream.h index b457ab4ae84..45de2a1322e 100644 --- a/src/aws-cpp-sdk-core/include/aws/core/utils/event/EventEncoderStream.h +++ b/src/aws-cpp-sdk-core/include/aws/core/utils/event/EventEncoderStream.h @@ -20,6 +20,19 @@ namespace Aws namespace Utils { + namespace Stream + { + class HttpWriteDataStreamBuf; + + /** + * Adapter that virtualizes streambuf lifecycle operations (close, drain) + * so EventEncoderStream can work with different underlying streambufs + * (e.g. ConcurrentStreamBuf for the pull model, HttpWriteDataStreamBuf + * for the push model). + */ + class CloseableStreamBuf; + } + namespace Event { extern AWS_CORE_API const size_t DEFAULT_BUF_SIZE; @@ -37,6 +50,12 @@ namespace Aws */ explicit EventEncoderStream(size_t bufferSize = DEFAULT_BUF_SIZE); + /** + * Creates a stream for encoding events sent by the client. + * @param streambuf the underlying buffer used by event encoder. + */ + explicit EventEncoderStream(std::shared_ptr streambuf); + /** * Sets the signature seed used by event-stream events. * Every event uses its previous event's signature to calculate its own signature. @@ -65,16 +84,15 @@ namespace Aws * Any writes to the stream after this call are not guaranteed to be read by another concurrent * read thread. */ - void Close() { m_streambuf.SetEofInput(this); } + void Close(); /** * Blocks the current thread until all submitted data is consumed. * Returns false on timeout, and true if GetArea and back buffer are empty. */ bool WaitForDrain(int64_t timeoutMs = 1000); - private: - Stream::ConcurrentStreamBuf m_streambuf; + std::shared_ptr m_streambuf; EventStreamEncoder m_encoder; }; } diff --git a/src/aws-cpp-sdk-core/include/aws/core/utils/local/stream/CloseableStreamBuf.h b/src/aws-cpp-sdk-core/include/aws/core/utils/local/stream/CloseableStreamBuf.h new file mode 100644 index 00000000000..abd923ae1fe --- /dev/null +++ b/src/aws-cpp-sdk-core/include/aws/core/utils/local/stream/CloseableStreamBuf.h @@ -0,0 +1,42 @@ +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. + */ +#pragma once + +#include +#include +#include + +namespace Aws { +namespace Utils { +namespace Stream { +class HttpWriteDataStreamBuf; + +class CloseableStreamBuf { + public: + explicit CloseableStreamBuf(std::shared_ptr streambuf) : m_streambuf(std::move(streambuf)) {} + std::streambuf* GetStreamBufHandler() { return m_streambuf.get(); } + virtual ~CloseableStreamBuf() = default; + virtual bool WaitForDrain(int64_t timeout); + virtual void Close(Aws::IOStream* pStreamToClose = nullptr); + + private: + std::shared_ptr m_streambuf; +}; + +class CloseableConcurrentStreamBuf : public Aws::Utils::Stream::CloseableStreamBuf { + public: + explicit CloseableConcurrentStreamBuf(size_t bufferSize); + bool WaitForDrain(int64_t timeoutMs) override; + void Close(Aws::IOStream* stream) override; +}; + +class CloseableHttpWriteDataStreamBuf : public Aws::Utils::Stream::CloseableStreamBuf { + public: + explicit CloseableHttpWriteDataStreamBuf(std::shared_ptr streambuf); + void Close(Aws::IOStream* stream) override; +}; +} // namespace Stream +} // namespace Utils +} // namespace Aws \ No newline at end of file diff --git a/src/aws-cpp-sdk-core/include/aws/core/utils/stream/HttpWriteDataStreamBuf.h b/src/aws-cpp-sdk-core/include/aws/core/utils/stream/HttpWriteDataStreamBuf.h new file mode 100644 index 00000000000..d76c463db6d --- /dev/null +++ b/src/aws-cpp-sdk-core/include/aws/core/utils/stream/HttpWriteDataStreamBuf.h @@ -0,0 +1,118 @@ +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. + */ +#pragma once + +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace Aws { +namespace Http { +class HttpClient; +class HttpRequest; +class HttpResponse; +class Connection; +class ClientStream; +} // namespace Http +namespace Utils { +namespace Stream { + +/** + * Write-only streambuf that pushes data to an HTTP/2 connection via ClientStream::WriteData. + * + * Lifecycle: construct → Initialize() → write via ostream → Close() → WaitForStreamComplete(). + * Close() sends endStream=true. Destructor calls Close() + WaitForStreamComplete(). + * + * Not copyable or movable. CRT HTTP client only. + */ +class AWS_CORE_API HttpWriteDataStreamBuf : public std::streambuf { + public: + explicit HttpWriteDataStreamBuf(const std::shared_ptr& client, size_t bufferLength = 8 * 1024); + HttpWriteDataStreamBuf(const HttpWriteDataStreamBuf& other) = delete; + HttpWriteDataStreamBuf(HttpWriteDataStreamBuf&& other) noexcept = delete; + HttpWriteDataStreamBuf& operator=(const HttpWriteDataStreamBuf& other) = delete; + HttpWriteDataStreamBuf& operator=(HttpWriteDataStreamBuf&& other) noexcept = delete; + ~HttpWriteDataStreamBuf() override; + + /** + * Acquires a connection and creates a client stream. Must be called before any writes. + * Returns an error if connection acquisition or stream creation fails. + */ + Aws::Crt::Optional> Initialize(const std::shared_ptr& request); + + /** + * Flushes remaining buffer with endStream=true. Idempotent. + * Does not block on stream completion — call WaitForStreamComplete() for that. + */ + void Close(); + + /** + * Returns the HTTP response from the underlying stream. Must be called after Initialize(). + */ + std::shared_ptr GetResponse() const; + + /** + * Blocks until the CRT fires the stream-complete callback. + * No-op if Initialize() was never called. Close() must be called first. + */ + void WaitForStreamComplete(); + + protected: + // Write support + int_type overflow(int_type c) override; + std::streamsize xsputn(const char* s, std::streamsize n) override; + int sync() override; + + // Disable reads + int_type underflow() override { return traits_type::eof(); } + int_type uflow() override { return traits_type::eof(); } + std::streamsize xsgetn(char*, std::streamsize) override { return 0; } + int_type pbackfail(int_type) override { return traits_type::eof(); } + + // Disable seeking + pos_type seekoff(off_type, std::ios_base::seekdir, std::ios_base::openmode) override { return pos_type(off_type(-1)); } + pos_type seekpos(pos_type, std::ios_base::openmode) override { return pos_type(off_type(-1)); } + + private: + /** + * Sends buffer contents via WriteData, blocks until completion callback fires. + * Returns false on error. + */ + bool SendBuffer(bool endStream = false); + void ResetPutArea(); + + // Client state + std::shared_ptr m_client; + std::shared_ptr m_connection; + std::shared_ptr m_stream; + + // Data buffer + Aws::Utils::Array m_buffer; + + // WriteData synchronization + std::mutex m_writeMutex; + std::condition_variable m_writeComplete; + bool m_writeInProgress{false}; + bool m_writeError{false}; + + // State management + enum class STATE { + UNINITIALIZED, + INITIALIZED, + SHUT_DOWN, + } m_state{STATE::UNINITIALIZED}; + std::condition_variable m_shutdownCondition; + std::mutex m_shutdownMutex; + bool m_streamComplete{false}; +}; +} // namespace Stream +} // namespace Utils +} // namespace Aws diff --git a/src/aws-cpp-sdk-core/include/smithy/client/AwsSmithyClient.h b/src/aws-cpp-sdk-core/include/smithy/client/AwsSmithyClient.h index 106b609277d..b5760248c53 100644 --- a/src/aws-cpp-sdk-core/include/smithy/client/AwsSmithyClient.h +++ b/src/aws-cpp-sdk-core/include/smithy/client/AwsSmithyClient.h @@ -142,6 +142,9 @@ namespace client template friend class SmithyBidirectionalStreamingTask; + template + friend class SmithyBidirectionalStreamingWriteDataTask; + void initClient() { if (m_endpointProvider && m_authSchemeResolver) { m_endpointProvider->InitBuiltInParameters(m_clientConfiguration); diff --git a/src/aws-cpp-sdk-core/include/smithy/client/AwsSmithyClientBase.h b/src/aws-cpp-sdk-core/include/smithy/client/AwsSmithyClientBase.h index 09087f225cd..1be3a25332c 100644 --- a/src/aws-cpp-sdk-core/include/smithy/client/AwsSmithyClientBase.h +++ b/src/aws-cpp-sdk-core/include/smithy/client/AwsSmithyClientBase.h @@ -167,6 +167,9 @@ namespace client protected: template friend class SmithyBidirectionalEventStreamingTask; + + template + friend class SmithyBidirectionalStreamingWriteDataTask; //for backwards compatibility const std::shared_ptr& GetErrorMarshaller() const diff --git a/src/aws-cpp-sdk-core/include/smithy/client/SmithyBidirectionalStreamingWriteDataTask.h b/src/aws-cpp-sdk-core/include/smithy/client/SmithyBidirectionalStreamingWriteDataTask.h new file mode 100644 index 00000000000..fb9f3aba0a1 --- /dev/null +++ b/src/aws-cpp-sdk-core/include/smithy/client/SmithyBidirectionalStreamingWriteDataTask.h @@ -0,0 +1,205 @@ +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. + */ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace smithy { +namespace client { + +/** + * Push-based bidirectional streaming task for Smithy clients using the WriteData API. + * Replaces SmithyBidirectionalStreamingTask when CRT HTTP client is available. + */ +template +class AWS_CORE_LOCAL SmithyBidirectionalStreamingWriteDataTask final { + public: + using AuthResolvedCallback = std::function)>; + using EndpointUpdateCallback = std::function; + + SmithyBidirectionalStreamingWriteDataTask( + const ClientT* client, + const std::shared_ptr& request, + const HandlerT& handler, + const std::shared_ptr& context, + const std::shared_ptr& stream, + std::shared_ptr writeDataStreamBuf, + EndpointUpdateCallback&& endpointCallback, + AuthResolvedCallback&& authCallback) + : m_client(client), + m_request(request), + m_handler(handler), + m_context(context), + m_stream(stream), + m_writeDataStreamBuf(std::move(writeDataStreamBuf)), + m_authCallback(std::move(authCallback)), + m_endpointCallback(std::move(endpointCallback)) { + + // Fix decoder pointer after copy construction + m_request->SetEventStreamHandler(m_request->GetEventStreamHandler()); + + std::weak_ptr wRequest = m_request; + + // Wire response stream factory + m_request->SetResponseStreamFactory([wRequest]() -> Aws::IOStream* { + auto req = wRequest.lock(); + if (!req) return nullptr; + req->GetEventStreamDecoder().Reset(); + return Aws::New("SmithyBidirectionalStreamingWriteDataTask", + req->GetEventStreamDecoder()); + }); + } + + void operator()() { + // Resolve auth and endpoint + auto pExecutor = Aws::MakeShared("SmithyBidiWriteData"); + auto authResolver = m_request->GetRequestSpecificSupportedAuth().empty() ? nullptr + : Aws::MakeShared>("SmithyBidiWriteData", + m_request->GetRequestSpecificSupportedAuth()); + auto ctx = Aws::MakeShared("SmithyBidiWriteData", + m_request.get(), m_request->GetServiceRequestName(), pExecutor, authResolver); + ctx->m_method = Aws::Http::HttpMethod::HTTP_POST; + + bool resolvedIdenity = m_client->ResolveIdentityAuth( + ctx, + [](AwsSmithyClientBase::HttpResponseOutcome&& outcome) { + if (!outcome.IsSuccess()) { + AWS_LOGSTREAM_ERROR("SmithyBidiWriteData", "Failed to resolve identity/auth"); + } + }, + EndpointUpdateCallback(m_endpointCallback)); + + if (!resolvedIdenity) { + m_handler(m_client, *m_request, + OutcomeT(Aws::Client::AWSError( + Aws::Client::CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to resolve identity", false)), + m_context); + return; + } + + // Wire event signing + if (m_authCallback) { + m_authCallback(ctx); + } + + // Validate host name + if (!Aws::Utils::IsValidHost(ctx->m_endpoint.GetURI().GetHost())) { + m_handler(m_client, *m_request, + OutcomeT(Aws::Client::AWSError(Aws::Client::CoreErrors::VALIDATION, "", + "Invalid DNS Label found in URI host", false)), + m_context); + return; + } + + // Build and sign HTTP request + m_client->UpdateAuthSchemeFromEndpoint(ctx->m_endpoint, ctx->m_authSchemeOption); + auto httpRequest = m_client->BuildHttpRequest(ctx, ctx->m_endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_POST); + if (!httpRequest) { + m_handler(m_client, *m_request, + OutcomeT(Aws::Client::AWSError( + Aws::Client::CoreErrors::VALIDATION, "", "Unable to create HttpRequest", false)), + m_context); + return; + } + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + httpRequest->SetResponseStreamFactory(m_request->GetResponseStreamFactory()); + + // Wire initial response handler on httpRequest (CRT reads it from here) + std::weak_ptr wReq = m_request; + httpRequest->SetHeadersReceivedEventHandler( + [wReq](const Aws::Http::HttpRequest*, Aws::Http::HttpResponse* response) { + auto req = wReq.lock(); + if (!req || !response) return; + auto& cb = req->GetEventStreamHandler().GetInitialResponseCallbackEx(); + if (cb) { + cb({response->GetHeaders()}, Aws::Utils::Event::InitialResponseType::ON_RESPONSE); + } + }); + + auto signingOutcome = m_client->SignHttpRequest(httpRequest, *ctx); + if (!signingOutcome.IsSuccess()) { + m_handler(m_client, *m_request, + OutcomeT(Aws::Client::AWSError( + Aws::Client::CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + m_context); + return; + } + httpRequest = signingOutcome.GetResultWithOwnership(); + + // Seed event signing + m_stream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + // Initialize connection + auto initError = m_writeDataStreamBuf->Initialize(httpRequest); + if (initError.has_value()) { + m_handler(m_client, *m_request, OutcomeT(initError.value()), m_context); + return; + } + + // Signal that stream is ready (caller is waiting on semaphore) + m_sem->ReleaseAll(); + + // Wait for stream to complete + m_writeDataStreamBuf->WaitForStreamComplete(); + auto response = m_writeDataStreamBuf->GetResponse(); + + // Flush any remaining buffered response data through the EventDecoderStream + if (response) { + response->GetResponseBody().flush(); + } + if (response && response->HasClientError()) { + m_stream->Close(); + m_handler(m_client, *m_request, + OutcomeT(Aws::Client::AWSError( + response->GetClientErrorType(), "", response->GetClientErrorMessage(), false)), + m_context); + } else { + m_handler(m_client, *m_request, OutcomeT(Aws::NoResult()), m_context); + } + } + + const std::shared_ptr& GetSemaphore() const { return m_sem; } + + private: + const ClientT* m_client; + std::shared_ptr m_request; + HandlerT m_handler; + std::shared_ptr m_context; + std::shared_ptr m_stream; + std::shared_ptr m_writeDataStreamBuf; + AuthResolvedCallback m_authCallback; + EndpointUpdateCallback m_endpointCallback; + std::shared_ptr m_sem{ + Aws::MakeShared("SmithyBidiWriteData", 0, 1)}; +}; + +template +static SmithyBidirectionalStreamingWriteDataTask +CreateSmithyBidirectionalWriteDataTask( + const ClientT* client, + std::shared_ptr request, + const HandlerT& handler, + const std::shared_ptr& context, + const std::shared_ptr& stream, + std::shared_ptr writeDataStreamBuf, + std::function&& endpointCallback, + std::function)>&& authCallback) { + return SmithyBidirectionalStreamingWriteDataTask( + client, request, handler, context, stream, std::move(writeDataStreamBuf), + std::move(endpointCallback), std::move(authCallback)); +} + +} // namespace client +} // namespace smithy diff --git a/src/aws-cpp-sdk-core/source/http/crt/CRTHttpClient.cpp b/src/aws-cpp-sdk-core/source/http/crt/CRTHttpClient.cpp index 710b7fb731b..71de62cdfc0 100644 --- a/src/aws-cpp-sdk-core/source/http/crt/CRTHttpClient.cpp +++ b/src/aws-cpp-sdk-core/source/http/crt/CRTHttpClient.cpp @@ -195,6 +195,10 @@ class CRTClientStream : public Aws::Http::ClientStream { int WriteData(std::shared_ptr stream, const std::function& onComplete, bool endStream) override { auto crtStream = std::make_shared(stream, Aws::Crt::ApiAllocator()); + if (!crtStream->IsValid()) { + onComplete(AWS_ERROR_HTTP_STREAM_HAS_COMPLETED); + return AWS_ERROR_HTTP_STREAM_HAS_COMPLETED; + } return m_stream->WriteData(crtStream, [onComplete](std::shared_ptr&, int errorCode) { onComplete(errorCode); diff --git a/src/aws-cpp-sdk-core/source/utils/event/EventEncoderStream.cpp b/src/aws-cpp-sdk-core/source/utils/event/EventEncoderStream.cpp index c413a09ede7..c3ad2b820ba 100644 --- a/src/aws-cpp-sdk-core/source/utils/event/EventEncoderStream.cpp +++ b/src/aws-cpp-sdk-core/source/utils/event/EventEncoderStream.cpp @@ -2,10 +2,13 @@ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. * SPDX-License-Identifier: Apache-2.0. */ +#include #include -#include +#include -#include +namespace { +const char* LOG_TAG = "EventEncoderStream"; +} // namespace namespace Aws { @@ -14,14 +17,22 @@ namespace Aws namespace Event { EventEncoderStream::EventEncoderStream(size_t bufferSize) : - Aws::IOStream(&m_streambuf), - m_streambuf(bufferSize) + Aws::IOStream(nullptr), + m_streambuf{Aws::MakeUnique(LOG_TAG, bufferSize)} { + rdbuf(m_streambuf->GetStreamBufHandler()); } + EventEncoderStream::EventEncoderStream(std::shared_ptr streambuf) + : Aws::IOStream(nullptr), m_streambuf{Aws::MakeShared(LOG_TAG, std::move(streambuf))} { + rdbuf(m_streambuf->GetStreamBufHandler()); + } + + void EventEncoderStream::Close() { m_streambuf->Close(this); } + bool EventEncoderStream::WaitForDrain(int64_t timeoutMs) { - return m_streambuf.WaitForDrain(timeoutMs); + return m_streambuf->WaitForDrain(timeoutMs); } EventEncoderStream& EventEncoderStream::WriteEvent(const Aws::Utils::Event::Message& msg) @@ -31,11 +42,9 @@ namespace Aws AWS_LOGSTREAM_TRACE("EventEncoderStream::WriteEvent", "Encoded event (base64 encoded): " << Aws::Utils::HashingUtils::Base64Encode(Aws::Utils::ByteBuffer(bits.data(), bits.size()))); - // write buffer to underlying rdbuf (ConcurrentStreamBuf), this may call overflow() - // and block until data is consumed by HTTP Client + // write encoded event to the underlying streambuf, may block + // if the streambuf applies backpressure (e.g. HttpWriteDataStreamBuf) write(reinterpret_cast(bits.data()), bits.size()); - // force flushing ConcurrentStreamBuf to move data from PutArea to the back buffer - // so that consuming HTTP Client will have data to send flush(); return *this; } diff --git a/src/aws-cpp-sdk-core/source/utils/local/stream/CloseableStreamBuf.cpp b/src/aws-cpp-sdk-core/source/utils/local/stream/CloseableStreamBuf.cpp new file mode 100644 index 00000000000..f49db6fed14 --- /dev/null +++ b/src/aws-cpp-sdk-core/source/utils/local/stream/CloseableStreamBuf.cpp @@ -0,0 +1,39 @@ +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. + */ +#include +#include +#include +#include + +namespace { +const char* LOG_TAG = "CloseableStreamBuf"; +} + +bool Aws::Utils::Stream::CloseableStreamBuf::WaitForDrain(int64_t timeout) { + AWS_UNREFERENCED_PARAM(timeout); + return true; +} + +void Aws::Utils::Stream::CloseableStreamBuf::Close(Aws::IOStream* pStreamToClose) { AWS_UNREFERENCED_PARAM(pStreamToClose); } + +Aws::Utils::Stream::CloseableConcurrentStreamBuf::CloseableConcurrentStreamBuf(size_t bufferSize) + : CloseableStreamBuf(Aws::MakeUnique(LOG_TAG, bufferSize)) {} + +bool Aws::Utils::Stream::CloseableConcurrentStreamBuf::WaitForDrain(int64_t timeoutMs) { + return static_cast(GetStreamBufHandler())->WaitForDrain(timeoutMs); +} + +void Aws::Utils::Stream::CloseableConcurrentStreamBuf::Close(Aws::IOStream* stream) { + static_cast(GetStreamBufHandler())->SetEofInput(stream); +} + +Aws::Utils::Stream::CloseableHttpWriteDataStreamBuf::CloseableHttpWriteDataStreamBuf( + std::shared_ptr streambuf) + : CloseableStreamBuf(std::move(streambuf)) {} + +void Aws::Utils::Stream::CloseableHttpWriteDataStreamBuf::Close(Aws::IOStream* stream) { + AWS_UNREFERENCED_PARAM(stream); + static_cast(GetStreamBufHandler())->Close(); +} diff --git a/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp b/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp new file mode 100644 index 00000000000..c04849cc2ee --- /dev/null +++ b/src/aws-cpp-sdk-core/source/utils/stream/HttpWriteDataStreamBuf.cpp @@ -0,0 +1,171 @@ +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. + */ +#include +#include + +#include + +namespace { +const char* WRITE_DATA_BUF_LOG_NAME = "HttpWriteDataStreamBuf"; +} + +Aws::Utils::Stream::HttpWriteDataStreamBuf::HttpWriteDataStreamBuf(const std::shared_ptr& client, + size_t bufferLength) + : m_client{client}, m_buffer{bufferLength} { + ResetPutArea(); +} + +void Aws::Utils::Stream::HttpWriteDataStreamBuf::ResetPutArea() { + auto* base = reinterpret_cast(m_buffer.GetUnderlyingData()); + setp(base, base + m_buffer.GetLength()); +} + +Aws::Utils::Stream::HttpWriteDataStreamBuf::~HttpWriteDataStreamBuf() { + Close(); + WaitForStreamComplete(); +} + +Aws::Crt::Optional> Aws::Utils::Stream::HttpWriteDataStreamBuf::Initialize( + const std::shared_ptr& request) { + std::mutex acquireMutex; + std::condition_variable acquireCondition; + bool connectionAcquired = false; + Aws::Crt::Optional> error; + + // Callback captures stack locals by reference. This is safe because + // we block on acquireCondition.wait() below, keeping the stack frame + // alive until the callback has fired. + m_client->AcquireConnection(request, + [&acquireMutex, &acquireCondition, &connectionAcquired, &error, this]( + std::shared_ptr connection, int errorCode) -> void { + { + std::unique_lock const lock{acquireMutex}; + if (errorCode != AWS_ERROR_SUCCESS) { + error = + Aws::Client::AWSError(Aws::Client::CoreErrors::NETWORK_CONNECTION, false); + } + m_connection = std::move(connection); + connectionAcquired = true; + } + acquireCondition.notify_all(); + }); + + std::unique_lock lock{acquireMutex}; + acquireCondition.wait(lock, [&connectionAcquired]() -> bool { return connectionAcquired; }); + + if (error.has_value()) { + return error; + } + + m_stream = m_connection->NewClientStream(request, [this](int errorCode) -> void { + { + std::unique_lock const lock{m_shutdownMutex}; + if (errorCode != AWS_ERROR_SUCCESS) { + AWS_LOGSTREAM_ERROR(WRITE_DATA_BUF_LOG_NAME, "Stream completed with error: " << errorCode); + m_writeError = true; + } + m_streamComplete = true; + } + m_shutdownCondition.notify_all(); + }); + + if (!m_stream) { + return Aws::Client::AWSError(Aws::Client::CoreErrors::NETWORK_CONNECTION, false); + } + + m_stream->Activate(); + m_state = STATE::INITIALIZED; + return error; +} + +void Aws::Utils::Stream::HttpWriteDataStreamBuf::Close() { + { + std::unique_lock const lock{m_shutdownMutex}; + if (m_state != STATE::INITIALIZED) { + return; + } + m_state = STATE::SHUT_DOWN; + } + + SendBuffer(/*endStream=*/true); +} + +std::shared_ptr Aws::Utils::Stream::HttpWriteDataStreamBuf::GetResponse() const { + assert(m_stream); + return m_stream->GetResponse(); +} + +void Aws::Utils::Stream::HttpWriteDataStreamBuf::WaitForStreamComplete() { + std::unique_lock lock{m_shutdownMutex}; + if (m_state == STATE::UNINITIALIZED) { + return; + } + m_shutdownCondition.wait(lock, [this]() -> bool { return m_streamComplete; }); +} + +std::streambuf::int_type Aws::Utils::Stream::HttpWriteDataStreamBuf::overflow(std::streambuf::int_type c) { + if (traits_type::eq_int_type(c, traits_type::eof())) { + return traits_type::not_eof(c); + } + + if (!SendBuffer(false)) { + return traits_type::eof(); + } + + *pptr() = traits_type::to_char_type(c); + pbump(1); + return c; +} + +std::streamsize Aws::Utils::Stream::HttpWriteDataStreamBuf::xsputn(const char* s, std::streamsize n) { + std::streamsize written = 0; + while (written < n) { + std::streamsize const space = epptr() - pptr(); + std::streamsize const chunk = std::min(space, n - written); + + std::memcpy(pptr(), s + written, static_cast(chunk)); + pbump(static_cast(chunk)); + written += chunk; + + if (pptr() == epptr()) { + if (!SendBuffer(false)) { + return written; + } + } + } + return written; +} + +int Aws::Utils::Stream::HttpWriteDataStreamBuf::sync() { + return SendBuffer(false) ? 0 : -1; +} + +bool Aws::Utils::Stream::HttpWriteDataStreamBuf::SendBuffer(bool endStream) { + if (!endStream && (m_writeError || m_state != STATE::INITIALIZED)) { + return false; + } + + auto data = Aws::MakeShared(WRITE_DATA_BUF_LOG_NAME); + data->write(pbase(), pptr() - pbase()); + + m_writeInProgress = true; + + m_stream->WriteData( + data, + [this](int errorCode) -> void { + std::unique_lock const lock{m_writeMutex}; + m_writeInProgress = false; + m_writeError = (errorCode != AWS_ERROR_SUCCESS); + m_writeComplete.notify_one(); + }, + endStream); + + std::unique_lock lock{m_writeMutex}; + m_writeComplete.wait(lock, [this]() -> bool { return !m_writeInProgress; }); + + ResetPutArea(); + + return !m_writeError; +} diff --git a/tests/aws-cpp-sdk-core-tests/utils/stream/HttpWriteDataStreamBufTest.cpp b/tests/aws-cpp-sdk-core-tests/utils/stream/HttpWriteDataStreamBufTest.cpp new file mode 100644 index 00000000000..ef6d054a0ae --- /dev/null +++ b/tests/aws-cpp-sdk-core-tests/utils/stream/HttpWriteDataStreamBufTest.cpp @@ -0,0 +1,182 @@ +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. + */ +#include +#include +#include + +namespace { +const char* TEST_ALLOCATION_LOG_TAG = "HttpWriteDataStreamBufTest"; +} + +class HttpWriteDataStreamBufTest : public Aws::Testing::AwsCppSdkGTestSuite { + protected: + void SetUp() override { + auto options = GetSdkOptions(); + options.httpOptions.httpClientFactory_create_fn = [this]() -> std::shared_ptr { + client_factory_->SetClient(client_); + return client_factory_; + }; + Aws::InitAPI(options); + client_factory_ = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + client_ = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + }; + + void TearDown() override { Aws::ShutdownAPI(GetSdkOptions()); } + + std::shared_ptr client_factory_; + std::shared_ptr client_; +}; + +TEST_F(HttpWriteDataStreamBufTest, TestShouldWriteData) { + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.response = response; + client_->SetConnectionTestCase(testCase); + { + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 10}; + data_stream_buf.Initialize(request); + Aws::IOStream stream(&data_stream_buf); + stream << "I backed my car into a cop car the other day "; + stream << "Well, he just drove off, sometimes life's okay "; + stream << "I ran my mouth off a bit too much, oh, what did I say? "; + stream << "Well, you just laughed it off, it was all okay "; + stream << "And we'll all float on, okay"; + } + + EXPECT_STREQ(output->str().c_str(), + "I backed my car into a cop car the other day Well, he just drove off, sometimes life's okay I ran my mouth off a bit too " + "much, oh, what did I say? Well, you just laughed it off, it was all okay And we'll all float on, okay"); +} + +TEST_F(HttpWriteDataStreamBufTest, TestConnectionAcquireFailure) { + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + ConnectionTestCase testCase{}; + testCase.connectionErrorCode = 1; + client_->SetConnectionTestCase(testCase); + + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 10}; + auto error = data_stream_buf.Initialize(request); + ASSERT_TRUE(error.has_value()); + EXPECT_EQ(error->GetErrorType(), Aws::Client::CoreErrors::NETWORK_CONNECTION); +} + +TEST_F(HttpWriteDataStreamBufTest, TestWriteDataCallbackError) { + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.writeDataCompleteErrorCode = 1; + testCase.response = response; + client_->SetConnectionTestCase(testCase); + + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 10}; + data_stream_buf.Initialize(request); + Aws::IOStream stream(&data_stream_buf); + stream << "Eating snowflakes with plastic forks "; + EXPECT_TRUE(stream.fail() || output->str().size() < 38); +} + +TEST_F(HttpWriteDataStreamBufTest, TestSingleCharOverflow) { + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.response = response; + client_->SetConnectionTestCase(testCase); + { + // Buffer of 1 forces overflow on every character + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 1}; + data_stream_buf.Initialize(request); + Aws::IOStream stream(&data_stream_buf); + stream << "Everything that keeps me together is falling apart"; + } + + EXPECT_STREQ(output->str().c_str(), "Everything that keeps me together is falling apart"); +} + +TEST_F(HttpWriteDataStreamBufTest, TestLargeBufferNoFlushUntilClose) { + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.response = response; + client_->SetConnectionTestCase(testCase); + { + // Buffer larger than the data, so it all flushes on Close + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 4096}; + data_stream_buf.Initialize(request); + Aws::IOStream stream(&data_stream_buf); + stream << "The universe is shaped exactly like the earth"; + EXPECT_STREQ(output->str().c_str(), ""); + } + + EXPECT_STREQ(output->str().c_str(), "The universe is shaped exactly like the earth"); +} + +TEST_F(HttpWriteDataStreamBufTest, TestEmptyWrite) { + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.response = response; + client_->SetConnectionTestCase(testCase); + { + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 10}; + data_stream_buf.Initialize(request); + Aws::IOStream stream(&data_stream_buf); + } + + EXPECT_STREQ(output->str().c_str(), ""); +} + +TEST_F(HttpWriteDataStreamBufTest, TestGetResponse) { + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + response->SetResponseCode(Aws::Http::HttpResponseCode::OK); + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.response = response; + client_->SetConnectionTestCase(testCase); + + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 10}; + data_stream_buf.Initialize(request); + auto resp = data_stream_buf.GetResponse(); + ASSERT_NE(resp, nullptr); + EXPECT_EQ(resp->GetResponseCode(), Aws::Http::HttpResponseCode::OK); +} + +TEST_F(HttpWriteDataStreamBufTest, TestExactBufferBoundary) { + auto output = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG); + auto request = Aws::Http::CreateHttpRequest(Aws::String{"http://www.amazon.com/"}, Aws::Http::HttpMethod::HTTP_POST, + []() -> Aws::IOStream* { return Aws::New(TEST_ALLOCATION_LOG_TAG); }); + auto response = Aws::MakeShared(TEST_ALLOCATION_LOG_TAG, request); + ConnectionTestCase testCase{}; + testCase.writeDataStream = output; + testCase.response = response; + client_->SetConnectionTestCase(testCase); + { + Aws::Utils::Stream::HttpWriteDataStreamBuf data_stream_buf{client_, 10}; + data_stream_buf.Initialize(request); + Aws::IOStream stream(&data_stream_buf); + stream << "well we are"; + } + + EXPECT_STREQ(output->str().c_str(), "well we are"); +} diff --git a/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h b/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h new file mode 100644 index 00000000000..b16fba0ffa5 --- /dev/null +++ b/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h @@ -0,0 +1,60 @@ +/** + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0. + */ + +#pragma once +#include +#include +#include +#include + +#include + +struct ConnectionTestCase { + Aws::Crt::Optional> connectionError{}; + int connectionErrorCode{0}; + int streamCompleteErrorCode{0}; + int writeDataErrorCode{0}; + int writeDataCompleteErrorCode{0}; + std::shared_ptr writeDataStream; + std::shared_ptr response; +}; + +class MockClientStream : public Aws::Http::ClientStream { + public: + MockClientStream(const ConnectionTestCase& testCase, std::function onStreamComplete) + : m_testCase{testCase}, m_onStreamComplete{std::move(onStreamComplete)} {} + ~MockClientStream() override = default; + + bool Activate() override { return true; } + + int WriteData(std::shared_ptr stream, const std::function& onComplete, bool endStream) override { + *m_testCase.writeDataStream << stream->rdbuf(); + onComplete(m_testCase.writeDataCompleteErrorCode); + if (endStream) { + m_onStreamComplete(m_testCase.streamCompleteErrorCode); + } + return m_testCase.writeDataErrorCode; + } + + std::shared_ptr GetResponse() const override { return m_testCase.response; } + + private: + ConnectionTestCase m_testCase; + std::function m_onStreamComplete; +}; + +class MockConnection : public Aws::Http::Connection { + public: + MockConnection(const ConnectionTestCase& testCase) : m_testCase{testCase} {} + ~MockConnection() override = default; + std::shared_ptr NewClientStream(const std::shared_ptr& request, + std::function onStreamComplete) override { + AWS_UNREFERENCED_PARAM(request); + return Aws::MakeShared("MockConnection", m_testCase, std::move(onStreamComplete)); + } + + private: + ConnectionTestCase m_testCase; +}; \ No newline at end of file diff --git a/tests/testing-resources/include/aws/testing/mocks/http/MockHttpClient.h b/tests/testing-resources/include/aws/testing/mocks/http/MockHttpClient.h index b8f4d7a3ca4..4192d1377b9 100644 --- a/tests/testing-resources/include/aws/testing/mocks/http/MockHttpClient.h +++ b/tests/testing-resources/include/aws/testing/mocks/http/MockHttpClient.h @@ -8,13 +8,14 @@ #include #include #include +#include #include #include #include -#include #include -#include #include +#include +#include static const char MockHttpAllocationTag[] = "MockHttp"; @@ -86,9 +87,19 @@ class MockHttpClient : public Aws::Http::HttpClient std::swap(m_responsesToUse, empty); } + void SetConnectionTestCase(const ConnectionTestCase& testCase) { m_connectionTestCase = testCase; } + Aws::Crt::Optional> AcquireConnection( + const std::shared_ptr& request, + const std::function, int)>& onClientConnectionAvailable) override { + AWS_UNREFERENCED_PARAM(request); + auto connection = Aws::MakeShared(MockHttpAllocationTag, m_connectionTestCase); + onClientConnectionAvailable(connection, m_connectionTestCase.connectionErrorCode); + return m_connectionTestCase.connectionError; + } -private: + private: + mutable ConnectionTestCase m_connectionTestCase; mutable Aws::Vector m_requestsMade; mutable Aws::Queue m_responsesToUse; mutable Aws::Queue m_responseAndRequestsCallback; diff --git a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/ServiceClientSourceHeaders.vm b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/ServiceClientSourceHeaders.vm index 4a371ad7690..b3662f6f492 100644 --- a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/ServiceClientSourceHeaders.vm +++ b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/ServiceClientSourceHeaders.vm @@ -12,8 +12,10 @@ #if($hasEventStreamRequest) #if($serviceModel.useSmithyClient) \#include +\#include #else \#include +\#include #end #end \#include diff --git a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/cbor/EventStreamHeader.vm b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/cbor/EventStreamHeader.vm index 68cb6f85d80..42fc6715c5f 100644 --- a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/cbor/EventStreamHeader.vm +++ b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/cbor/EventStreamHeader.vm @@ -9,6 +9,7 @@ \#include $header #end \#include +\#include #foreach($entry in $shape.members.entrySet()) #if($entry.value.shape.isEvent()) #foreach($eventMemberEntry in $entry.value.shape.members.entrySet()) @@ -43,6 +44,10 @@ namespace Model class $typeInfo.exportValue $typeInfo.className : public Aws::Utils::Event::EventEncoderStream { public: + ${typeInfo.className}() = default; + explicit ${typeInfo.className}(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} + #foreach($entry in $shape.members.entrySet()) #if($entry.value.shape.isEvent()) ${typeInfo.className}& Write${entry.value.shape.name}(const ${entry.value.shape.name}& value) diff --git a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/EventStreamHeader.vm b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/EventStreamHeader.vm index d21f5bd226c..c1107b35e6f 100644 --- a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/EventStreamHeader.vm +++ b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/EventStreamHeader.vm @@ -9,6 +9,7 @@ \#include $header #end \#include +\#include #foreach($entry in $shape.members.entrySet()) #if($entry.value.shape.isEvent()) #foreach($eventMemberEntry in $entry.value.shape.members.entrySet()) @@ -43,6 +44,9 @@ namespace Model class $typeInfo.exportValue $typeInfo.className : public Aws::Utils::Event::EventEncoderStream { public: + ${typeInfo.className}() = default; + explicit ${typeInfo.className}(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} #foreach($entry in $shape.members.entrySet()) #if($entry.value.shape.isEvent()) ${typeInfo.className}& Write${entry.value.shape.name}(const ${entry.value.shape.name}& value) diff --git a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/JsonServiceEventStreamOperationsSource.vm b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/JsonServiceEventStreamOperationsSource.vm index 35c186007ae..a71775efd9e 100644 --- a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/JsonServiceEventStreamOperationsSource.vm +++ b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/json/JsonServiceEventStreamOperationsSource.vm @@ -25,11 +25,48 @@ void ${className}::${operation.name}Async(Model::${operation.request.shape.name} #end #end #set($streamModelNameWithFirstLetterCapitalized = $CppViewHelper.capitalizeFirstChar($streamModelName)) - auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); +\#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, GetHttpClient()); + auto signer = GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER); + + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + eventEncoderStream->SetSigner(signer); + + auto requestCopy = Aws::MakeShared<${operation.request.shape.name}>(ALLOCATION_TAG, request); + request.Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream); + + auto& endpoint = endpointResolutionOutcome.GetResult(); + auto httpRequest = CreateHttpRequest( + endpoint.GetURI(), Aws::Http::HttpMethod::HTTP_${operation.http.method}, + Aws::Utils::Stream::DefaultResponseStreamFactoryMethod); + httpRequest->SetEventStreamRequest(true); + httpRequest->SetHasEventStreamResponse(true); + BuildHttpRequest(*requestCopy, httpRequest); + + if (!signer->SignRequest(*httpRequest, nullptr, nullptr, true)) { + handler(this, request, + ${operation.name}Outcome(Aws::Client::AWSError( + CoreErrors::CLIENT_SIGNING_FAILURE, "", "Failed to sign request", false)), + handlerContext); + return; + } + eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(*httpRequest)); + + Aws::Client::SubmitBidirectionalStreamingRequest< + ${className}, + ${operation.name}Outcome, + ${operation.request.shape.name}, + Model::${streamModelType}>( + this, request, requestCopy, eventEncoderStream, writeDataStreamBuf, + httpRequest, m_clientConfiguration.executor.get(), + streamReadyHandler, handler, handlerContext); +\#else + // Pull-based path (curl/WinHTTP) + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER)); auto requestCopy = Aws::MakeShared<${operation.request.shape.name}>("${operation.name}", request); - requestCopy->Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream); // this becomes the body of the request -##TODO 1.12: remove next line + requestCopy->Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream); request.Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream); auto asyncTask = CreateBidirectionalEventStreamTask<${operation.name}Outcome>(this, @@ -42,4 +79,5 @@ void ${className}::${operation.name}Async(Model::${operation.request.shape.name} m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif } diff --git a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyEventStreamHeader.vm b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyEventStreamHeader.vm index aac66739df0..ed578640b96 100644 --- a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyEventStreamHeader.vm +++ b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyEventStreamHeader.vm @@ -9,6 +9,7 @@ \#include $header #end \#include +\#include #foreach($entry in $shape.members.entrySet()) #if($entry.value.shape.isEvent()) #foreach($eventMemberEntry in $entry.value.shape.members.entrySet()) @@ -43,6 +44,9 @@ namespace Model class $typeInfo.exportValue $typeInfo.className : public Aws::Utils::Event::EventEncoderStream { public: + ${typeInfo.className}() = default; + explicit ${typeInfo.className}(std::shared_ptr streambuf) + : Aws::Utils::Event::EventEncoderStream(std::move(streambuf)) {} #foreach($entry in $shape.members.entrySet()) #if($entry.value.shape.isEvent()) ${typeInfo.className}& Write${entry.value.shape.name}(const ${entry.value.shape.name}& value) diff --git a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyJsonServiceEventStreamOperationsSource.vm b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyJsonServiceEventStreamOperationsSource.vm index a307ee20b62..7ece21874a2 100644 --- a/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyJsonServiceEventStreamOperationsSource.vm +++ b/tools/code-generation/generator/src/main/resources/com/amazonaws/util/awsclientgenerator/velocity/cpp/smithy/SmithyJsonServiceEventStreamOperationsSource.vm @@ -27,7 +27,31 @@ void ${className}::${operation.name}Async(Model::${operation.request.shape.name} #end #end #set($streamModelNameWithFirstLetterCapitalized = $CppViewHelper.capitalizeFirstChar($streamModelName)) - auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); +\#if AWS_SDK_USE_CRT_HTTP + // Push-based WriteData path (CRT HTTP client only) + auto writeDataStreamBuf = Aws::MakeShared(ALLOCATION_TAG, m_httpClient); + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG, writeDataStreamBuf); + request.Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream); + + auto requestCopy = Aws::MakeShared<${operation.request.shape.name}>(ALLOCATION_TAG, request); + + auto authCallback = [&](std::shared_ptr ctx) -> void { + eventEncoderStream->SetSigningCallback([this, ctx, eventEncoderStream](Aws::Utils::Event::Message& message, Aws::String& seed) -> bool { + auto outcome = SignEventMessage(message, seed, ctx); + return outcome.IsSuccess(); + }); + }; + + auto asyncTask = smithy::client::CreateSmithyBidirectionalWriteDataTask<${operation.name}Outcome>( + this, requestCopy, handler, handlerContext, eventEncoderStream, writeDataStreamBuf, + std::move(endpointCallback), std::move(authCallback)); + auto sem = asyncTask.GetSemaphore(); + m_clientConfiguration.executor->Submit(std::move(asyncTask)); + sem->WaitOne(); + streamReadyHandler(*eventEncoderStream); +\#else + // Pull-based path + auto eventEncoderStream = Aws::MakeShared(ALLOCATION_TAG); auto authCallback = [&](std::shared_ptr ctx) -> void { eventEncoderStream->SetSigningCallback([this, ctx, eventEncoderStream](Aws::Utils::Event::Message& message, Aws::String& seed) -> bool { auto outcome = SignEventMessage(message, seed, ctx); @@ -35,8 +59,8 @@ void ${className}::${operation.name}Async(Model::${operation.request.shape.name} }); }; auto requestCopy = Aws::MakeShared<${operation.request.shape.name}>("${operation.name}", request); - requestCopy->Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream); // this becomes the body of the request - request.Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream); // this becomes the body of the request + requestCopy->Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream); + request.Set${streamModelNameWithFirstLetterCapitalized}(eventEncoderStream); auto asyncTask = smithy::client::CreateSmithyBidirectionalEventStreamTask<${operation.name}Outcome>(this, requestCopy, @@ -49,4 +73,5 @@ void ${className}::${operation.name}Async(Model::${operation.request.shape.name} m_clientConfiguration.executor->Submit(std::move(asyncTask)); sem->WaitOne(); streamReadyHandler(*eventEncoderStream); +#endif }