Multithreaded support for Eventhubs management family of APIs

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

@@ -6,10 +6,23 @@
 
 ### Breaking Changes
 
+- Claims Based Security authentication now longer throws a `std::runtime_error`, and instead follows the pattern of the rest of the AMQP library and returns an error.
+- Authentication now throws `Azure::Core::Credentials::AuthenticationException` instead of `std::runtime_error`.
+- Added `Cancelled` status to `CbsOperationResult` and `ManagementOperationStatus`.
+
 ### Bugs Fixed
 
+- [[#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
 
+- `std::ostream` inserter for message body no longer prints the body of the message.
+- Tidied up the output of the `AmqpMessage` `std::ostream` inserter.
+- Added several `std::ostream` inserters.
+- Pass numeric values to `std::ostream` inserters by value not by reference.
+
 ## 1.0.0-beta.7 (2024-02-02)
 
 ### Features Added

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

@@ -17,8 +17,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     Error,
     Failed,
     InstanceClosed,
+    Cancelled,
   };
-  std::ostream& operator<<(std::ostream& os, CbsOperationResult const& operationResult);
+  std::ostream& operator<<(std::ostream& os, CbsOperationResult operationResult);
 
   enum class CbsOpenResult
   {
@@ -27,7 +28,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/management.hpp

@@ -26,6 +26,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     Error,
     FailedBadStatus,
     InstanceClosed,
+    Cancelled,
   };
 
   enum class ManagementOpenStatus

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

@@ -31,6 +31,8 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     Timeout,
     Cancelled,
   };
+  std::ostream& operator<<(std::ostream& stream, MessageSendStatus status);
+
   enum class MessageSenderState
   {
     Invalid,
@@ -40,7 +42,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

@@ -76,6 +76,16 @@ namespace Azure { namespace Core { namespace Amqp { namespace Models {
     Unknown,
   };
 
+  /**
+   * @brief ostream insertion operator for AmqpValueType.
+   *
+   * @param os - stream to insert to.
+   * @param value - value to insert.
+   *
+   * @returns the input ostream.
+   */
+  std::ostream& operator<<(std::ostream& os, AmqpValueType value);
+
   class AmqpArray;
   class AmqpMap;
   class AmqpList;

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

@@ -102,13 +102,6 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
         message,
         context);
     if (result.Status != ManagementOperationStatus::Ok)
-    {
-      throw std::runtime_error(
-          "Could not authenticate to client. Error Status: " + std::to_string(result.StatusCode)
-          + " condition: " + result.Error.Condition.ToString()
-          + " reason: " + result.Error.Description);
-    }
-    else
     {
       CbsOperationResult cbsResult;
       switch (result.Status)
@@ -128,16 +121,23 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
         case ManagementOperationStatus::InstanceClosed:
           cbsResult = CbsOperationResult::InstanceClosed;
           break;
+        case ManagementOperationStatus::Cancelled:
+          cbsResult = CbsOperationResult::Cancelled;
+          break;
         default:
           throw std::runtime_error("Unknown management operation status.");
       }
       Log::Stream(Logger::Level::Informational)
           << "CBS PutToken result: " << cbsResult << " status code: " << result.StatusCode
-          << " Error: " << result.Error.Description << ".";
+          << " Error: " << result.Error << ".";
       return std::make_tuple(cbsResult, result.StatusCode, result.Error.Description);
     }
+    else
+    {
+      return std::make_tuple(CbsOperationResult::Ok, result.StatusCode, result.Error.Description);
+    }
   }
-  std::ostream& operator<<(std::ostream& os, CbsOperationResult const& operationResult)
+  std::ostream& operator<<(std::ostream& os, CbsOperationResult operationResult)
   {
     switch (operationResult)
     {
@@ -156,14 +156,14 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
       case CbsOperationResult::InstanceClosed:
         os << "InstanceClosed";
         break;
-      default:
-        os << "Unknown CbsOperationResult."
-           << static_cast<std::underlying_type<CbsOperationResult>::type>(operationResult);
+      case CbsOperationResult::Cancelled:
+        os << "Cancelled";
+        break;
     }
     return os;
   }
 
-  std::ostream& operator<<(std::ostream& os, CbsOpenResult const& openResult)
+  std::ostream& operator<<(std::ostream& os, CbsOpenResult openResult)
   {
     switch (openResult)
     {
@@ -179,9 +179,6 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
       case CbsOpenResult::Cancelled:
         os << "Cancelled";
         break;
-      default:
-        os << "Unknown CbsOpenResult."
-           << static_cast<std::underlying_type<CbsOpenResult>::type>(openResult);
     }
     return os;
   }

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

@@ -677,7 +677,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
             context);
         if (std::get<0>(result) != CbsOperationResult::Ok)
         {
-          throw std::runtime_error("Could not put Claims Based Security token.");
+          throw Azure::Core::Credentials::AuthenticationException(
+              "Could not authenticate client. Error Status: " + std::to_string(std::get<1>(result))
+              + " reason: " + std::get<2>(result));
         }
         Log::Stream(Logger::Level::Verbose) << "Close CBS object";
         claimsBasedSecurity->Close(context);
@@ -690,7 +692,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
         m_tokenStore.emplace(audienceUrl, accessToken);
         return accessToken;
       }
-      catch (std::runtime_error const&)
+      catch (...)
       {
         // Ensure that the claims based security object is closed before we leave this scope.
         claimsBasedSecurity->Close(context);

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)
     {
@@ -209,7 +209,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
       Azure::Core::_internal::AzureNoReturnPath(
           "Destroying link while link detach subscription is still active.");
     }
-
+    auto lock{m_session->GetConnection()->Lock()};
     if (m_link)
     {
       link_destroy(m_link);
@@ -512,19 +512,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
   {
     {
       auto lock{m_session->GetConnection()->Lock()};
-      if (m_eventHandler)
+      if (link_attach(m_link, OnTransferReceivedFn, OnLinkStateChangedFn, OnLinkFlowOnFn, this))
       {
-        if (link_attach(m_link, OnTransferReceivedFn, OnLinkStateChangedFn, OnLinkFlowOnFn, this))
-        {
-          throw std::runtime_error("Could not set attach properties.");
-        }
-      }
-      else
-      {
-        if (link_attach(m_link, nullptr, nullptr, nullptr, this))
-        {
-          throw std::runtime_error("Could not set attach properties.");
-        }
+        throw std::runtime_error("Could not set attach properties.");
       }
     }
     // Mark the connection as async so that we can use the async APIs.