[ServiceBus] revert link credit calc to cumulative (#40300)

* [ServiceBus] revert link credit calc to cumulative

* copy sb pyamqp to eh

* fix link credit pyamqp test

Author: swathipil

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

@@ -147,7 +147,7 @@ async def _keep_alive_async(self):
                 elapsed_time = current_time - start_time
                 if elapsed_time >= self._keep_alive_interval:
                     await asyncio.shield(
-                        self._connection.listen(wait=self._socket_timeout, batch=self._link.total_link_credit)
+                        self._connection.listen(wait=self._socket_timeout, batch=self._link.current_link_credit)
                     )
                     start_time = current_time
                 await asyncio.sleep(1)
@@ -733,7 +733,7 @@ async def _client_run_async(self, **kwargs):
         :rtype: bool
         """
         try:
-            if self._link.total_link_credit <= 0:
+            if self._link.current_link_credit<= 0:
                 await self._link.flow(link_credit=self._link_credit)
             await self._connection.listen(wait=self._socket_timeout, **kwargs)
         except ValueError:

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

@@ -96,7 +96,6 @@ def __init__(
         self._on_link_state_change = kwargs.get("on_link_state_change")
         self._on_attach = kwargs.get("on_attach")
         self._error: Optional[AMQPLinkError] = None
-        self.total_link_credit = self.link_credit
 
     async def __aenter__(self) -> "Link":
         await self.attach()
@@ -276,19 +275,6 @@ async def detach(self, close: bool = False, error: Optional[AMQPError] = None) -
             await self._set_state(LinkState.DETACHED)
 
     async def flow(self, *, link_credit: Optional[int] = None, **kwargs) -> None:
-        # Given the desired link credit `link_credit`, the link credit sent via
-        # FlowFrame is calculated as follows: The link credit to flow on the wire
-        # `self.current_link_credit` is the desired link credit `link_credit`
-        # minus the current link credit on the wire `self.total_link_credit`.
-        self.current_link_credit = link_credit - self.total_link_credit if link_credit is not None else self.link_credit
-
-        # If the link credit to flow is greater than 0 (i.e the desired link credit
-        # is greater than the current link credit on the wire), then we will send a
-        # flow to issue more link credit. Otherwise link credit on the wire is sufficient.
-        if self.current_link_credit > 0:
-            # Calculate the total link credit on the wire, by adding the credit
-            # we will flow to the total link credit.
-            self.total_link_credit = (
-                self.current_link_credit + self.total_link_credit if link_credit is not None else self.link_credit
-            )
-            await self._outgoing_flow(**kwargs)
+        # Reset link credit to the default and flow
+        self.current_link_credit = link_credit if link_credit is not None else self.link_credit
+        await self._outgoing_flow(**kwargs)

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

@@ -64,7 +64,6 @@ async def _incoming_transfer(self, frame):
         if not frame[5]:
             self.delivery_count += 1
             self.current_link_credit -= 1
-            self.total_link_credit -= 1
         if self.received_delivery_id is not None:
             self._first_frame = frame
         if not self.received_delivery_id and not self._received_payload:

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

@@ -237,7 +237,7 @@ def _keep_alive(self):
                 current_time = time.time()
                 elapsed_time = current_time - start_time
                 if elapsed_time >= self._keep_alive_interval:
-                    self._connection.listen(wait=self._socket_timeout, batch=self._link.total_link_credit)
+                    self._connection.listen(wait=self._socket_timeout, batch=self._link.current_link_credit)
                     start_time = current_time
                 time.sleep(1)
         except Exception as e:  # pylint: disable=broad-except
@@ -852,7 +852,7 @@ def _client_run(self, **kwargs):
         :rtype: bool
         """
         try:
-            if self._link.total_link_credit <= 0:
+            if self._link.current_link_credit <= 0:
                 self._link.flow(link_credit=self._link_credit)
             self._connection.listen(wait=self._socket_timeout, **kwargs)
         except ValueError:

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

@@ -92,7 +92,6 @@ def __init__(
         self._on_link_state_change = kwargs.get("on_link_state_change")
         self._on_attach = kwargs.get("on_attach")
         self._error: Optional[AMQPLinkError] = None
-        self.total_link_credit = self.link_credit
 
     def __enter__(self) -> "Link":
         self.attach()
@@ -273,18 +272,5 @@ def detach(self, close: bool = False, error: Optional[AMQPError] = None) -> None
             self._set_state(LinkState.DETACHED)
 
     def flow(self, *, link_credit: Optional[int] = None, **kwargs: Any) -> None:
-        # Given the desired link credit `link_credit`, the link credit sent via
-        # FlowFrame is calculated as follows: The link credit to flow on the wire
-        # `self.current_link_credit` is the desired link credit
-        # `link_credit` minus the current link credit on the wire `self.total_link_credit`.
-        self.current_link_credit = link_credit - self.total_link_credit if link_credit is not None else self.link_credit
-
-        # If the link credit to flow is greater than 0 (i.e the desired link credit is greater than
-        # the current link credit on the wire), then we will send a flow to issue more link credit.
-        # Otherwise link credit on the wire is sufficient.
-        if self.current_link_credit > 0:
-            # Calculate the total link credit on the wire, by adding the credit we will flow to the total link credit.
-            self.total_link_credit = (
-                self.current_link_credit + self.total_link_credit if link_credit is not None else self.link_credit
-            )
-            self._outgoing_flow(**kwargs)
+        self.current_link_credit = link_credit if link_credit is not None else self.link_credit
+        self._outgoing_flow(**kwargs)

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

@@ -60,7 +60,6 @@ def _incoming_transfer(self, frame):
         # If more is false --> this is the last frame of the message
         if not frame[5]:
             self.current_link_credit -= 1
-            self.total_link_credit -= 1
             self.delivery_count += 1
         self.received_delivery_id = frame[1]  # delivery_id
         if self.received_delivery_id is not None:

sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_link.py

@@ -131,8 +131,6 @@ def mock_outgoing():
     )
 
     link._outgoing_flow = mock_outgoing
-    link.total_link_credit = 0  # Set the total link credit to 0 to start, no credit on the wire
-
     link.flow(link_credit=100)  # Send a flow frame with desired link credit of 100
 
     # frame: handle, delivery_id, delivery_tag, message_format, settled, more, rcv_settle_mode, state, resume, aborted, batchable, payload
@@ -142,20 +140,15 @@ def mock_outgoing():
 
     link._incoming_transfer(transfer_frame_one)
     assert link.current_link_credit == 99
-    assert link.total_link_credit == 99
 
     # Only received 1 transfer frame per receive call, we set desired link credit again
     # this will send a flow of 1
     link.flow(link_credit=100)
-    assert link.current_link_credit == 1
-    assert link.total_link_credit == 100
-
+    assert link.current_link_credit == 100
     link._incoming_transfer(transfer_frame_two)
-    assert link.current_link_credit == 0
-    assert link.total_link_credit == 99
+    assert link.current_link_credit == 99
     link._incoming_transfer(transfer_frame_three)
-    assert link.current_link_credit == -1
-    assert link.total_link_credit == 98
+    assert link.current_link_credit == 98
 
 @pytest.mark.parametrize(
     "frame",

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_client_async.py

@@ -147,7 +147,7 @@ async def _keep_alive_async(self):
                 elapsed_time = current_time - start_time
                 if elapsed_time >= self._keep_alive_interval:
                     await asyncio.shield(
-                        self._connection.listen(wait=self._socket_timeout, batch=self._link.total_link_credit)
+                        self._connection.listen(wait=self._socket_timeout, batch=self._link.current_link_credit)
                     )
                     start_time = current_time
                 await asyncio.sleep(1)
@@ -733,7 +733,7 @@ async def _client_run_async(self, **kwargs):
         :rtype: bool
         """
         try:
-            if self._link.total_link_credit <= 0:
+            if self._link.current_link_credit<= 0:
                 await self._link.flow(link_credit=self._link_credit)
             await self._connection.listen(wait=self._socket_timeout, **kwargs)
         except ValueError:

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

@@ -96,7 +96,6 @@ def __init__(
         self._on_link_state_change = kwargs.get("on_link_state_change")
         self._on_attach = kwargs.get("on_attach")
         self._error: Optional[AMQPLinkError] = None
-        self.total_link_credit = self.link_credit
 
     async def __aenter__(self) -> "Link":
         await self.attach()
@@ -276,19 +275,6 @@ async def detach(self, close: bool = False, error: Optional[AMQPError] = None) -
             await self._set_state(LinkState.DETACHED)
 
     async def flow(self, *, link_credit: Optional[int] = None, **kwargs) -> None:
-        # Given the desired link credit `link_credit`, the link credit sent via
-        # FlowFrame is calculated as follows: The link credit to flow on the wire
-        # `self.current_link_credit` is the desired link credit `link_credit`
-        # minus the current link credit on the wire `self.total_link_credit`.
-        self.current_link_credit = link_credit - self.total_link_credit if link_credit is not None else self.link_credit
-
-        # If the link credit to flow is greater than 0 (i.e the desired link credit
-        # is greater than the current link credit on the wire), then we will send a
-        # flow to issue more link credit. Otherwise link credit on the wire is sufficient.
-        if self.current_link_credit > 0:
-            # Calculate the total link credit on the wire, by adding the credit
-            # we will flow to the total link credit.
-            self.total_link_credit = (
-                self.current_link_credit + self.total_link_credit if link_credit is not None else self.link_credit
-            )
-            await self._outgoing_flow(**kwargs)
+        # Reset link credit to the default and flow
+        self.current_link_credit = link_credit if link_credit is not None else self.link_credit
+        await self._outgoing_flow(**kwargs)

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_receiver_async.py

@@ -64,7 +64,6 @@ async def _incoming_transfer(self, frame):
         if not frame[5]:
             self.delivery_count += 1
             self.current_link_credit -= 1
-            self.total_link_credit -= 1
         if self.received_delivery_id is not None:
             self._first_frame = frame
         if not self.received_delivery_id and not self._received_payload:

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

@@ -237,7 +237,7 @@ def _keep_alive(self):
                 current_time = time.time()
                 elapsed_time = current_time - start_time
                 if elapsed_time >= self._keep_alive_interval:
-                    self._connection.listen(wait=self._socket_timeout, batch=self._link.total_link_credit)
+                    self._connection.listen(wait=self._socket_timeout, batch=self._link.current_link_credit)
                     start_time = current_time
                 time.sleep(1)
         except Exception as e:  # pylint: disable=broad-except
@@ -852,7 +852,7 @@ def _client_run(self, **kwargs):
         :rtype: bool
         """
         try:
-            if self._link.total_link_credit <= 0:
+            if self._link.current_link_credit <= 0:
                 self._link.flow(link_credit=self._link_credit)
             self._connection.listen(wait=self._socket_timeout, **kwargs)
         except ValueError:

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

@@ -92,7 +92,6 @@ def __init__(
         self._on_link_state_change = kwargs.get("on_link_state_change")
         self._on_attach = kwargs.get("on_attach")
         self._error: Optional[AMQPLinkError] = None
-        self.total_link_credit = self.link_credit
 
     def __enter__(self) -> "Link":
         self.attach()
@@ -273,18 +272,5 @@ def detach(self, close: bool = False, error: Optional[AMQPError] = None) -> None
             self._set_state(LinkState.DETACHED)
 
     def flow(self, *, link_credit: Optional[int] = None, **kwargs: Any) -> None:
-        # Given the desired link credit `link_credit`, the link credit sent via
-        # FlowFrame is calculated as follows: The link credit to flow on the wire
-        # `self.current_link_credit` is the desired link credit
-        # `link_credit` minus the current link credit on the wire `self.total_link_credit`.
-        self.current_link_credit = link_credit - self.total_link_credit if link_credit is not None else self.link_credit
-
-        # If the link credit to flow is greater than 0 (i.e the desired link credit is greater than
-        # the current link credit on the wire), then we will send a flow to issue more link credit.
-        # Otherwise link credit on the wire is sufficient.
-        if self.current_link_credit > 0:
-            # Calculate the total link credit on the wire, by adding the credit we will flow to the total link credit.
-            self.total_link_credit = (
-                self.current_link_credit + self.total_link_credit if link_credit is not None else self.link_credit
-            )
-            self._outgoing_flow(**kwargs)
+        self.current_link_credit = link_credit if link_credit is not None else self.link_credit
+        self._outgoing_flow(**kwargs)

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

@@ -60,7 +60,6 @@ def _incoming_transfer(self, frame):
         # If more is false --> this is the last frame of the message
         if not frame[5]:
             self.current_link_credit -= 1
-            self.total_link_credit -= 1
             self.delivery_count += 1
         self.received_delivery_id = frame[1]  # delivery_id
         if self.received_delivery_id is not None: