[ServiceBus] Pyamqp Changes from EH in SB (#29499)

* frame fix sync

* frame fix async

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

@@ -10,7 +10,7 @@
 from urllib.parse import urlparse
 import socket
 from ssl import SSLError
-from typing import Any, Tuple, Optional, NamedTuple, Union, cast
+from typing import Any, Dict, Tuple, Optional, NamedTuple, Union, cast
 
 from ._transport import Transport
 from .sasl import SASLTransport, SASLWithWebSocket
@@ -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: Optional[Dict[str, str]] = None
 
         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/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_connection_async.py

@@ -11,7 +11,7 @@
 import socket
 from ssl import SSLError
 import asyncio
-from typing import Any, Tuple, Optional, NamedTuple, Union, cast
+from typing import Any, Dict, Tuple, Optional, NamedTuple, Union, cast
 
 from ._transport_async import AsyncTransport
 from ._sasl_async import SASLTransport, SASLWithWebSocket
@@ -136,6 +136,8 @@ def __init__(self, endpoint, **kwargs):  # pylint:disable=too-many-statements
             "properties", None
         )  # type: Optional[Dict[str, str]]
 
+        self._remote_properties: Optional[Dict[str, str]] = None
+
         self._allow_pipelined_open = kwargs.pop(
             "allow_pipelined_open", True
         )  # type: bool
@@ -431,6 +433,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/servicebus/azure-servicebus/azure/servicebus/_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,7 @@ 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/servicebus/azure-servicebus/azure/servicebus/_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]
         if self.state == SessionState.BEGIN_SENT:
             self.remote_channel = frame[0]  # remote_channel
             await self._set_state(SessionState.MAPPED)

sdk/servicebus/azure-servicebus/azure/servicebus/_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,7 @@ 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:
             self._set_state(LinkState.ATTACH_RCVD)
         elif self.state == LinkState.ATTACH_SENT:

sdk/servicebus/azure-servicebus/azure/servicebus/_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]
         if self.state == SessionState.BEGIN_SENT:
             self.remote_channel = frame[0]  # remote_channel
             self._set_state(SessionState.MAPPED)