Multithreaded support for Eventhubs management family of APIs

sdk/core/azure-core-amqp/CHANGELOG.md

@@ -6,12 +6,12 @@
 
 ### Breaking Changes
 
-- `ClaimBasedSecurity::PutToken` throws `Azure::Core::Credentials::AuthenticationException` instead of `std::runtime_error`.
+- Claims Based Security authentication now throws `Azure::Core::Credentials::AuthenticationException` instead of `std::runtime_error`.
 
 ### Bugs Fixed
 
-- Fixed [#5284](https://github.com/Azure/azure-sdk-for-cpp/issues/5284).
-- Fixed [#5297](https://github.com/Azure/azure-sdk-for-cpp/issues/5297). Enabled multiple simultaneous `ExecuteOperation` calls.
+- [#5284](https://github.com/Azure/azure-sdk-for-cpp/issues/5284): [azure-identity][azure-messaging-eventhubs] Impossible to catch exception resulting in SIGABRT signal.
+- [#5297](https://github.com/Azure/azure-sdk-for-cpp/issues/5297): Enabled multiple simultaneous `ExecuteOperation` calls.
 - Fixed crash when Link Detach message is received while link is being destroyed.
 
 ### Other Changes

sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/amqp_settle_mode.hpp

@@ -14,13 +14,13 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     Mixed,
   };
 
-  std::ostream& operator<<(std::ostream& os, SenderSettleMode const mode);
+  std::ostream& operator<<(std::ostream& os, SenderSettleMode mode);
 
   enum class ReceiverSettleMode
   {
     First,
     Second,
   };
-  std::ostream& operator<<(std::ostream& os, ReceiverSettleMode const mode);
+  std::ostream& operator<<(std::ostream& os, ReceiverSettleMode mode);
 
 }}}} // namespace Azure::Core::Amqp::_internal

sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/claims_based_security.hpp

@@ -18,7 +18,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     Failed,
     InstanceClosed,
   };
-  std::ostream& operator<<(std::ostream& os, CbsOperationResult const operationResult);
+  std::ostream& operator<<(std::ostream& os, CbsOperationResult operationResult);
 
   enum class CbsOpenResult
   {
@@ -27,7 +27,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     Error,
     Cancelled,
   };
-  std::ostream& operator<<(std::ostream& os, CbsOpenResult const operationResult);
+  std::ostream& operator<<(std::ostream& os, CbsOpenResult operationResult);
 
   enum class CbsTokenType
   {

sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/connection.hpp

@@ -158,7 +158,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     Error,
   };
 
-  std::ostream& operator<<(std::ostream& stream, ConnectionState const value);
+  std::ostream& operator<<(std::ostream& stream, ConnectionState value);
 
   class Connection;
 

sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/link.hpp

@@ -47,7 +47,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     Error,
   };
 
-  std::ostream& operator<<(std::ostream& stream, LinkState const linkState);
+  std::ostream& operator<<(std::ostream& stream, LinkState linkState);
 
   enum class LinkTransferResult
   {

sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/message_receiver.hpp

@@ -33,7 +33,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     Closing,
     Error,
   };
-  std::ostream& operator<<(std::ostream& stream, _internal::MessageReceiverState const state);
+  std::ostream& operator<<(std::ostream& stream, _internal::MessageReceiverState state);
 
   class MessageReceiver;
 

sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/message_sender.hpp

@@ -40,7 +40,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     Closing,
     Error,
   };
-  std::ostream& operator<<(std::ostream& stream, MessageSenderState const state);
+  std::ostream& operator<<(std::ostream& stream, MessageSenderState state);
 
   class MessageSender;
   class MessageSenderEvents {

sdk/core/azure-core-amqp/inc/azure/core/amqp/models/amqp_value.hpp

@@ -84,7 +84,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Models {
    *
    * @returns the input ostream.
    */
-  std::ostream& operator<<(std::ostream& os, AmqpValueType const value);
+  std::ostream& operator<<(std::ostream& os, AmqpValueType value);
 
   class AmqpArray;
   class AmqpMap;

sdk/core/azure-core-amqp/src/amqp/claim_based_security.cpp

@@ -137,7 +137,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
       return std::make_tuple(cbsResult, result.StatusCode, result.Error.Description);
     }
   }
-  std::ostream& operator<<(std::ostream& os, CbsOperationResult const operationResult)
+  std::ostream& operator<<(std::ostream& os, CbsOperationResult operationResult)
   {
     switch (operationResult)
     {
@@ -163,7 +163,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     return os;
   }
 
-  std::ostream& operator<<(std::ostream& os, CbsOpenResult const openResult)
+  std::ostream& operator<<(std::ostream& os, CbsOpenResult openResult)
   {
     switch (openResult)
     {

sdk/core/azure-core-amqp/src/amqp/link.cpp

@@ -126,7 +126,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
   }
 #endif
 
-  std::ostream& operator<<(std::ostream& os, LinkState const linkState)
+  std::ostream& operator<<(std::ostream& os, LinkState linkState)
   {
     switch (linkState)
     {

sdk/core/azure-core-amqp/src/amqp/management.cpp

@@ -75,6 +75,12 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
   _internal::ManagementOpenStatus ManagementClientImpl::Open(Context const& context)
   {
+    std::unique_lock<std::mutex> lock(m_openCloseLock);
+    if (m_isOpen)
+    {
+      throw std::runtime_error("Management object is already open.");
+    }
+
     try
     {
       /** Authentication needs to happen *before* the links are created.
@@ -261,7 +267,13 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
   void ManagementClientImpl::Close(Context const& context)
   {
+    std::unique_lock<std::mutex> lock(m_openCloseLock);
     Log::Stream(Logger::Level::Verbose) << "ManagementClient::Close" << std::endl;
+    if (!m_isOpen)
+    {
+      throw std::runtime_error("Management object is not open.");
+    }
+
     SetState(ManagementState::Closing);
     if (m_messageSender && m_messageSenderOpen)
     {

sdk/core/azure-core-amqp/src/amqp/message_receiver.cpp

@@ -41,7 +41,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
 namespace Azure { namespace Core { namespace Amqp { namespace _internal {
 
-  std::ostream& operator<<(std::ostream& stream, ReceiverSettleMode const settleMode)
+  std::ostream& operator<<(std::ostream& stream, ReceiverSettleMode settleMode)
   {
     switch (settleMode)
     {
@@ -109,7 +109,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
   }
 
   std::string MessageReceiver::GetLinkName() const { return m_impl->GetLinkName(); }
-  std::ostream& operator<<(std::ostream& stream, _internal::MessageReceiverState const state)
+  std::ostream& operator<<(std::ostream& stream, _internal::MessageReceiverState state)
   {
     switch (state)
     {
@@ -399,7 +399,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
       "MESSAGE_RECEIVER_STATE_ERROR",
   };
 
-  std::ostream& operator<<(std::ostream& stream, MESSAGE_RECEIVER_STATE const state)
+  std::ostream& operator<<(std::ostream& stream, MESSAGE_RECEIVER_STATE state)
   {
     if (state < sizeof(MESSAGE_RECEIVER_STATEStrings) / sizeof(MESSAGE_RECEIVER_STATEStrings[0]))
     {

sdk/core/azure-core-amqp/src/amqp/message_sender.cpp

@@ -37,7 +37,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 }}}} // namespace Azure::Core::Amqp::_detail
 
 namespace Azure { namespace Core { namespace Amqp { namespace _internal {
-  std::ostream& operator<<(std::ostream& stream, SenderSettleMode const settleMode)
+  std::ostream& operator<<(std::ostream& stream, SenderSettleMode settleMode)
   {
     switch (settleMode)
     {
@@ -66,7 +66,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
   std::uint64_t MessageSender::GetMaxMessageSize() const { return m_impl->GetMaxMessageSize(); }
   std::string MessageSender::GetLinkName() const { return m_impl->GetLinkName(); }
   MessageSender::~MessageSender() noexcept {}
-  std::ostream& operator<<(std::ostream& stream, _internal::MessageSenderState const state)
+  std::ostream& operator<<(std::ostream& stream, _internal::MessageSenderState state)
   {
     switch (state)
     {

sdk/core/azure-core-amqp/src/amqp/private/management_impl.hpp

@@ -82,6 +82,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     std::shared_ptr<MessageSenderImpl> m_messageSender;
     std::shared_ptr<MessageReceiverImpl> m_messageReceiver;
     ManagementState m_state = ManagementState::Idle;
+    std::mutex m_openCloseLock;
     bool m_isOpen{false};
     bool m_messageSenderOpen{false};
     bool m_messageReceiverOpen{false};

sdk/core/azure-core-amqp/src/models/amqp_value.cpp

@@ -46,7 +46,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 namespace Azure { namespace Core { namespace Amqp { namespace Models {
   namespace _detail {
 
-    std::ostream& operator<<(std::ostream& os, AMQP_TYPE const value)
+    std::ostream& operator<<(std::ostream& os, AMQP_TYPE value)
     {
       switch (value)
       {
@@ -146,7 +146,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Models {
     }
   } // namespace _detail
 
-  std::ostream& operator<<(std::ostream& os, AmqpValueType const value)
+  std::ostream& operator<<(std::ostream& os, AmqpValueType value)
   {
 
     switch (value)

sdk/core/azure-core-amqp/src/models/private/value_impl.hpp

@@ -58,7 +58,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Models { namespace
   private:
     UniqueAmqpValueHandle m_value;
   };
-  std::ostream& operator<<(std::ostream& os, AMQP_TYPE const value);
+  std::ostream& operator<<(std::ostream& os, AMQP_TYPE value);
   std::ostream& operator<<(std::ostream& os, AMQP_VALUE const value);
 
 }}}}} // namespace Azure::Core::Amqp::Models::_detail

sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md

@@ -12,7 +12,6 @@
 
 ### Other Changes
 
-
 ## 1.0.0-beta.6 (2024-02-06)
 
 ### Breaking Changes

sdk/eventhubs/azure-messaging-eventhubs/test/ut/processor_test.cpp

@@ -121,7 +121,6 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace Test {
       producerOptions.Name = "Producer for LoadBalancerTest";
       ProducerClient producerClient{connectionString, eventHubName, producerOptions};
 
-#if TRUE
       std::thread processEventsThread([&]() {
         std::set<std::string> partitionsAcquired;
         std::vector<std::thread> processEventsThreads;
@@ -160,43 +159,6 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace Test {
       processor.Run(context);
 
       processEventsThread.join();
-#else
-      // Run the processor on a background thread and the test on the foreground.
-      processor.Start(context);
-
-      std::set<std::string> partitionsAcquired;
-      std::vector<std::thread> processEventsThreads;
-      // When we exit the process thread, cancel the context to unblock the processor.
-      //      scope_guard onExit([&context] { context.Cancel(); });
-
-      WaitGroup waitGroup;
-      for (auto const& partitionId : eventHubProperties.PartitionIds)
-      {
-        std::shared_ptr<ProcessorPartitionClient> partitionClient
-            = processor.NextPartitionClient(context);
-        waitGroup.AddWaiter();
-        ASSERT_EQ(partitionsAcquired.find(partitionId), partitionsAcquired.end())
-            << "No previous client for " << partitionClient->PartitionId();
-        processEventsThreads.push_back(
-            std::thread([&waitGroup, &producerClient, partitionClient, &context, this] {
-              scope_guard onExit([&] { waitGroup.CompleteWaiter(); });
-              ProcessEventsForLoadBalancerTest(producerClient, partitionClient, context);
-            }));
-      }
-      // Block until all the events have been processed.
-      waitGroup.Wait();
-
-      // And wait until all the threads have completed.
-      for (auto& thread : processEventsThreads)
-      {
-        if (thread.joinable())
-        {
-          thread.join();
-        }
-      }
-      // Stop the processor, we're done with the test.
-      processor.Stop();
-#endif
     }
 
     void TestWithLoadBalancerSingleThreaded(Models::ProcessorStrategy processorStrategy)