Repository: XKNX/xknx

Issue:
Routing flow control: Implement RoutingBusy KNXnet service and act on reception.
We should wait a while before sending next RoutingIndication frames after a RoutingBusy frame was received.
See KNX Standard 3/8/5 Routing §2.3.5 Flow control handling.

As #323 this would benefit from incoming and outgoing telegram queues to be completely separated and the interface (`xknx.io.Routing()`) consuming these queues directly.

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/1031?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#1031](https://codecov.io/gh/XKNX/xknx/pull/1031?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (921bffd) into [main](https://codecov.io/gh/XKNX/xknx/commit/c21bbb948008673727cf26d5c48c8b007b0ccfc0?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (c21bbb9) will **decrease** coverage by `0.01%`.
> The diff coverage is `84.61%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/1031/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://codecov.io/gh/XKNX/xknx/pull/1031?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main    #1031      +/-   ##
==========================================
- Coverage   96.43%   96.42%   -0.02%     
==========================================
  Files         135      136       +1     
  Lines        8215     8246      +31     
==========================================
+ Hits         7922     7951      +29     
- Misses        293      295       +2     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/1031?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/io/tunnel.py](https://codecov.io/gh/XKNX/xknx/pull/1031/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90dW5uZWwucHk=) | `88.72% <ø> (ø)` | |
| [xknx/knxip/routing\_indication.py](https://codecov.io/gh/XKNX/xknx/pull/1031/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9yb3V0aW5nX2luZGljYXRpb24ucHk=) | `96.00% <ø> (ø)` | |
| [xknx/knxip/tunnelling\_ack.py](https://codecov.io/gh/XKNX/xknx/pull/1031/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC90dW5uZWxsaW5nX2Fjay5weQ==) | `96.00% <66.66%> (ø)` | |
| [xknx/io/routing.py](https://codecov.io/gh/XKNX/xknx/pull/1031/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9yb3V0aW5nLnB5) | `80.76% <71.42%> (-0.87%)` | :arrow_down: |
| [xknx/knxip/routing\_busy.py](https://codecov.io/gh/XKNX/xknx/pull/1031/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9yb3V0aW5nX2J1c3kucHk=) | `95.83% <95.83%> (ø)` | |
| [xknx/knxip/\_\_init\_\_.py](https://codecov.io/gh/XKNX/xknx/pull/1031/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9fX2luaXRfXy5weQ==) | `100.00% <100.00%> (ø)` | |
| [xknx/knxip/knxip.py](https://codecov.io/gh/XKNX/xknx/pull/1031/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9rbnhpcC5weQ==) | `100.00% <100.00%> (ø)` | |


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index f33b8928..02f03e95 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -12,6 +12,14 @@ nav_order: 2
 
 - Convenience functions for KNX group communication (`xknx.tools`)
 
+### Routing
+
+- Support flow control for routing
+
+### Protocol
+
+- Add RoutingBusy frame parser
+
 ## 1.0.2 Route-back reconnect 2022-08-31
 
 ### Bugfixes
diff --git a/requirements/production.txt b/requirements/production.txt
index c1e3a466..7a4e9266 100644
--- a/requirements/production.txt
+++ b/requirements/production.txt
@@ -1,2 +1,2 @@
-cryptography==37.0.4
+cryptography==38.0.1
 ifaddr==0.2.0
diff --git a/xknx/io/routing.py b/xknx/io/routing.py
index 15540500..12fb9c84 100644
--- a/xknx/io/routing.py
+++ b/xknx/io/routing.py
@@ -6,8 +6,11 @@ Routing uses UDP Multicast to broadcast and receive KNX/IP messages.
 from __future__ import annotations
 
 import asyncio
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
 import logging
-from typing import TYPE_CHECKING
+import random
+from typing import TYPE_CHECKING, Final
 
 from xknx.core import XknxConnectionState
 from xknx.exceptions import CommunicationError
@@ -17,19 +20,110 @@ from xknx.knxip import (
     CEMIMessageCode,
     KNXIPFrame,
     KNXIPServiceType,
+    RoutingBusy,
     RoutingIndication,
 )
-from xknx.telegram import TelegramDirection
+from xknx.telegram import Telegram, TelegramDirection
 
 from .interface import Interface, TelegramCallbackType
 from .transport import KNXIPTransport, UDPTransport
 
 if TYPE_CHECKING:
-    from xknx.telegram import Telegram
     from xknx.xknx import XKNX
 
 logger = logging.getLogger("xknx.log")
 
+BUSY_DECREMENT_TIME: Final = 0.005  # 5 ms
+BUSY_INCREMENT_COOLDOWN: Final = 0.01  # 10 ms
+BUSY_RANDOM_TIME_FACTOR: Final = 0.05  # 50 ms
+BUSY_SLOWDURATION_TIME_FACTOR: Final = 0.1  # 100 ms
+ROUTING_INDICATION_WAIT_TIME: Final = 0.02  # 20 ms
+
+
+class _RoutingFlowControl:
+    """
+    Class for hanling KNX/IP routing flow control.
+
+    See KNX Specifications 3.8.5 Routing §2.3.5 Flow control handling
+    """
+
+    def __init__(self) -> None:
+        self._last_busy_frame_time: float = 0.0
+        self._last_sent_routing_indication_time: float = 0.0
+        self._loop = asyncio.get_running_loop()
+        self._ready = asyncio.Event()
+        self._ready.set()
+        self._received_busy_frames: int = 0
+        self._timer_task: asyncio.Task[None] | None = None
+        self._wait_start_time: float | None = None
+        self._wait_time_ms: int = 0
+
+    def cancel(self) -> None:
+        """Cancel internal tasks."""
+        if self._timer_task:
+            self._timer_task.cancel()
+
+    @asynccontextmanager
+    async def throttle(self) -> AsyncIterator[None]:
+        """Context manager to wait for ready state and throttle outgoing frames."""
+        # limit RoutingIndication transmission rate according to
+        # KNX Specifications 3.2.6 Communication Medium KNX IP §2.1
+        # simplified version - pause 20 ms after transmit a RoutingIndication
+        elapsed = self._loop.time() - self._last_sent_routing_indication_time
+        if elapsed < ROUTING_INDICATION_WAIT_TIME:
+            await asyncio.sleep(ROUTING_INDICATION_WAIT_TIME - elapsed)
+
+        await self._ready.wait()
+        yield
+        self._last_sent_routing_indication_time = self._loop.time()
+
+    def handle_routing_busy(self, routing_busy: RoutingBusy) -> None:
+        """Handle incoming RoutingBusy."""
+        self._ready.clear()
+        now = self._loop.time()
+        previous_busy_frame_time = self._last_busy_frame_time
+        self._last_busy_frame_time = now
+        if self._wait_start_time is None:
+            logger.warning(
+                "RoutingBusy received: %s",
+                routing_busy,
+            )
+        else:
+            # only apply if we have already received a RoutingBusy frame and are still pausing
+            if (now - previous_busy_frame_time) > BUSY_INCREMENT_COOLDOWN:
+                self._received_busy_frames += 1
+            logger.debug(
+                "RoutingBusy received: %s - %s ms since previous - number %s in moving time window",
+                routing_busy,
+                round((now - previous_busy_frame_time) * 1000),
+                self._received_busy_frames,
+            )
+            # discard frame if wait time is lower than remaining time
+            remaining_ms = (now - self._wait_start_time) * 1000
+            if remaining_ms >= routing_busy.wait_time:
+                return
+        self._wait_time_ms = routing_busy.wait_time
+        self._wait_start_time = now
+
+        if self._timer_task:
+            self._timer_task.cancel()
+        self._timer_task = asyncio.create_task(self._resume_sending())
+
+    async def _resume_sending(self) -> None:
+        """Reset ready flag after wait_time_ms and fade out slowduration."""
+        random_wait_extension = (
+            random.random() * self._received_busy_frames * BUSY_RANDOM_TIME_FACTOR
+        )
+        slowduration = self._received_busy_frames * BUSY_SLOWDURATION_TIME_FACTOR
+        await asyncio.sleep(self._wait_time_ms / 1000 + random_wait_extension)
+
+        self._ready.set()
+        self._wait_start_time = None
+        await asyncio.sleep(slowduration)
+        while self._received_busy_frames > 0:
+            await asyncio.sleep(BUSY_DECREMENT_TIME)
+            self._received_busy_frames -= 1
+
 
 class Routing(Interface):
     """Class for handling KNX/IP routing."""
@@ -50,48 +144,20 @@ class Routing(Interface):
             remote_addr=(self.xknx.multicast_group, self.xknx.multicast_port),
             multicast=True,
         )
-
         self.udp_transport.register_callback(
-            self.response_rec_callback, [KNXIPServiceType.ROUTING_INDICATION]
+            self._handle_frame,
+            [
+                KNXIPServiceType.ROUTING_INDICATION,
+                KNXIPServiceType.ROUTING_BUSY,
+            ],
         )
+        self._flow_control = _RoutingFlowControl()
 
-    def response_rec_callback(
-        self, knxipframe: KNXIPFrame, source: HPAI, _: KNXIPTransport
-    ) -> None:
-        """Verify and handle knxipframe. Callback from internal udp_transport."""
-        if not isinstance(knxipframe.body, RoutingIndication):
-            logger.warning("Service type not implemented: %s", knxipframe)
-        elif knxipframe.body.cemi is None:
-            # ignore unsupported CEMI frame
-            return
-        elif knxipframe.body.cemi.src_addr == self.xknx.own_address:
-            logger.debug("Ignoring own packet")
-        else:
-            # TODO: is cemi message code L_DATA.req or .con valid for routing? if not maybe warn and ignore
-            asyncio.create_task(self.handle_cemi_frame(knxipframe.body.cemi))
-
-    async def handle_cemi_frame(self, cemi: CEMIFrame) -> None:
-        """Handle incoming telegram and send responses if applicable (device management)."""
-        telegram = cemi.telegram
-        telegram.direction = TelegramDirection.INCOMING
-
-        if response_tgs := await self.telegram_received_callback(telegram):
-            for response in response_tgs:
-                await self.send_telegram(response)
-
-    async def send_telegram(self, telegram: "Telegram") -> None:
-        """Send Telegram to routing connected device."""
-        cemi = CEMIFrame.init_from_telegram(
-            telegram=telegram,
-            code=CEMIMessageCode.L_DATA_IND,
-            src_addr=self.xknx.own_address,
-        )
-        routing_indication = RoutingIndication(cemi=cemi)
-        await self.send_knxipframe(KNXIPFrame.init_from_body(routing_indication))
-
-    async def send_knxipframe(self, knxipframe: KNXIPFrame) -> None:
-        """Send KNXIPFrame to connected routing device."""
-        self.udp_transport.send(knxipframe)
+    ####################
+    #
+    # CONNECT DISCONNECT
+    #
+    ####################
 
     async def connect(self) -> bool:
         """Start routing."""
@@ -123,3 +189,64 @@ class Routing(Interface):
         await self.xknx.connection_manager.connection_state_changed(
             XknxConnectionState.DISCONNECTED
         )
+        self._flow_control.cancel()
+
+    ##################
+    #
+    # OUTGOING FRAMES
+    #
+    ##################
+
+    async def send_telegram(self, telegram: Telegram) -> None:
+        """Send Telegram to routing connected device."""
+        cemi = CEMIFrame.init_from_telegram(
+            telegram=telegram,
+            code=CEMIMessageCode.L_DATA_IND,
+            src_addr=self.xknx.own_address,
+        )
+        routing_indication = RoutingIndication(cemi=cemi)
+
+        async with self._flow_control.throttle():
+            self._send_knxipframe(KNXIPFrame.init_from_body(routing_indication))
+
+    def _send_knxipframe(self, knxipframe: KNXIPFrame) -> None:
+        """Send KNXIPFrame to connected routing device."""
+        self.udp_transport.send(knxipframe)
+
+    ##################
+    #
+    # INCOMING FRAMES
+    #
+    ##################
+
+    def _handle_frame(
+        self, knxipframe: KNXIPFrame, source: HPAI, _: KNXIPTransport
+    ) -> None:
+        """Handle incoming KNXIPFrames. Callback from internal udp_transport."""
+        if isinstance(knxipframe.body, RoutingIndication):
+            self._handle_routing_indication(knxipframe.body)
+        elif isinstance(knxipframe.body, RoutingBusy):
+            self._flow_control.handle_routing_busy(knxipframe.body)
+        else:
+            logger.warning("Service not implemented: %s", knxipframe)
+
+    def _handle_routing_indication(self, routing_indication: RoutingIndication) -> None:
+        """Handle incoming RoutingIndication."""
+        if routing_indication.cemi is None:
+            # Don't handle invalid cemi frames (None)
+            return
+        if routing_indication.cemi.src_addr == self.xknx.own_address:
+            logger.debug("Ignoring own packet")
+            return
+
+        # TODO: is cemi message code L_DATA.req or .con valid for routing? if not maybe warn and ignore
+        asyncio.create_task(self.handle_cemi_frame(routing_indication.cemi))
+
+    async def handle_cemi_frame(self, cemi: CEMIFrame) -> None:
+        """Handle incoming telegram and send responses if applicable (device management)."""
+        telegram = cemi.telegram
+        telegram.direction = TelegramDirection.INCOMING
+
+        if response_tgs := await self.telegram_received_callback(telegram):
+            for response in response_tgs:
+                await self.send_telegram(response)
diff --git a/xknx/io/tunnel.py b/xknx/io/tunnel.py
index 9cb2c4c9..bedc0284 100644
--- a/xknx/io/tunnel.py
+++ b/xknx/io/tunnel.py
@@ -77,7 +77,10 @@ class _Tunnel(Interface):
         self._init_transport()
         self.transport.register_callback(
             self._request_received,
-            [KNXIPServiceType.TUNNELLING_REQUEST, KNXIPServiceType.DISCONNECT_REQUEST],
+            [
+                KNXIPServiceType.TUNNELLING_REQUEST,
+                KNXIPServiceType.DISCONNECT_REQUEST,
+            ],
         )
 
     @abstractmethod
diff --git a/xknx/knxip/__init__.py b/xknx/knxip/__init__.py
index 721aebe2..99431368 100644
--- a/xknx/knxip/__init__.py
+++ b/xknx/knxip/__init__.py
@@ -33,6 +33,7 @@ from .knxip_enum import (
     KNXMedium,
     SearchRequestParameterType,
 )
+from .routing_busy import RoutingBusy
 from .routing_indication import RoutingIndication
 from .search_request import SearchRequest
 from .search_request_extended import SearchRequestExtended
@@ -77,6 +78,7 @@ __all__ = [
     "HostProtocol",
     "KNXIPServiceType",
     "KNXMedium",
+    "RoutingBusy",
     "RoutingIndication",
     "SearchRequest",
     "SearchRequestExtended",
diff --git a/xknx/knxip/knxip.py b/xknx/knxip/knxip.py
index 74fd68d8..e4ebf0c7 100644
--- a/xknx/knxip/knxip.py
+++ b/xknx/knxip/knxip.py
@@ -19,6 +19,7 @@ from .disconnect_request import DisconnectRequest
 from .disconnect_response import DisconnectResponse
 from .header import KNXIPHeader
 from .knxip_enum import KNXIPServiceType
+from .routing_busy import RoutingBusy
 from .routing_indication import RoutingIndication
 from .search_request import SearchRequest
 from .search_request_extended import SearchRequestExtended
@@ -79,6 +80,8 @@ class KNXIPFrame:
         # Routing
         elif service_type_ident == KNXIPServiceType.ROUTING_INDICATION:
             body = RoutingIndication()
+        elif service_type_ident == KNXIPServiceType.ROUTING_BUSY:
+            body = RoutingBusy()
         # Secure
         elif service_type_ident == KNXIPServiceType.SECURE_WRAPPER:
             body = SecureWrapper()
diff --git a/xknx/knxip/routing_busy.py b/xknx/knxip/routing_busy.py
new file mode 100644
index 00000000..6b5c70e5
--- /dev/null
+++ b/xknx/knxip/routing_busy.py
@@ -0,0 +1,61 @@
+"""
+Module for Serialization and Deserialization of KNX RoutingBusy frames.
+
+RoutingBusy frames are used for flow control.
+"""
+from __future__ import annotations
+
+from xknx.exceptions import CouldNotParseKNXIP
+
+from .body import KNXIPBody
+from .knxip_enum import KNXIPServiceType
+
+
+class RoutingBusy(KNXIPBody):
+    """Representation of a KNX RoutingBusy frame."""
+
+    SERVICE_TYPE = KNXIPServiceType.ROUTING_BUSY
+    BODY_LENGTH = 6
+
+    def __init__(
+        self,
+        device_state: int = 0,
+        wait_time: int = 100,
+        control_field: int = 0,
+    ):
+        """Initialize RoutingBusy object."""
+        self.device_state = device_state
+        self.wait_time = wait_time
+        self.control_field = control_field
+
+    def calculated_length(self) -> int:
+        """Get length of KNX/IP body."""
+        return RoutingBusy.BODY_LENGTH
+
+    def from_knx(self, raw: bytes) -> int:
+        """Parse/deserialize from KNX/IP raw data."""
+        if raw[0] != RoutingBusy.BODY_LENGTH:  # structure_length field
+            raise CouldNotParseKNXIP("RoutingBusy body has invalid length")
+        if len(raw) != RoutingBusy.BODY_LENGTH:
+            raise CouldNotParseKNXIP("RoutingBusy has wrong length")
+        self.device_state = raw[1]
+        self.wait_time = raw[2] * 256 + raw[3]
+        self.control_field = raw[4] * 256 + raw[5]
+        return RoutingBusy.BODY_LENGTH
+
+    def to_knx(self) -> bytes:
+        """Serialize to KNX/IP raw data."""
+        return (
+            bytes((RoutingBusy.BODY_LENGTH, self.device_state))
+            + self.wait_time.to_bytes(2, "big")
+            + self.control_field.to_bytes(2, "big")
+        )
+
+    def __repr__(self) -> str:
+        """Return object as readable string."""
+        return (
+            "<RoutingBusy "
+            f"device_state={self.device_state} "
+            f"wait_time={self.wait_time} "
+            f"control_field={self.control_field} />"
+        )
diff --git a/xknx/knxip/routing_indication.py b/xknx/knxip/routing_indication.py
index e60fe7eb..fd069671 100644
--- a/xknx/knxip/routing_indication.py
+++ b/xknx/knxip/routing_indication.py
@@ -1,5 +1,5 @@
 """
-Module for Serialization and Deserialization of KNX Routing Indications.
+Module for Serialization and Deserialization of KNX RoutingIndication frames.
 
 Routing indications are used to transport CEMI Messages.
 """
@@ -17,12 +17,12 @@ logger = logging.getLogger("xknx.log")
 
 
 class RoutingIndication(KNXIPBody):
-    """Representation of a KNX Routing Indication."""
+    """Representation of a KNX RoutingIndication frame."""
 
     SERVICE_TYPE = KNXIPServiceType.ROUTING_INDICATION
 
     def __init__(self, cemi: CEMIFrame | None = None):
-        """Initialize SearchRequest object."""
+        """Initialize RoutingIndication object."""
         self.cemi: CEMIFrame | None = (
             cemi if cemi is not None else CEMIFrame(code=CEMIMessageCode.L_DATA_IND)
         )
diff --git a/xknx/knxip/tunnelling_ack.py b/xknx/knxip/tunnelling_ack.py
index 844cb6b3..1f403922 100644
--- a/xknx/knxip/tunnelling_ack.py
+++ b/xknx/knxip/tunnelling_ack.py
@@ -31,10 +31,10 @@ class TunnellingAck(KNXIPBodyResponse):
 
     def from_knx(self, raw: bytes) -> int:
         """Parse/deserialize from KNX/IP raw data."""
-        if raw[0] != TunnellingAck.BODY_LENGTH:
-            raise CouldNotParseKNXIP("connection header wrong length")
+        if raw[0] != TunnellingAck.BODY_LENGTH:  # structure_length field
+            raise CouldNotParseKNXIP("TunnellingAck body has invalid length")
         if len(raw) != TunnellingAck.BODY_LENGTH:
-            raise CouldNotParseKNXIP("connection header wrong length")
+            raise CouldNotParseKNXIP("TunnellingAck body has wrong length")
         self.communication_channel_id = raw[1]
         self.sequence_counter = raw[2]
         self.status_code = ErrorCode(raw[3])


Test patch:
diff --git a/requirements/testing.txt b/requirements/testing.txt
index 778a8411..f96c1619 100644
--- a/requirements/testing.txt
+++ b/requirements/testing.txt
@@ -4,10 +4,10 @@ isort==5.10.1
 flake8==5.0.4
 flake8-isort==4.2.0
 pydocstyle==6.1.1
-pylint==2.15.2
+pylint==2.15.3
 pytest==7.1.3
 pytest-asyncio==0.19.0
 pytest-cov==3.0.0
-setuptools==65.3.0
+setuptools==65.4.0
 tox==3.26.0
 mypy==0.971
diff --git a/test/io_tests/routing_test.py b/test/io_tests/routing_test.py
index b96534bd..e9157655 100644
--- a/test/io_tests/routing_test.py
+++ b/test/io_tests/routing_test.py
@@ -1,11 +1,18 @@
 """Tests for KNX/IP routing connections."""
 import asyncio
-from unittest.mock import AsyncMock, call, patch
+from unittest.mock import AsyncMock, Mock, call, patch
 
 from xknx import XKNX
 from xknx.io import Routing
-from xknx.knxip import CEMIFrame
-from xknx.telegram import Telegram, TelegramDirection
+from xknx.io.routing import (
+    BUSY_DECREMENT_TIME,
+    BUSY_INCREMENT_COOLDOWN,
+    BUSY_SLOWDURATION_TIME_FACTOR,
+    ROUTING_INDICATION_WAIT_TIME,
+    _RoutingFlowControl,
+)
+from xknx.knxip import CEMIFrame, KNXIPFrame, RoutingBusy, RoutingIndication
+from xknx.telegram import Telegram, TelegramDirection, tpci
 
 
 class TestRouting:
@@ -16,15 +23,15 @@ class TestRouting:
         # pylint: disable=attribute-defined-outside-init
         self.xknx = XKNX()
         self.tg_received_mock = AsyncMock()
-        self.routing = Routing(
+
+    @patch("xknx.io.Routing._send_knxipframe")
+    async def test_request_received_callback(self, send_knxipframe_mock):
+        """Test Routing for responding to a transport layer connection."""
+        routing = Routing(
             self.xknx,
             self.tg_received_mock,
             local_ip="192.168.1.1",
         )
-
-    @patch("xknx.io.Routing.send_telegram")
-    async def test_request_received_callback(self, send_telegram_mock):
-        """Test Tunnel for responding to L_DATA.req with confirmation and indication."""
         # L_Data.ind T_Connect from 1.0.250 to 1.0.255 (xknx tunnel endpoint)
         # communication_channel_id: 0x02   sequence_counter: 0x81
         raw_ind = bytes.fromhex("0610 0530 0010 2900b06010fa10ff0080")
@@ -34,19 +41,113 @@ class TestRouting:
         test_telegram = _cemi.telegram
         test_telegram.direction = TelegramDirection.INCOMING
 
-        response_telegram = Telegram(source_address=self.xknx.own_address)
+        response_telegram = Telegram(tpci=tpci.TDisconnect())
+        response_frame = KNXIPFrame.init_from_body(
+            RoutingIndication(
+                cemi=CEMIFrame.init_from_telegram(
+                    telegram=response_telegram, src_addr=self.xknx.own_address
+                )
+            )
+        )
 
         async def tg_received_mock(telegram):
             """Mock for telegram_received_callback."""
             assert telegram == test_telegram
             return [response_telegram]
 
-        self.routing.telegram_received_callback = tg_received_mock
-        self.routing.udp_transport.data_received_callback(
-            raw_ind, ("192.168.1.2", 3671)
-        )
-        await asyncio.sleep(0)
+        routing.telegram_received_callback = tg_received_mock
+        routing.udp_transport.data_received_callback(raw_ind, ("192.168.1.2", 3671))
         await asyncio.sleep(0)
-        assert send_telegram_mock.call_args_list == [
-            call(response_telegram),
+        assert send_knxipframe_mock.call_args_list == [
+            call(response_frame),
         ]
+
+
+class TestFlowControl:
+    """Test class for KNXnet/IP routing flow control."""
+
+    async def test_basic_throttling(self, time_travel):
+        """Test throttling outgoing frames."""
+        flow_control = _RoutingFlowControl()
+        mock = Mock()
+
+        async def test_send():
+            async with flow_control.throttle():
+                mock()
+
+        # first send is called immediately
+        task = asyncio.create_task(test_send())
+        await asyncio.sleep(0)
+        assert mock.call_count == 1
+        assert task.done()
+        mock.reset_mock()
+
+        # second send is throttled
+        task = asyncio.create_task(test_send())
+        await asyncio.sleep(0)
+        assert mock.call_count == 0
+        await time_travel(ROUTING_INDICATION_WAIT_TIME / 4)
+        assert not task.done()
+        await time_travel(ROUTING_INDICATION_WAIT_TIME / 4 * 3)
+        assert task.done()
+        assert mock.call_count == 1
+        mock.reset_mock()
+
+        # later send is called immediately
+        await time_travel(ROUTING_INDICATION_WAIT_TIME)
+        task = asyncio.create_task(test_send())
+        await asyncio.sleep(0)
+        assert mock.call_count == 1
+        assert task.done()
+        mock.reset_mock()
+
+    @patch("random.random")
+    async def test_routing_busy(self, random_mock, time_travel):
+        """Test throttling on received RoutingBusy frame."""
+        flow_control = _RoutingFlowControl()
+        mock = Mock()
+        test_wait_time_ms = 100
+        random_mock.return_value = 0.5
+
+        async def test_send():
+            async with flow_control.throttle():
+                mock()
+
+        test_busy = RoutingBusy(wait_time=test_wait_time_ms)
+
+        flow_control.handle_routing_busy(test_busy)
+        task = asyncio.create_task(test_send())
+        await asyncio.sleep(0)
+        assert mock.call_count == 0
+        await time_travel(test_wait_time_ms / 1000)
+        assert mock.call_count == 1
+        assert task.done()
+        # no slowduration for just 1 RoutingBusy
+        assert flow_control._timer_task.done()
+        mock.reset_mock()
+
+        # multiple RoutingBusy frames
+        flow_control.handle_routing_busy(test_busy)
+        # after cooldown - with different wait times updating wait time for 2x time
+        # not counting one frame due to cooldown time
+        await time_travel(BUSY_INCREMENT_COOLDOWN)
+        flow_control.handle_routing_busy(RoutingBusy(wait_time=test_wait_time_ms // 2))
+        flow_control.handle_routing_busy(RoutingBusy(wait_time=test_wait_time_ms * 2))
+        assert flow_control._received_busy_frames == 1
+        # add second busy frame after cooldown has passed
+        await time_travel(BUSY_INCREMENT_COOLDOWN)
+        flow_control.handle_routing_busy(RoutingBusy(wait_time=test_wait_time_ms // 2))
+        assert flow_control._received_busy_frames == 2
+
+        task = asyncio.create_task(test_send())
+        assert mock.call_count == 0
+        await time_travel(test_wait_time_ms * 2 / 1000 + 2 * 0.5)  # add random time
+        assert mock.call_count == 1
+        assert task.done()
+        # slowduration
+        assert not flow_control._timer_task.done()
+        await time_travel(2 * BUSY_SLOWDURATION_TIME_FACTOR)  # _received_busy_frames 2
+        await time_travel(BUSY_DECREMENT_TIME)  # and decrement time
+        assert not flow_control._timer_task.done()
+        await time_travel(BUSY_DECREMENT_TIME)  # and second decrement time
+        assert flow_control._timer_task.done()
diff --git a/test/knxip_tests/routing_busy_test.py b/test/knxip_tests/routing_busy_test.py
new file mode 100644
index 00000000..62f204b4
--- /dev/null
+++ b/test/knxip_tests/routing_busy_test.py
@@ -0,0 +1,43 @@
+"""Unit test for KNX/IP TunnellingAck objects."""
+import pytest
+
+from xknx.exceptions import CouldNotParseKNXIP
+from xknx.knxip import KNXIPFrame, RoutingBusy
+
+
+class TestKNXIPRoutingBusy:
+    """Test class for KNX/IP RoutingBusy objects."""
+
+    def test_routing_busy(self):
+        """Test parsing and streaming RoutingBusy KNX/IP packet."""
+        raw = bytes(
+            (0x06, 0x10, 0x05, 0x32, 0x00, 0x0C, 0x06, 0x00, 0x00, 0x64, 0x00, 0x00)
+        )
+        knxipframe = KNXIPFrame()
+        knxipframe.from_knx(raw)
+
+        assert isinstance(knxipframe.body, RoutingBusy)
+        assert knxipframe.body.device_state == 0
+        assert knxipframe.body.wait_time == 100
+        assert knxipframe.body.control_field == 0
+
+        routing_busy = RoutingBusy(wait_time=100)
+        knxipframe2 = KNXIPFrame.init_from_body(routing_busy)
+
+        assert knxipframe2.to_knx() == raw
+
+    def test_from_knx_wrong_busy_information(self):
+        """Test parsing and streaming wrong RoutingBusy (wrong length byte)."""
+        raw = bytes(
+            (0x06, 0x10, 0x05, 0x32, 0x00, 0x0C, 0x08, 0x00, 0x00, 0x64, 0x00, 0x00)
+        )
+        knxipframe = KNXIPFrame()
+        with pytest.raises(CouldNotParseKNXIP):
+            knxipframe.from_knx(raw)
+
+    def test_from_knx_wrong_busy_information2(self):
+        """Test parsing and streaming wrong RoutingBusy (wrong length)."""
+        raw = bytes((0x06, 0x10, 0x05, 0x32, 0x00, 0x0C, 0x06, 0x00, 0x00, 0x64, 0x00))
+        knxipframe = KNXIPFrame()
+        with pytest.raises(CouldNotParseKNXIP):
+            knxipframe.from_knx(raw)
diff --git a/test/knxip_tests/tunnelling_ack_test.py b/test/knxip_tests/tunnelling_ack_test.py
index 59998c72..be0ce273 100644
--- a/test/knxip_tests/tunnelling_ack_test.py
+++ b/test/knxip_tests/tunnelling_ack_test.py
@@ -8,7 +8,7 @@ from xknx.knxip import ErrorCode, KNXIPFrame, TunnellingAck
 class TestKNXIPTunnellingAck:
     """Test class for KNX/IP TunnellingAck objects."""
 
-    def test_connect_request(self):
+    def test_tunnelling_ack(self):
         """Test parsing and streaming tunneling ACK KNX/IP packet."""
         raw = bytes((0x06, 0x10, 0x04, 0x21, 0x00, 0x0A, 0x04, 0x2A, 0x17, 0x00))
         knxipframe = KNXIPFrame()