Copy bi-directional event stream request (#3311)

* Copy bi-directional event stream operation request

* Code generation example

Author: SergeyRyabinin

generated/src/aws-cpp-sdk-kinesis/include/aws/kinesis/model/SubscribeToShardHandler.h

@@ -36,6 +36,7 @@ namespace Model
     public:
         AWS_KINESIS_API SubscribeToShardHandler();
         AWS_KINESIS_API SubscribeToShardHandler& operator=(const SubscribeToShardHandler&) = default;
+        AWS_KINESIS_API SubscribeToShardHandler(const SubscribeToShardHandler&) = default;
 
         AWS_KINESIS_API virtual void OnEvent() override;
 

generated/src/aws-cpp-sdk-kinesis/include/aws/kinesis/model/SubscribeToShardRequest.h

@@ -45,7 +45,7 @@ namespace Model
     /**
      * Underlying Event Stream Handler which is used to define callback functions.
      */
-    inline const SubscribeToShardHandler& GetEventStreamHandler() const { return m_handler; }
+    inline SubscribeToShardHandler& GetEventStreamHandler() { return m_handler; }
 
     /**
      * Underlying Event Stream Handler which is used to define callback functions.

generated/src/aws-cpp-sdk-kinesis/source/KinesisClient.cpp

@@ -836,6 +836,14 @@ SubscribeToShardOutcome KinesisClient::SubscribeToShard(SubscribeToShardRequest&
       request.SetResponseStreamFactory(
           [&] { request.GetEventStreamDecoder().Reset(); return Aws::New<Aws::Utils::Event::EventDecoderStream>(ALLOCATION_TAG, request.GetEventStreamDecoder()); }
       );
+      if (!request.GetHeadersReceivedEventHandler()) {
+        request.SetHeadersReceivedEventHandler([&request](const Http::HttpRequest*, Http::HttpResponse* response) {
+          AWS_CHECK_PTR("SubscribeToShard", response);
+          if (const auto initialResponseHandler = request.GetEventStreamHandler().GetInitialResponseCallbackEx()) {
+            initialResponseHandler({response->GetHeaders()}, Utils::Event::InitialResponseType::ON_RESPONSE);
+          }
+        });
+      }
       return SubscribeToShardOutcome(MakeRequestDeserialize(&request, request.GetServiceRequestName(), Aws::Http::HttpMethod::HTTP_POST, [&](Aws::Endpoint::AWSEndpoint& resolvedEndpoint) ->  void {
     AWS_UNREFERENCED_PARAM(resolvedEndpoint);
       }));

generated/src/aws-cpp-sdk-lambda/include/aws/lambda/model/InvokeWithResponseStreamHandler.h

@@ -39,6 +39,7 @@ namespace Model
     public:
         AWS_LAMBDA_API InvokeWithResponseStreamHandler();
         AWS_LAMBDA_API InvokeWithResponseStreamHandler& operator=(const InvokeWithResponseStreamHandler&) = default;
+        AWS_LAMBDA_API InvokeWithResponseStreamHandler(const InvokeWithResponseStreamHandler&) = default;
 
         AWS_LAMBDA_API virtual void OnEvent() override;
 

generated/src/aws-cpp-sdk-lambda/include/aws/lambda/model/InvokeWithResponseStreamRequest.h

@@ -51,7 +51,7 @@ namespace Model
     /**
      * Underlying Event Stream Handler which is used to define callback functions.
      */
-    inline const InvokeWithResponseStreamHandler& GetEventStreamHandler() const { return m_handler; }
+    inline InvokeWithResponseStreamHandler& GetEventStreamHandler() { return m_handler; }
 
     /**
      * Underlying Event Stream Handler which is used to define callback functions.

generated/src/aws-cpp-sdk-lambda/source/LambdaClient.cpp

@@ -1467,6 +1467,14 @@ InvokeWithResponseStreamOutcome LambdaClient::InvokeWithResponseStream(InvokeWit
       request.SetResponseStreamFactory(
           [&] { request.GetEventStreamDecoder().Reset(); return Aws::New<Aws::Utils::Event::EventDecoderStream>(ALLOCATION_TAG, request.GetEventStreamDecoder()); }
       );
+      if (!request.GetHeadersReceivedEventHandler()) {
+        request.SetHeadersReceivedEventHandler([&request](const Http::HttpRequest*, Http::HttpResponse* response) {
+          AWS_CHECK_PTR("InvokeWithResponseStream", response);
+          if (const auto initialResponseHandler = request.GetEventStreamHandler().GetInitialResponseCallbackEx()) {
+            initialResponseHandler({response->GetHeaders()}, Utils::Event::InitialResponseType::ON_RESPONSE);
+          }
+        });
+      }
       return InvokeWithResponseStreamOutcome(MakeRequest(request, endpointResolutionOutcome.GetResult(), Aws::Http::HttpMethod::HTTP_POST));
     },
     TracingUtils::SMITHY_CLIENT_DURATION_METRIC,

generated/src/aws-cpp-sdk-lambda/source/model/InvokeWithResponseStreamRequest.cpp

@@ -27,13 +27,6 @@ InvokeWithResponseStreamRequest::InvokeWithResponseStreamRequest() :
     m_qualifierHasBeenSet(false),
     m_handler(), m_decoder(Aws::Utils::Event::EventStreamDecoder(&m_handler))
 {
-    AmazonWebServiceRequest::SetHeadersReceivedEventHandler([this](const Http::HttpRequest*, Http::HttpResponse* response)
-    {
-        auto& initialResponseHandler = m_handler.GetInitialResponseCallbackEx();
-        if (initialResponseHandler) {
-            initialResponseHandler(InvokeWithResponseStreamInitialResponse(response->GetHeaders()), Utils::Event::InitialResponseType::ON_RESPONSE);
-        }
-    });
 }
 
 

generated/src/aws-cpp-sdk-lexv2-runtime/include/aws/lexv2-runtime/model/StartConversationHandler.h

@@ -51,6 +51,7 @@ namespace Model
     public:
         AWS_LEXRUNTIMEV2_API StartConversationHandler();
         AWS_LEXRUNTIMEV2_API StartConversationHandler& operator=(const StartConversationHandler&) = default;
+        AWS_LEXRUNTIMEV2_API StartConversationHandler(const StartConversationHandler&) = default;
 
         AWS_LEXRUNTIMEV2_API virtual void OnEvent() override;
 

generated/src/aws-cpp-sdk-lexv2-runtime/include/aws/lexv2-runtime/model/StartConversationRequest.h

@@ -50,7 +50,7 @@ namespace Model
     /**
      * Underlying Event Stream Handler which is used to define callback functions.
      */
-    inline const StartConversationHandler& GetEventStreamHandler() const { return m_handler; }
+    inline StartConversationHandler& GetEventStreamHandler() { return m_handler; }
 
     /**
      * Underlying Event Stream Handler which is used to define callback functions.

generated/src/aws-cpp-sdk-lexv2-runtime/source/LexRuntimeV2Client.cpp

@@ -17,6 +17,7 @@
 #include <aws/core/utils/DNS.h>
 #include <aws/core/utils/logging/LogMacros.h>
 #include <aws/core/utils/logging/ErrorMacros.h>
+#include <aws/core/client/AWSClientEventStreamingAsyncTask.h>
 #include <aws/core/utils/event/EventStream.h>
 
 #include <aws/lexv2-runtime/LexRuntimeV2Client.h>
@@ -502,29 +503,21 @@ void LexRuntimeV2Client::StartConversationAsync(Model::StartConversationRequest&
   endpointResolutionOutcome.GetResult().AddPathSegments("/sessions/");
   endpointResolutionOutcome.GetResult().AddPathSegment(request.GetSessionId());
   endpointResolutionOutcome.GetResult().AddPathSegments("/conversation");
-  request.SetResponseStreamFactory(
-      [&] { request.GetEventStreamDecoder().Reset(); return Aws::New<Aws::Utils::Event::EventDecoderStream>(ALLOCATION_TAG, request.GetEventStreamDecoder()); }
-  );
 
   auto eventEncoderStream = Aws::MakeShared<Model::StartConversationRequestEventStream>(ALLOCATION_TAG);
   eventEncoderStream->SetSigner(GetSignerByName(Aws::Auth::EVENTSTREAM_SIGV4_SIGNER));
-  request.SetRequestEventStream(eventEncoderStream); // this becomes the body of the request
-  auto sem = Aws::MakeShared<Aws::Utils::Threading::Semaphore>(ALLOCATION_TAG, 0, 1);
-  request.SetRequestSignedHandler([eventEncoderStream, sem](const Aws::Http::HttpRequest& httpRequest) { eventEncoderStream->SetSignatureSeed(Aws::Client::GetAuthorizationHeader(httpRequest)); sem->ReleaseAll(); });
+  auto requestCopy = Aws::MakeShared<StartConversationRequest>("StartConversation", request);
+  requestCopy->SetRequestEventStream(eventEncoderStream); // this becomes the body of the request
+  request.SetRequestEventStream(eventEncoderStream);
 
-  m_clientConfiguration.executor->Submit([this, endpointResolutionOutcome, &request, handler, handlerContext] () mutable {
-      JsonOutcome outcome = MakeRequest(request, endpointResolutionOutcome.GetResult(), Aws::Http::HttpMethod::HTTP_POST, Aws::Auth::EVENTSTREAM_SIGV4_SIGNER);
-      if(outcome.IsSuccess())
-      {
-        handler(this, request, StartConversationOutcome(NoResult()), handlerContext);
-      }
-      else
-      {
-        request.GetRequestEventStream()->Close();
-        handler(this, request, StartConversationOutcome(outcome.GetError()), handlerContext);
-      }
-      return StartConversationOutcome(NoResult());
-  });
+  auto asyncTask = CreateBidirectionalEventStreamTask<StartConversationOutcome>(this,
+                                         endpointResolutionOutcome.GetResultWithOwnership(),
+                                         requestCopy,
+                                         handler,
+                                         handlerContext,
+                                         eventEncoderStream);
+  auto sem = asyncTask.GetSemaphore();
+  m_clientConfiguration.executor->Submit(std::move(asyncTask));
   sem->WaitOne();
-  streamReadyHandler(*request.GetRequestEventStream());
+  streamReadyHandler(*eventEncoderStream);
 }