[Eventhub] Fix conflicting link properties (Azure#29101)

* Store remote properties

* Revert error change for now

Author: annatisch

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_connection.py

@@ -139,6 +139,7 @@ def __init__(self, endpoint, **kwargs):  # pylint:disable=too-many-statements
         self._offered_capabilities = None  # type: Optional[str]
         self._desired_capabilities = kwargs.pop("desired_capabilities", None)  # type: Optional[str]
         self._properties = kwargs.pop("properties", None)  # type: Optional[Dict[str, str]]
+        self._remote_properties = None  # type: Optional[Dict[str, str]]
 
         self._allow_pipelined_open = kwargs.pop("allow_pipelined_open", True)  # type: bool
         self._remote_idle_timeout = None  # type: Optional[int]
@@ -416,6 +417,7 @@ def _incoming_open(self, channel, frame):
             )
             return
         self._remote_max_frame_size = frame[2]
+        self._remote_properties = frame[9]
         if self.state == ConnectionState.OPEN_SENT:
             self._set_state(ConnectionState.OPENED)
         elif self.state == ConnectionState.HDR_EXCH:

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_connection_async.py

@@ -134,6 +134,7 @@ def __init__(self, endpoint, **kwargs):  # pylint:disable=too-many-statements
         self._properties = kwargs.pop(
             "properties", None
         )  # type: Optional[Dict[str, str]]
+        self._remote_properties = None  # type: Optional[Dict[str, str]]
 
         self._allow_pipelined_open = kwargs.pop(
             "allow_pipelined_open", True
@@ -430,6 +431,7 @@ async def _incoming_open(self, channel, frame):
             )
             return
         self._remote_max_frame_size = frame[2]
+        self._remote_properties = frame[9]
         if self.state == ConnectionState.OPEN_SENT:
             await self._set_state(ConnectionState.OPENED)
         elif self.state == ConnectionState.HDR_EXCH:

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_link_async.py

@@ -78,6 +78,7 @@ def __init__(self, session, handle, name, role, **kwargs):
         self.remote_max_message_size = None
         self.available = kwargs.pop("available", None)
         self.properties = kwargs.pop("properties", None)
+        self.remote_properties = None
         self.offered_capabilities = None
         self.desired_capabilities = kwargs.pop("desired_capabilities", None)
 
@@ -173,10 +174,8 @@ async def _incoming_attach(self, frame):
         self.remote_handle = frame[1]  # handle
         self.remote_max_message_size = frame[10]  # max_message_size
         self.offered_capabilities = frame[11]  # offered_capabilities
-        if self.properties:
-            self.properties.update(frame[13])  # properties
-        else:
-            self.properties = frame[13]
+        self.remote_properties = frame[13]
+
         if self.state == LinkState.DETACHED:
             await self._set_state(LinkState.ATTACH_RCVD)
         elif self.state == LinkState.ATTACH_SENT:

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_session_async.py

@@ -45,6 +45,7 @@ def __init__(self, connection, channel, **kwargs):
         self.state = SessionState.UNMAPPED
         self.handle_max = kwargs.get("handle_max", 4294967295)
         self.properties = kwargs.pop("properties", None)
+        self.remote_properties = None
         self.channel = channel
         self.remote_channel = None
         self.next_outgoing_id = kwargs.pop("next_outgoing_id", 0)
@@ -137,6 +138,7 @@ async def _incoming_begin(self, frame):
         self.next_incoming_id = frame[1]  # next_outgoing_id
         self.remote_incoming_window = frame[2]  # incoming_window
         self.remote_outgoing_window = frame[3]  # outgoing_window
+        self.remote_properties = frame[7]  # properties
         if self.state == SessionState.BEGIN_SENT:
             self.remote_channel = frame[0]  # remote_channel
             await self._set_state(SessionState.MAPPED)

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/link.py

@@ -76,6 +76,7 @@ def __init__(self, session, handle, name, role, **kwargs):
         self.remote_max_message_size = None
         self.available = kwargs.pop("available", None)
         self.properties = kwargs.pop("properties", None)
+        self.remote_properties = None
         self.offered_capabilities = None
         self.desired_capabilities = kwargs.pop("desired_capabilities", None)
 
@@ -172,10 +173,8 @@ def _incoming_attach(self, frame):
         self.remote_handle = frame[1]  # handle
         self.remote_max_message_size = frame[10]  # max_message_size
         self.offered_capabilities = frame[11]  # offered_capabilities
-        if self.properties:
-            self.properties.update(frame[13])  # properties
-        else:
-            self.properties = frame[13]
+        self.remote_properties = frame[13]  # properties
+
         if self.state == LinkState.DETACHED:
             self._set_state(LinkState.ATTACH_RCVD)
         elif self.state == LinkState.ATTACH_SENT:

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/session.py

@@ -44,6 +44,7 @@ def __init__(self, connection, channel, **kwargs):
         self.state = SessionState.UNMAPPED
         self.handle_max = kwargs.get("handle_max", 4294967295)
         self.properties = kwargs.pop("properties", None)
+        self.remote_properties = None
         self.channel = channel
         self.remote_channel = None
         self.next_outgoing_id = kwargs.pop("next_outgoing_id", 0)
@@ -150,6 +151,7 @@ def _incoming_begin(self, frame):
         self.next_incoming_id = frame[1]  # next_outgoing_id
         self.remote_incoming_window = frame[2]  # incoming_window
         self.remote_outgoing_window = frame[3]  # outgoing_window
+        self.remote_properties = frame[7]  # properties
         if self.state == SessionState.BEGIN_SENT:
             self.remote_channel = frame[0]  # remote_channel
             self._set_state(SessionState.MAPPED)