Update the interface for OnStreamError to support the base error type and decouple RpcError

Author: yourslab

eventstream_rpc/source/EventStreamClient.cpp

@@ -839,9 +839,8 @@ namespace Aws
         {
             if (m_callbackData)
             {
-                m_callbackData->callbackMutex.lock();
+                const std::lock_guard<std::mutex> lock(m_callbackData->callbackMutex);
                 m_callbackData->continuationDestroyed = true;
-                m_callbackData->callbackMutex.unlock();
             }
         }
 
@@ -1067,8 +1066,8 @@ 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_expectedCloses(0), m_streamClosedCalled(false)
+              m_streamHandler(streamHandler), m_clientContinuation(connection.NewStream(*this)), m_expectedCloses(0),
+              m_streamClosedCalled(false)
         {
         }
 
@@ -1084,7 +1083,7 @@ namespace Aws
         {
             Close().wait();
             std::unique_lock<std::mutex> lock(m_continuationMutex);
-            m_closeReady.wait(lock, [this]{return m_expectedCloses.load() == 0;});
+            m_closeReady.wait(lock, [this] { return m_expectedCloses.load() == 0; });
         }
 
         TaggedResult::TaggedResult(Crt::ScopedResource<AbstractShapeBase> operationResponse) noexcept
@@ -1431,7 +1430,6 @@ namespace Aws
             /* Promises must be reset in case the client would like to send a subsequent request with the same
              * `ClientOperation`. */
             m_initialResponsePromise = {};
-            //m_closedPromise = {};
             {
                 const std::lock_guard<std::mutex> lock(m_continuationMutex);
                 m_resultReceived = false;

eventstream_rpc/tests/EchoTestRpcModel.cpp

@@ -1065,13 +1065,19 @@ namespace Awstest
         Aws::Crt::ScopedResource<OperationError> operationError,
         RpcError rpcError)
     {
-        if (operationError == nullptr)
-            return OnStreamError(rpcError);
-        if (operationError->GetModelName() == Aws::Crt::String("awstest#ServiceError"))
+        bool streamShouldTerminate = false;
+        if (operationError == nullptr && rpcError.baseStatus != EVENT_STREAM_RPC_SUCCESS)
         {
-            return OnStreamError(static_cast<ServiceError *>(operationError.get()), rpcError);
+            streamShouldTerminate = OnStreamError(rpcError);
         }
-        return true;
+        if (operationError != nullptr && operationError->GetModelName() == Aws::Crt::String("awstest#ServiceError") &&
+            !streamShouldTerminate)
+        {
+            streamShouldTerminate = OnStreamError(static_cast<ServiceError *>(operationError.get()));
+        }
+        if (operationError != nullptr && !streamShouldTerminate)
+            streamShouldTerminate = OnStreamError(operationError.get());
+        return streamShouldTerminate;
     }
 
     CauseStreamServiceToErrorOperationContext::CauseStreamServiceToErrorOperationContext(
@@ -1151,9 +1157,14 @@ namespace Awstest
         Aws::Crt::ScopedResource<OperationError> operationError,
         RpcError rpcError)
     {
-        if (operationError == nullptr)
-            return OnStreamError(rpcError);
-        return true;
+        bool streamShouldTerminate = false;
+        if (operationError == nullptr && rpcError.baseStatus != EVENT_STREAM_RPC_SUCCESS)
+        {
+            streamShouldTerminate = OnStreamError(rpcError);
+        }
+        if (operationError != nullptr && !streamShouldTerminate)
+            streamShouldTerminate = OnStreamError(operationError.get());
+        return streamShouldTerminate;
     }
 
     EchoStreamMessagesOperationContext::EchoStreamMessagesOperationContext(

eventstream_rpc/tests/EventStreamClientTest.cpp

@@ -504,14 +504,11 @@ static int s_TestStressClient(struct aws_allocator *allocator, void *ctx)
             auto echoMessage = client.NewEchoMessage();
             messageData.SetStringMessage(expectedMessage);
             echoMessageRequest.SetMessage(messageData);
-            auto requestFuture = echoMessage.Activate(echoMessageRequest, s_onMessageFlush);
-            std::future_status status = requestFuture.wait_for(std::chrono::seconds(1));
-            if (status != std::future_status::ready)
-            {
-                return AWS_OP_SUCCESS;
-            }
+            auto requestStatus = echoMessage.Activate(echoMessageRequest, s_onMessageFlush).get();
             auto resultFuture = echoMessage.GetResult();
-            status = resultFuture.wait_for(std::chrono::seconds(1));
+            /* The response may never arrive depending on how many ongoing requests are made
+             * so in case of timeout, assume success. */
+            std::future_status status = resultFuture.wait_for(std::chrono::seconds(5));
             if (status != std::future_status::ready)
             {
                 return AWS_OP_SUCCESS;
@@ -525,7 +522,7 @@ static int s_TestStressClient(struct aws_allocator *allocator, void *ctx)
             return AWS_OP_SUCCESS;
         };
 
-        for (int i = 0; i < 10000; i++)
+        for (int i = 0; i < 1000; i++)
             threadPool.AddTask(invokeOperation);
 
         threadPool.BlockUntilTasksFinish();

eventstream_rpc/tests/include/awstest/EchoTestRpcClient.h

@@ -24,6 +24,13 @@ namespace Awstest
         EchoTestRpcClient(
             Aws::Crt::Io::ClientBootstrap &clientBootstrap,
             Aws::Crt::Allocator *allocator = Aws::Crt::g_allocator) noexcept;
+        /**
+         * Connect the client to the server
+         * @param lifecycleHandler An interface that is called upon when lifecycle events relating to the connection
+         * occur.
+         * @param connectionConfig The configuration parameters used for establishing the connection.
+         * @return An `RpcError` that can be used to check whether the connection was established.
+         */
         std::future<RpcError> Connect(
             ConnectionLifecycleHandler &lifecycleHandler,
             const ConnectionConfig &connectionConfig = DefaultConnectionConfig()) noexcept;