feat(routing): separate task routing from broker transport

Introduce Flow, TaskiqRouter, route and subscription contracts to move
task dispatch decisions out of brokers while keeping brokers as transport
adapters.
Preserve backward-compatible

Author: GefMar

taskiq/abc/broker.py

@@ -142,7 +142,8 @@ def find_task(self, task_name: str) -> AsyncTaskiqDecoratedTask[Any, Any] | None
 
         It searches task by name in dict of tasks that
         were registered for this broker directly.
-        If it fails, it checks global dict of all available tasks.
+        If it fails, it checks tasks registered in the broker's router,
+        then global dict of all available tasks.
 
         :param task_name: name of a task.
         :returns: found task or None.
@@ -161,8 +162,9 @@ def get_all_tasks(self) -> dict[str, AsyncTaskiqDecoratedTask[Any, Any]]:
         """
         Method to fetch all tasks available in broker.
 
-        This method returns all tasks, globally and locally
-        available in broker. With local tasks having higher priority.
+        This method returns all tasks globally, through the broker's router
+        and locally available in broker. With local tasks having higher
+        priority.
 
         So, if you have two tasks with the same name,
         one registered in global registry and one registered

taskiq/router.py

@@ -55,6 +55,8 @@ def default_broker(self) -> AsyncBroker | None:
 
     @default_broker.setter
     def default_broker(self, broker: AsyncBroker | None) -> None:
+        if broker is not None:
+            self._brokers.resolve(broker)
         self._brokers.default_broker = broker
 
     @property

tests/routing/test_router_core.py

@@ -34,6 +34,15 @@ def test_router_rejects_broker_attached_to_another_router() -> None:
         second_router.set_broker(broker, name="broker")
 
 
+def test_router_default_broker_setter_rejects_foreign_broker() -> None:
+    first_router = TaskiqRouter()
+    second_router = TaskiqRouter()
+    broker = RecordingBroker(router=first_router, broker_name="broker")
+
+    with pytest.raises(ValueError, match="not registered"):
+        second_router.default_broker = broker
+
+
 def test_router_rejects_string_broker_references() -> None:
     router = TaskiqRouter()
     RecordingBroker(router=router, broker_name="broker")