AMQP Session related fixes. (#5299)

* Implemented connection handle table to fix half closed session problem

* Correctly handle session begin after connection is open

* Split out amqp settle mode to separate header; added test coverage for AMQP Transfer and Detach performatives

* Use absolute path for settle mode header

Author: LarryOsterman

.clang-format

@@ -47,7 +47,9 @@ IncludeCategories:
    Priority: 90
  - Regex: '<Azure/.*>'
    Priority: 50
- - Regex: '^<azure_uamqp_c.*/(amqp_definitions_sequence_no|amqp_definitions_milliseconds|amqp_definitions_terminus_expiry_policy|amqp_definitions_terminus_durability|amqp_definitions_fields).h>$'
+ - Regex: '^<azure_uamqp_c.*/(amqp_definitions_fields).h>$'
+   Priority: 59
+ - Regex: '^<azure_uamqp_c.*/(amqp_definitions_sequence_no|amqp_definitions_milliseconds|amqp_definitions_terminus_expiry_policy|amqp_definitions_terminus_durability|amqp_definitions_number|amqp_definitions_error|amqp_definitions_handle).h>$'
    Priority: 60
  - Regex: '<azure_.*>'
    Priority: 70

.vscode/cspell.json

@@ -255,9 +255,16 @@
       "filename": "CMakePresets.json",
       "words": [
         "ASAN",
+        "asan",
         "fsanitize"
       ]
     },
+    {
+      "filename": ".clang-format",
+      "words": [
+        "Dont"
+      ]
+    },
     {
       "filename": "**/eng/pipelines/templates/**/*.yml",
       "words": [

CMakePresets.json

@@ -344,6 +344,16 @@
         "RUN_LONG_UNIT_TESTS": true
       }
     },
+    {
+      "name": "x86-static-release",
+      "displayName": "x86 Release, static",
+      "description": "Windows x86 Release build",
+      "inherits": [
+        "x86-static",
+        "release-build"
+      ]
+
+    },
     {
       "name": "x64-debug-tests",
       "displayName": "x64 Debug With Tests",

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

@@ -12,6 +12,9 @@ The `Close` method on AMQP Message Sender and Message Receiver now blocks until
 
 ### Bugs Fixed
 
+- Fixed uAMQP connection channel so that a channel is released when an END performative is received from the remote node instead of when the END performative is sent to the remote node.
+- Enabled more than one uAMQP session to be created on a single connection.
+
 ### Other Changes
 
 ## 1.0.0-beta.6 (2024-01-11)

sdk/core/azure-core-amqp/CMakeLists.txt

@@ -61,6 +61,7 @@ find_package(azure_c_shared_utility CONFIG REQUIRED)
 set (AZURE_CORE_AMQP_HEADER
     inc/azure/core/amqp.hpp
     inc/azure/core/amqp/dll_import_export.hpp
+    inc/azure/core/amqp/internal/amqp_settle_mode.hpp
     inc/azure/core/amqp/internal/cancellable.hpp
     inc/azure/core/amqp/internal/claims_based_security.hpp
     inc/azure/core/amqp/internal/common/async_operation_queue.hpp
@@ -79,6 +80,8 @@ set (AZURE_CORE_AMQP_HEADER
     inc/azure/core/amqp/internal/models/message_source.hpp
     inc/azure/core/amqp/internal/models/message_target.hpp
     inc/azure/core/amqp/internal/models/messaging_values.hpp
+    inc/azure/core/amqp/internal/models/performatives/amqp_detach.hpp
+    inc/azure/core/amqp/internal/models/performatives/amqp_transfer.hpp
     inc/azure/core/amqp/internal/network/amqp_header_detect_transport.hpp
     inc/azure/core/amqp/internal/network/sasl_transport.hpp
     inc/azure/core/amqp/internal/network/socket_listener.hpp
@@ -112,23 +115,28 @@ set(AZURE_CORE_AMQP_SOURCE
     src/amqp/private/unique_handle.hpp
     src/amqp/session.cpp
     src/common/global_state.cpp
+    src/models/amqp_detach.cpp
     src/models/amqp_error.cpp
     src/models/amqp_header.cpp
     src/models/amqp_message.cpp
     src/models/amqp_properties.cpp
+    src/models/amqp_transfer.cpp
     src/models/amqp_value.cpp
     src/models/message_source.cpp
     src/models/message_target.cpp
     src/models/messaging_values.cpp
     src/models/private/error_impl.hpp
     src/models/private/header_impl.hpp
     src/models/private/message_impl.hpp
+    src/models/private/performatives/detach_impl.hpp
+    src/models/private/performatives/transfer_impl.hpp
     src/models/private/properties_impl.hpp
     src/models/private/source_impl.hpp
     src/models/private/target_impl.hpp
     src/models/private/value_impl.hpp
     src/network/amqp_header_transport.cpp
     src/network/private/transport_impl.hpp
+    src/network/private/transport_impl.hpp
     src/network/sasl_transport.cpp
     src/network/socket_listener.cpp
     src/network/socket_transport.cpp
@@ -137,6 +145,7 @@ set(AZURE_CORE_AMQP_SOURCE
     src/private/package_version.hpp
 )
 
+
 add_library(azure-core-amqp ${AZURE_CORE_AMQP_SOURCE} ${AZURE_CORE_AMQP_HEADER} $<TARGET_OBJECTS:uamqp>)
 
 if (VENDOR_UAMQP)

sdk/core/azure-core-amqp/cspell.json

@@ -12,6 +12,7 @@
   "words": [
     "blang",
     "dowork",
+    "performatives",
     "SASLCLIENTIO",
     "socketio",
     "stringized",

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

@@ -0,0 +1,26 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+#pragma once
+
+#include <iostream>
+
+namespace Azure { namespace Core { namespace Amqp { namespace _internal {
+
+  enum class SenderSettleMode
+  {
+    Unsettled,
+    Settled,
+    Mixed,
+  };
+
+  std::ostream& operator<<(std::ostream& os, SenderSettleMode const& mode);
+
+  enum class ReceiverSettleMode
+  {
+    First,
+    Second,
+  };
+  std::ostream& operator<<(std::ostream& os, ReceiverSettleMode const& mode);
+
+}}}} // namespace Azure::Core::Amqp::_internal

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

@@ -35,6 +35,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
   class TestSocketListenerEvents;
   class LinkSocketListenerEvents;
   class TestLinks_LinkAttachDetach_Test;
+  class TestSessions_MultipleSessionBeginEnd_Test;
   class TestMessages_SenderOpenClose_Test;
   class TestMessages_TestLocalhostVsTls_Test;
   class TestMessages_SenderSendAsync_Test;
@@ -166,7 +167,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
    */
   class ConnectionEvents {
   protected:
-    ~ConnectionEvents(){};
+    virtual ~ConnectionEvents() = default;
 
   public:
     /** @brief Called when the connection state changes.
@@ -181,27 +182,28 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
         ConnectionState oldState)
         = 0;
 
-    /** @brief Called when a new endpoint connects to the connection.
+    /** @brief called when an I/O error has occurred on the connection.
      *
      * @param connection The connection object.
-     * @param endpoint The endpoint that connected.
-     * @return true if the endpoint was accepted, false otherwise.
-     *
-     * @remarks Note that this function should only be overriden if the application is listening
-     * on the connection.
      */
-    virtual bool OnNewEndpoint(Connection const& connection, Endpoint& endpoint)
-    {
-      (void)connection;
-      (void)endpoint;
-      return false;
-    }
+    virtual void OnIOError(Connection const& connection) = 0;
+  };
 
-    /** @brief called when an I/O error has occurred on the connection.
+  class ConnectionEndpointEvents {
+  protected:
+    virtual ~ConnectionEndpointEvents() = default;
+
+  public:
+    /** @brief Called when a new endpoint connects to the connection.
      *
      * @param connection The connection object.
+     * @param endpoint The endpoint that connected.
+     * @return true if the endpoint was accepted, false otherwise.
+     *
+     * @remarks Note that this function should only be overriden if
+     * the application is listening on the connection.
      */
-    virtual void OnIOError(Connection const& connection) = 0;
+    virtual bool OnNewEndpoint(Connection const& connection, Endpoint& endpoint) = 0;
   };
 
   /** @brief Options used to create a connection. */
@@ -299,7 +301,8 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     Connection(
         Network::_internal::Transport const& transport,
         ConnectionOptions const& options,
-        ConnectionEvents* eventHandler = nullptr);
+        ConnectionEvents* eventHandler,
+        ConnectionEndpointEvents* endpointEvents);
 
     /** @brief Destroy an AMQP connection */
     ~Connection();
@@ -457,6 +460,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     friend class Azure::Core::Amqp::Tests::TestConnections_ConnectionAttributes_Test;
     friend class Azure::Core::Amqp::Tests::TestConnections_ConnectionOpenClose_Test;
     friend class Azure::Core::Amqp::Tests::TestConnections_ConnectionListenClose_Test;
+    friend class Azure::Core::Amqp::Tests::TestSessions_MultipleSessionBeginEnd_Test;
     friend class Azure::Core::Amqp::Tests::TestLinks_LinkAttachDetach_Test;
     friend class Azure::Core::Amqp::Tests::TestMessages_SenderOpenClose_Test;
     friend class Azure::Core::Amqp::Tests::TestMessages_TestLocalhostVsTls_Test;

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

@@ -22,6 +22,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
   class LinkEndpointFactory;
 }}}} // namespace Azure::Core::Amqp::_detail
 namespace Azure { namespace Core { namespace Amqp { namespace _internal {
+  class ConnectionEvents;
 
   // An "Endpoint" is an intermediate type used to create sessions in an OnNewSession callback.
   class Endpoint final {
@@ -43,6 +44,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
       return rv;
     }
     friend class _detail::EndpointFactory;
+    friend class _internal::ConnectionEvents;
   };
 
   // A "Link Endpoint" is an intermediate type used to create new Links in an OnLinkAttached
@@ -58,6 +60,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     }
     ~LinkEndpoint(){};
 
+    LINK_ENDPOINT_INSTANCE_TAG* Get() const { return m_endpoint; }
+    std::uint32_t GetHandle() const;
+
   private:
     LINK_ENDPOINT_INSTANCE_TAG* m_endpoint;
 

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

@@ -5,7 +5,9 @@
 
 #include "azure/core/amqp/internal/models/message_source.hpp"
 #include "azure/core/amqp/internal/models/message_target.hpp"
+#include "azure/core/amqp/internal/models/performatives/amqp_transfer.hpp"
 #include "azure/core/amqp/models/amqp_value.hpp"
+#include "azure/core/context.hpp"
 
 #include <chrono>
 #include <memory>
@@ -40,11 +42,13 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     Invalid,
     Detached,
     HalfAttachedAttachSent,
-    HalfAttachAttachReceived,
+    HalfAttachedAttachReceived,
     Attached,
     Error,
   };
 
+  std::ostream& operator<<(std::ostream& stream, LinkState const& linkState);
+
   enum class LinkTransferResult
   {
     Error,
@@ -58,24 +62,62 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     NotDelivered,
     Timeout,
     Cancelled,
+    Invalid
+  };
+
+  class Link;
+
+  class LinkEvents {
+  public:
+    virtual Models::AmqpValue OnTransferReceived(
+#if defined(TESTING_BUILD)
+        Link const& link,
+#else
+        std::shared_ptr<LinkImpl> link,
+#endif
+        Models::_internal::Performatives::AmqpTransfer transfer,
+        uint32_t payloadSize,
+        const unsigned char* payloadBytes)
+        = 0;
+    virtual void OnLinkStateChanged(
+#if defined(TESTING_BUILD)
+        Link const& link,
+#else
+        std::shared_ptr<LinkImpl> link,
+#endif
+        LinkState newLinkState,
+        LinkState previousLinkState)
+        = 0;
+    virtual void OnLinkFlowOn(
+#if defined(TESTING_BUILD)
+        Link const& link
+#else
+        std::shared_ptr<LinkImpl> link
+#endif
+        )
+        = 0;
+    virtual ~LinkEvents() = default;
   };
 
 #if defined(TESTING_BUILD)
+
   class Link final {
   public:
     Link(
         _internal::Session const& session,
         std::string const& name,
         _internal::SessionRole role,
         Models::_internal::MessageSource const& source,
-        Models::_internal::MessageTarget const& target);
+        Models::_internal::MessageTarget const& target,
+        LinkEvents* events = nullptr);
     Link(
         _internal::Session const& session,
         _internal::LinkEndpoint& linkEndpoint,
         std::string const& name,
         _internal::SessionRole role,
         Models::_internal::MessageSource const& source,
-        Models::_internal::MessageTarget const& target);
+        Models::_internal::MessageTarget const& target,
+        LinkEvents* events = nullptr);
     ~Link() noexcept;
 
     Link(Link const&) = default;
@@ -97,9 +139,14 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
     uint64_t GetPeerMaxMessageSize() const;
 
-    void SetAttachProperties(Models::AmqpValue attachProperties);
+    void SetAttachProperties(Models::AmqpValue const& attachProperties);
     void SetMaxLinkCredit(uint32_t maxLinkCredit);
 
+    void SetDesiredCapabilities(Models::AmqpValue const& desiredCapabilities);
+    Models::AmqpValue GetDesiredCapabilities() const;
+
+    void ResetLinkCredit(std::uint32_t linkCredit, bool drain);
+
     std::string GetName() const;
 
     Models::_internal::MessageTarget const& GetTarget() const;
@@ -109,13 +156,18 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
     void Attach();
 
+    std::tuple<uint32_t, LinkDeliverySettleReason, Models::AmqpValue> Transfer(
+        std::vector<uint8_t> const& payload,
+        Azure::Core::Context const& context);
+
     void Detach(
         bool close,
         std::string const& errorCondition,
         std::string const& errorDescription,
-        Models::AmqpValue& info);
+        const Models::AmqpValue& info);
 
   private:
+    friend class LinkImpl;
     Link(std::shared_ptr<LinkImpl> impl) : m_impl{impl} {}
 
     std::shared_ptr<LinkImpl> m_impl;