Verify behavior works

Author: yourslab

eventstream_rpc/source/EventStreamClient.cpp

@@ -1226,7 +1226,7 @@ namespace Aws
                     /* Missing required content type header. */
                     errorCode = EVENT_STREAM_RPC_UNMAPPED_DATA;
                 }
-                else if (contentHeader == nullptr && contentHeader->GetValueAsString(contentType) && contentType != CONTENT_TYPE_APPLICATION_JSON)
+                else if (contentHeader->GetValueAsString(contentType) && contentType != CONTENT_TYPE_APPLICATION_JSON)
                 {
                     errorCode = EVENT_STREAM_RPC_UNSUPPORTED_CONTENT_TYPE;
                 }
@@ -1309,23 +1309,15 @@ namespace Aws
 
         std::future<RpcError> ClientOperation::Close(OnMessageFlushCallback onMessageFlushCallback) noexcept
         {
-            if (m_clientContinuation.IsClosed())
-            {
-                std::promise<RpcError> alreadyClosedPromise;
-                alreadyClosedPromise.set_value({EVENT_STREAM_RPC_CONTINUATION_CLOSED, 0});
-                return alreadyClosedPromise.get_future();
-            }
-            else
-            {
-                auto rpcError = m_clientContinuation.SendMessage(
+            std::promise<RpcError> closedPromise;
+            closedPromise.set_value(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).get();
-                m_clientContinuation.Close();
-                return m_closedPromise.get_future();
-            }
+                    onMessageFlushCallback).get());
+            m_clientContinuation.Close();
+            return closedPromise.get_future();
         }
 
         void OperationError::s_customDeleter(OperationError *shape) noexcept

eventstream_rpc/tests/EchoTestRpcClient.cpp

@@ -10,7 +10,6 @@ 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(
@@ -22,7 +21,7 @@ namespace Awstest
 
     std::future<RpcError> EchoTestRpcClient::Close() noexcept { return m_connection.Close(); }
 
-    EchoTestRpcClient::~EchoTestRpcClient() noexcept { Close().wait(); aws_event_stream_library_clean_up(); }
+    EchoTestRpcClient::~EchoTestRpcClient() noexcept { Close().wait(); }
 
     GetAllProductsOperation EchoTestRpcClient::NewGetAllProducts() noexcept
     {

eventstream_rpc/tests/EventStreamClientTest.cpp

@@ -196,7 +196,6 @@ static int s_TestEchoOperation(struct aws_allocator *allocator, void *ctx)
         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();
@@ -212,7 +211,6 @@ static int s_TestEchoOperation(struct aws_allocator *allocator, void *ctx)
             auto response = result.GetOperationResponse();
             ASSERT_NOT_NULL(response);
             ASSERT_TRUE(response->GetMessage().value().GetStringMessage().value() == expectedMessage);
-            //aws_event_stream_library_clean_up();
         }
     }