Before changing ClientOperation::Close behavior

yourslab · yourslab · commit 39b1d18b0d6a · 2021-06-23T09:40:06.000-07:00
diff --git a/eventstream_rpc/include/aws/eventstreamrpc/EventStreamClient.h b/eventstream_rpc/include/aws/eventstreamrpc/EventStreamClient.h
@@ -266,6 +266,7 @@ namespace Aws
             EventStreamRpcStatusCode baseStatus;
             int crtError;
             operator bool() const noexcept { return baseStatus == EVENT_STREAM_RPC_SUCCESS; }
+            Crt::String ErrorToString();
         };
 
         template <class T> class ProtectedPromise
@@ -407,6 +408,7 @@ namespace Aws
                 uint32_t messageFlags,
                 OnMessageFlushCallback onMessageFlushCallback) noexcept;
             bool IsClosed() noexcept;
+            void Close() noexcept;
             std::future<RpcError> SendMessage(
                 const Crt::List<EventStreamHeader> &headers,
                 const Crt::Optional<Crt::ByteBuf> &payload,
@@ -436,7 +438,7 @@ namespace Aws
             virtual void SerializeToJsonObject(Crt::JsonObject &payloadObject) const = 0;
             virtual Crt::String GetModelName() const noexcept = 0;
 
-          private:
+          protected:
             Crt::Allocator *m_allocator;
         };
 
@@ -635,8 +637,7 @@ namespace Aws
             ClientContinuation m_clientContinuation;
             ProtectedPromise<TaggedResult> m_initialResponsePromise;
             /* ProtectedPromise not necessary because it's only ever being set by one thread. */
-            std::promise<void> m_closedPromise;
-            std::atomic<bool> m_isClosed;
+            std::promise<RpcError> m_closedPromise;
         };
     } // namespace Eventstreamrpc
 } // namespace Aws
diff --git a/eventstream_rpc/source/EventStreamClient.cpp b/eventstream_rpc/source/EventStreamClient.cpp
@@ -494,7 +494,7 @@ namespace Aws
                 return m_closedPromise.get_future();
             } else {
                 std::promise<RpcError> closedPromise;
-                closedPromise.set_value({EVENT_STREAM_RPC_CONNECTION_CLOSED, 0});
+                closedPromise.set_value({EVENT_STREAM_RPC_SUCCESS, 0});
                 return closedPromise.get_future();
             }
         }
@@ -749,7 +749,7 @@ namespace Aws
 
         void AbstractShapeBase::s_customDeleter(AbstractShapeBase *shape) noexcept
         {
-            if (shape->m_allocator)
+            if (shape->m_allocator != nullptr)
                 Crt::Delete<AbstractShapeBase>(shape, shape->m_allocator);
         }
 
@@ -768,7 +768,13 @@ namespace Aws
                 aws_event_stream_rpc_client_connection_new_stream(connection->m_underlyingConnection, &options);
         }
 
-        ClientContinuation::~ClientContinuation() noexcept {}
+        void ClientContinuation::Close() noexcept {
+            aws_event_stream_rpc_client_continuation_release(m_continuationToken);
+        }
+
+        ClientContinuation::~ClientContinuation() noexcept {
+            Close();
+        }
 
         void ClientContinuation::s_onContinuationMessage(
             struct aws_event_stream_rpc_client_continuation_token *continuationToken,
@@ -962,7 +968,7 @@ namespace Aws
             const OperationModelContext &operationModelContext,
             Crt::Allocator *allocator) noexcept
             : m_operationModelContext(operationModelContext), m_messageCount(0), m_allocator(allocator),
-              m_streamHandler(streamHandler), m_clientContinuation(connection.NewStream(*this)), m_isClosed(false)
+              m_streamHandler(streamHandler), m_clientContinuation(connection.NewStream(*this))
         {
         }
 
@@ -973,7 +979,6 @@ namespace Aws
               m_initialResponsePromise(std::move(rhs.m_initialResponsePromise)),
               m_closedPromise(std::move(rhs.m_closedPromise))
         {
-            m_isClosed.store(rhs.m_isClosed.load());
         }
 
         ClientOperation::~ClientOperation() noexcept { Close().wait(); }
@@ -1146,11 +1151,6 @@ namespace Aws
         {
             bool streamAlreadyTerminated = messageFlags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM;
 
-            if (streamAlreadyTerminated)
-            {
-                m_isClosed.store(true);
-            }
-
             Crt::StringView payloadStringView;
             if (payload.has_value())
             {
@@ -1169,7 +1169,7 @@ namespace Aws
                 m_initialResponsePromise.SetValue(std::move(taggedResult));
                 /* Close the stream unless the server already closed it for us. This condition is checked
                  * so that TERMINATE_STREAM messages aren't resent by the client. */
-                if (!streamAlreadyTerminated && !m_isClosed.exchange(true))
+                if (!streamAlreadyTerminated && !m_clientContinuation.IsClosed())
                 {
                     Close().wait();
                 }
@@ -1179,7 +1179,7 @@ namespace Aws
                 bool shouldCloseNow = true;
                 if (m_streamHandler)
                     shouldCloseNow = m_streamHandler->OnStreamError(std::move(error), {EVENT_STREAM_RPC_SUCCESS, 0});
-                if (!streamAlreadyTerminated && shouldCloseNow && !m_isClosed.exchange(true))
+                if (!streamAlreadyTerminated && shouldCloseNow && !m_clientContinuation.IsClosed())
                 {
                     Close().wait();
                 }
@@ -1222,10 +1222,11 @@ namespace Aws
                 contentHeader = GetHeaderByName(headers, Crt::String(CONTENT_TYPE_HEADER));
                 if (contentHeader == nullptr)
                 {
+                    /* TODO: Log an error. */
                     /* Missing required content type header. */
                     errorCode = EVENT_STREAM_RPC_UNMAPPED_DATA;
                 }
-                else if (contentHeader->GetValueAsString(contentType) && contentType != CONTENT_TYPE_APPLICATION_JSON)
+                else if (contentHeader == nullptr && contentHeader->GetValueAsString(contentType) && contentType != CONTENT_TYPE_APPLICATION_JSON)
                 {
                     errorCode = EVENT_STREAM_RPC_UNSUPPORTED_CONTENT_TYPE;
                 }
@@ -1257,7 +1258,7 @@ namespace Aws
                     bool shouldClose = true;
                     if (m_streamHandler)
                         shouldClose = m_streamHandler->OnStreamError(nullptr, {errorCode, 0});
-                    if (!m_isClosed.load() && shouldClose)
+                    if (!m_clientContinuation.IsClosed() && shouldClose)
                     {
                         Close().wait();
                     }
@@ -1298,7 +1299,7 @@ namespace Aws
              * potentially set by `OnContinuationMessage` will just do nothing. */
             m_initialResponsePromise.SetValue(TaggedResult({EVENT_STREAM_RPC_CONTINUATION_CLOSED, 0}));
 
-            m_closedPromise.set_value();
+            m_closedPromise.set_value({EVENT_STREAM_RPC_SUCCESS, 0});
 
             if (m_streamHandler)
             {
@@ -1308,20 +1309,22 @@ namespace Aws
 
         std::future<RpcError> ClientOperation::Close(OnMessageFlushCallback onMessageFlushCallback) noexcept
         {
-            if (m_isClosed.load())
+            if (m_clientContinuation.IsClosed())
             {
                 std::promise<RpcError> alreadyClosedPromise;
                 alreadyClosedPromise.set_value({EVENT_STREAM_RPC_CONTINUATION_CLOSED, 0});
                 return alreadyClosedPromise.get_future();
             }
             else
             {
-                return m_clientContinuation.SendMessage(
+                auto rpcError = m_clientContinuation.SendMessage(
                     Crt::List<EventStreamHeader>(),
                     Crt::Optional<Crt::ByteBuf>(),
                     AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_APPLICATION_MESSAGE,
                     AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM,
-                    onMessageFlushCallback);
+                    onMessageFlushCallback).get();
+                m_clientContinuation.Close();
+                return m_closedPromise.get_future();
             }
         }
 
diff --git a/eventstream_rpc/tests/CMakeLists.txt b/eventstream_rpc/tests/CMakeLists.txt
@@ -44,6 +44,7 @@ aws_use_package(aws-crt-cpp)
 aws_use_package(EventstreamRpc-cpp)
 
 add_test_case(EventStreamConnect)
+add_test_case(EchoOperation)
 add_test_case(OperateWhileDisconnected)
 generate_cpp_test_driver(${TEST_BINARY_NAME})
 target_include_directories(${TEST_BINARY_NAME} PUBLIC
diff --git a/eventstream_rpc/tests/EchoTestRpcClient.cpp b/eventstream_rpc/tests/EchoTestRpcClient.cpp
@@ -10,6 +10,7 @@ namespace Awstest
     {
         m_echoTestRpcServiceModel.AssignModelNameToErrorResponse(
             Aws::Crt::String("awstest#ServiceError"), ServiceError::s_allocateFromPayload);
+        aws_event_stream_library_init(m_allocator);
     }
 
     std::future<RpcError> EchoTestRpcClient::Connect(
@@ -21,7 +22,7 @@ namespace Awstest
 
     std::future<RpcError> EchoTestRpcClient::Close() noexcept { return m_connection.Close(); }
 
-    EchoTestRpcClient::~EchoTestRpcClient() noexcept { Close().wait(); }
+    EchoTestRpcClient::~EchoTestRpcClient() noexcept { Close().wait(); aws_event_stream_library_clean_up(); }
 
     GetAllProductsOperation EchoTestRpcClient::NewGetAllProducts() noexcept
     {
diff --git a/eventstream_rpc/tests/EventStreamClientTest.cpp b/eventstream_rpc/tests/EventStreamClientTest.cpp
@@ -15,6 +15,7 @@ using namespace Aws::Eventstreamrpc;
 using namespace Awstest;
 
 static int s_TestEventStreamConnect(struct aws_allocator *allocator, void *ctx);
+static int s_TestEchoOperation(struct aws_allocator *allocator, void *ctx);
 static int s_TestOperationWhileDisconnected(struct aws_allocator *allocator, void *ctx);
 
 class TestLifecycleHandler : public ConnectionLifecycleHandler
@@ -75,7 +76,7 @@ static int s_TestEventStreamConnect(struct aws_allocator *allocator, void *ctx)
         Aws::Crt::Io::EventLoopGroup eventLoopGroup(0, allocator);
         ASSERT_TRUE(eventLoopGroup);
 
-        Aws::Crt::Io::DefaultHostResolver defaultHostResolver(eventLoopGroup, 8, 30, allocator);
+        Aws::Crt::Io::DefaultHostResolver defaultHostResolver(eventLoopGroup, 1, 1, allocator);
         ASSERT_TRUE(defaultHostResolver);
 
         Aws::Crt::Io::ClientBootstrap clientBootstrap(eventLoopGroup, defaultHostResolver, allocator);
@@ -87,6 +88,7 @@ static int s_TestEventStreamConnect(struct aws_allocator *allocator, void *ctx)
         ConnectionConfig config;
         config.SetHostName(Aws::Crt::String("127.0.0.1"));
         config.SetPort(8033U);
+        aws_event_stream_library_init(allocator);
 
         /* Empty amendment headers. */
         {
@@ -115,6 +117,8 @@ static int s_TestEventStreamConnect(struct aws_allocator *allocator, void *ctx)
             ASSERT_TRUE(connectedStatus.get().baseStatus == EVENT_STREAM_RPC_SUCCESS);
             client.Close().wait();
         }
+
+        //aws_event_stream_library_clean_up();
     }
 
     return AWS_OP_SUCCESS;
@@ -134,7 +138,7 @@ static int s_TestOperationWhileDisconnected(struct aws_allocator *allocator, voi
         Aws::Crt::Io::EventLoopGroup eventLoopGroup(0, allocator);
         ASSERT_TRUE(eventLoopGroup);
 
-        Aws::Crt::Io::DefaultHostResolver defaultHostResolver(eventLoopGroup, 8, 30, allocator);
+        Aws::Crt::Io::DefaultHostResolver defaultHostResolver(eventLoopGroup, 1, 1, allocator);
         ASSERT_TRUE(defaultHostResolver);
 
         Aws::Crt::Io::ClientBootstrap clientBootstrap(eventLoopGroup, defaultHostResolver, allocator);
@@ -148,30 +152,73 @@ static int s_TestOperationWhileDisconnected(struct aws_allocator *allocator, voi
             Awstest::EchoTestRpcClient client(clientBootstrap, allocator);
             auto connectedStatus = client.Connect(lifecycleHandler);
             ASSERT_TRUE(connectedStatus.get().baseStatus == EVENT_STREAM_RPC_SUCCESS);
+            ASSERT_TRUE(client.Close().get().baseStatus == EVENT_STREAM_RPC_SUCCESS);
             auto echoMessage = client.NewEchoMessage();
             EchoMessageRequest echoMessageRequest;
             MessageData messageData;
-            messageData.SetStringMessage("wtf");
+            Aws::Crt::String expectedMessage("wtf");
+            messageData.SetStringMessage(expectedMessage);
             echoMessageRequest.SetMessage(messageData);
             auto requestFuture = echoMessage.Activate(echoMessageRequest, s_onMessageFlush);
             ASSERT_TRUE(requestFuture.get().baseStatus == EVENT_STREAM_RPC_CONNECTION_CLOSED);
             auto result = echoMessage.GetResult().get();
-            if (result) {
-                std::cout << result.GetOperationResponse() << std::endl;
-            } else {
-                auto errorType = result.GetResultType();
-                if (errorType == RPC_ERROR) {
-                    std::cout << result.GetRpcError() << std::endl;
-                } else {
-                    auto *error = result.GetOperationError();
-                    (void)error;
-                }
-            }
+            ASSERT_TRUE(result);
+            auto error = result.GetRpcError();
+            ASSERT_TRUE(error.baseStatus == EVENT_STREAM_RPC_CONNECTION_CLOSED);
+        }
+    }
+
+    return AWS_OP_SUCCESS;
+}
+
+static int s_TestEchoOperation(struct aws_allocator *allocator, void *ctx)
+{
+    (void)ctx;
+    {
+        Aws::Crt::ApiHandle apiHandle(allocator);
+        Aws::Crt::Io::TlsContextOptions tlsCtxOptions = Aws::Crt::Io::TlsContextOptions::InitDefaultClient();
+        Aws::Crt::Io::TlsContext tlsContext(tlsCtxOptions, Aws::Crt::Io::TlsMode::CLIENT, allocator);
+        ASSERT_TRUE(tlsContext);
+
+        Aws::Crt::Io::TlsConnectionOptions tlsConnectionOptions = tlsContext.NewConnectionOptions();
+
+        Aws::Crt::Io::EventLoopGroup eventLoopGroup(0, allocator);
+        ASSERT_TRUE(eventLoopGroup);
+
+        Aws::Crt::Io::DefaultHostResolver defaultHostResolver(eventLoopGroup, 8, 30, allocator);
+        ASSERT_TRUE(defaultHostResolver);
+
+        Aws::Crt::Io::ClientBootstrap clientBootstrap(eventLoopGroup, defaultHostResolver, allocator);
+        ASSERT_TRUE(clientBootstrap);
+
+        clientBootstrap.EnableBlockingShutdown();
+        ConnectionLifecycleHandler lifecycleHandler;
+        Awstest::EchoTestRpcClient client(clientBootstrap, allocator);
+
+        {
+            //aws_event_stream_library_init(allocator);
+            auto connectedStatus = client.Connect(lifecycleHandler);
+            ASSERT_TRUE(connectedStatus.get().baseStatus == EVENT_STREAM_RPC_SUCCESS);
+            auto echoMessage = client.NewEchoMessage();
+            EchoMessageRequest echoMessageRequest;
+            MessageData messageData;
+            Aws::Crt::String expectedMessage("wtf");
+            messageData.SetStringMessage(expectedMessage);
+            echoMessageRequest.SetMessage(messageData);
+            auto requestFuture = echoMessage.Activate(echoMessageRequest, s_onMessageFlush);
+            requestFuture.wait();
+            auto result = echoMessage.GetResult().get();
+            ASSERT_TRUE(result);
+            auto response = result.GetOperationResponse();
+            ASSERT_NOT_NULL(response);
+            ASSERT_TRUE(response->GetMessage().value().GetStringMessage().value() == expectedMessage);
+            //aws_event_stream_library_clean_up();
         }
     }
 
     return AWS_OP_SUCCESS;
 }
 
 AWS_TEST_CASE(EventStreamConnect, s_TestEventStreamConnect)
+AWS_TEST_CASE(EchoOperation, s_TestEchoOperation)
 AWS_TEST_CASE(OperateWhileDisconnected, s_TestOperationWhileDisconnected)
diff --git a/eventstream_rpc/tests/include/awstest/EchoTestRpcModel.h b/eventstream_rpc/tests/include/awstest/EchoTestRpcModel.h
diff --git a/greengrass_ipc/include/aws/greengrass/GreengrassCoreIpcModel.h b/greengrass_ipc/include/aws/greengrass/GreengrassCoreIpcModel.h

Original file line number	Diff line number	Diff line change
`@@ -494,7 +494,7 @@ namespace Aws`
`494`	`494`	`return m_closedPromise.get_future();`
`495`	`495`	`} else {`
`496`	`496`	`std::promise<RpcError> closedPromise;`
`497`		`- closedPromise.set_value({EVENT_STREAM_RPC_CONNECTION_CLOSED, 0});`
	`497`	`+ closedPromise.set_value({EVENT_STREAM_RPC_SUCCESS, 0});`
`498`	`498`	`return closedPromise.get_future();`
`499`	`499`	`}`
`500`	`500`	`}`
`@@ -749,7 +749,7 @@ namespace Aws`
`749`	`749`
`750`	`750`	`void AbstractShapeBase::s_customDeleter(AbstractShapeBase *shape) noexcept`
`751`	`751`	`{`
`752`		`- if (shape->m_allocator)`
	`752`	`+ if (shape->m_allocator != nullptr)`
`753`	`753`	`Crt::Delete<AbstractShapeBase>(shape, shape->m_allocator);`
`754`	`754`	`}`
`755`	`755`
`@@ -768,7 +768,13 @@ namespace Aws`
`768`	`768`	`aws_event_stream_rpc_client_connection_new_stream(connection->m_underlyingConnection, &options);`
`769`	`769`	`}`
`770`	`770`
`771`		`- ClientContinuation::~ClientContinuation() noexcept {}`
	`771`	`+ void ClientContinuation::Close() noexcept {`
	`772`	`+ aws_event_stream_rpc_client_continuation_release(m_continuationToken);`
	`773`	`+ }`
	`774`	`+`
	`775`	`+ ClientContinuation::~ClientContinuation() noexcept {`
	`776`	`+ Close();`
	`777`	`+ }`
`772`	`778`
`773`	`779`	`void ClientContinuation::s_onContinuationMessage(`
`774`	`780`	`struct aws_event_stream_rpc_client_continuation_token *continuationToken,`
`@@ -962,7 +968,7 @@ namespace Aws`
`962`	`968`	`const OperationModelContext &operationModelContext,`
`963`	`969`	`Crt::Allocator *allocator) noexcept`
`964`	`970`	`: m_operationModelContext(operationModelContext), m_messageCount(0), m_allocator(allocator),`
`965`		`- m_streamHandler(streamHandler), m_clientContinuation(connection.NewStream(*this)), m_isClosed(false)`
	`971`	`+ m_streamHandler(streamHandler), m_clientContinuation(connection.NewStream(*this))`
`966`	`972`	`{`
`967`	`973`	`}`
`968`	`974`
`@@ -973,7 +979,6 @@ namespace Aws`
`973`	`979`	`m_initialResponsePromise(std::move(rhs.m_initialResponsePromise)),`
`974`	`980`	`m_closedPromise(std::move(rhs.m_closedPromise))`
`975`	`981`	`{`
`976`		`- m_isClosed.store(rhs.m_isClosed.load());`
`977`	`982`	`}`
`978`	`983`
`979`	`984`	`ClientOperation::~ClientOperation() noexcept { Close().wait(); }`
`@@ -1146,11 +1151,6 @@ namespace Aws`
`1146`	`1151`	`{`
`1147`	`1152`	`bool streamAlreadyTerminated = messageFlags & AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM;`
`1148`	`1153`
`1149`		`- if (streamAlreadyTerminated)`
`1150`		`- {`
`1151`		`- m_isClosed.store(true);`
`1152`		`- }`
`1153`		`-`
`1154`	`1154`	`Crt::StringView payloadStringView;`
`1155`	`1155`	`if (payload.has_value())`
`1156`	`1156`	`{`
`@@ -1169,7 +1169,7 @@ namespace Aws`
`1169`	`1169`	`m_initialResponsePromise.SetValue(std::move(taggedResult));`
`1170`	`1170`	`/* Close the stream unless the server already closed it for us. This condition is checked`
`1171`	`1171`	`* so that TERMINATE_STREAM messages aren't resent by the client. */`
`1172`		`- if (!streamAlreadyTerminated && !m_isClosed.exchange(true))`
	`1172`	`+ if (!streamAlreadyTerminated && !m_clientContinuation.IsClosed())`
`1173`	`1173`	`{`
`1174`	`1174`	`Close().wait();`
`1175`	`1175`	`}`
`@@ -1179,7 +1179,7 @@ namespace Aws`
`1179`	`1179`	`bool shouldCloseNow = true;`
`1180`	`1180`	`if (m_streamHandler)`
`1181`	`1181`	`shouldCloseNow = m_streamHandler->OnStreamError(std::move(error), {EVENT_STREAM_RPC_SUCCESS, 0});`
`1182`		`- if (!streamAlreadyTerminated && shouldCloseNow && !m_isClosed.exchange(true))`
	`1182`	`+ if (!streamAlreadyTerminated && shouldCloseNow && !m_clientContinuation.IsClosed())`
`1183`	`1183`	`{`
`1184`	`1184`	`Close().wait();`
`1185`	`1185`	`}`
`@@ -1222,10 +1222,11 @@ namespace Aws`
`1222`	`1222`	`contentHeader = GetHeaderByName(headers, Crt::String(CONTENT_TYPE_HEADER));`
`1223`	`1223`	`if (contentHeader == nullptr)`
`1224`	`1224`	`{`
	`1225`	`+ /* TODO: Log an error. */`
`1225`	`1226`	`/* Missing required content type header. */`
`1226`	`1227`	`errorCode = EVENT_STREAM_RPC_UNMAPPED_DATA;`
`1227`	`1228`	`}`
`1228`		`- else if (contentHeader->GetValueAsString(contentType) && contentType != CONTENT_TYPE_APPLICATION_JSON)`
	`1229`	`+ else if (contentHeader == nullptr && contentHeader->GetValueAsString(contentType) && contentType != CONTENT_TYPE_APPLICATION_JSON)`
`1229`	`1230`	`{`
`1230`	`1231`	`errorCode = EVENT_STREAM_RPC_UNSUPPORTED_CONTENT_TYPE;`
`1231`	`1232`	`}`
`@@ -1257,7 +1258,7 @@ namespace Aws`
`1257`	`1258`	`bool shouldClose = true;`
`1258`	`1259`	`if (m_streamHandler)`
`1259`	`1260`	`shouldClose = m_streamHandler->OnStreamError(nullptr, {errorCode, 0});`
`1260`		`- if (!m_isClosed.load() && shouldClose)`
	`1261`	`+ if (!m_clientContinuation.IsClosed() && shouldClose)`
`1261`	`1262`	`{`
`1262`	`1263`	`Close().wait();`
`1263`	`1264`	`}`
`@@ -1298,7 +1299,7 @@ namespace Aws`
`1298`	`1299`	* potentially set by `OnContinuationMessage` will just do nothing. */
`1299`	`1300`	`m_initialResponsePromise.SetValue(TaggedResult({EVENT_STREAM_RPC_CONTINUATION_CLOSED, 0}));`
`1300`	`1301`
`1301`		`- m_closedPromise.set_value();`
	`1302`	`+ m_closedPromise.set_value({EVENT_STREAM_RPC_SUCCESS, 0});`
`1302`	`1303`
`1303`	`1304`	`if (m_streamHandler)`
`1304`	`1305`	`{`
`@@ -1308,20 +1309,22 @@ namespace Aws`
`1308`	`1309`
`1309`	`1310`	`std::future<RpcError> ClientOperation::Close(OnMessageFlushCallback onMessageFlushCallback) noexcept`
`1310`	`1311`	`{`
`1311`		`- if (m_isClosed.load())`
	`1312`	`+ if (m_clientContinuation.IsClosed())`
`1312`	`1313`	`{`
`1313`	`1314`	`std::promise<RpcError> alreadyClosedPromise;`
`1314`	`1315`	`alreadyClosedPromise.set_value({EVENT_STREAM_RPC_CONTINUATION_CLOSED, 0});`
`1315`	`1316`	`return alreadyClosedPromise.get_future();`
`1316`	`1317`	`}`
`1317`	`1318`	`else`
`1318`	`1319`	`{`
`1319`		`- return m_clientContinuation.SendMessage(`
	`1320`	`+ auto rpcError = m_clientContinuation.SendMessage(`
`1320`	`1321`	`Crt::List<EventStreamHeader>(),`
`1321`	`1322`	`Crt::Optional<Crt::ByteBuf>(),`
`1322`	`1323`	`AWS_EVENT_STREAM_RPC_MESSAGE_TYPE_APPLICATION_MESSAGE,`
`1323`	`1324`	`AWS_EVENT_STREAM_RPC_MESSAGE_FLAG_TERMINATE_STREAM,`
`1324`		`- onMessageFlushCallback);`
	`1325`	`+ onMessageFlushCallback).get();`
	`1326`	`+ m_clientContinuation.Close();`
	`1327`	`+ return m_closedPromise.get_future();`
`1325`	`1328`	`}`
`1326`	`1329`	`}`
`1327`	`1330`
Original file line number	Diff line number	Diff line change
`@@ -10,6 +10,7 @@ namespace Awstest`
`10`	`10`	`{`
`11`	`11`	`m_echoTestRpcServiceModel.AssignModelNameToErrorResponse(`
`12`	`12`	`Aws::Crt::String("awstest#ServiceError"), ServiceError::s_allocateFromPayload);`
	`13`	`+ aws_event_stream_library_init(m_allocator);`
`13`	`14`	`}`
`14`	`15`
`15`	`16`	`std::future<RpcError> EchoTestRpcClient::Connect(`
`@@ -21,7 +22,7 @@ namespace Awstest`
`21`	`22`
`22`	`23`	`std::future<RpcError> EchoTestRpcClient::Close() noexcept { return m_connection.Close(); }`
`23`	`24`
`24`		`- EchoTestRpcClient::~EchoTestRpcClient() noexcept { Close().wait(); }`
	`25`	`+ EchoTestRpcClient::~EchoTestRpcClient() noexcept { Close().wait(); aws_event_stream_library_clean_up(); }`
`25`	`26`
`26`	`27`	`GetAllProductsOperation EchoTestRpcClient::NewGetAllProducts() noexcept`
`27`	`28`	`{`