experiment separate Broker + Router

docs/available-components/brokers.md

@@ -12,6 +12,12 @@ This is a special broker for local development. It uses the same functions to ex
 but all tasks are executed locally in the current thread.
 By default it uses `InMemoryResultBackend` but this can be overridden.
 
+Because one `InMemoryBroker` instance acts as both client and worker, its
+startup and shutdown run both event-handler phases. Middleware and result
+backend lifecycle hooks still run once per broker lifecycle. Shutdown waits for
+background in-memory tasks before closing those resources and the sync-task
+executor.
+
 ## ZeroMQBroker
 
 This broker uses [ZMQ](https://zeromq.org/) to communicate between worker and client processes.

docs/examples/router/multi_broker_worker.py

@@ -0,0 +1,55 @@
+"""Expose an explicit multi-broker listener tuple for the worker CLI."""
+
+import asyncio
+from collections.abc import AsyncGenerator
+
+from taskiq import AsyncBroker, BrokerMessage, Flow, TaskiqRouter
+
+
+class DemoBroker(AsyncBroker):
+    """Small in-process transport used to keep this configuration executable."""
+
+    def __init__(self, router: TaskiqRouter, broker_name: str) -> None:
+        super().__init__(router=router, broker_name=broker_name)
+        self.messages: asyncio.Queue[bytes] = asyncio.Queue()
+
+    async def kick(self, message: BrokerMessage) -> None:
+        await self.messages.put(message.message)
+
+    async def listen(self) -> AsyncGenerator[bytes, None]:
+        while True:
+            yield await self.messages.get()
+
+
+router = TaskiqRouter()
+
+commands_broker = DemoBroker(router, "commands")
+events_broker = DemoBroker(router, "events")
+outbound_broker = DemoBroker(router, "outbound")
+
+
+@commands_broker.task(task_name="orders.notify")
+async def notify_order(order_id: str) -> str:
+    return f"notified:{order_id}"
+
+
+@commands_broker.task(task_name="orders.process")
+async def process_order(order_id: str) -> str:
+    await notify_order.kiq(order_id)
+    return f"processed:{order_id}"
+
+
+orders_flow = Flow("orders.events")
+notifications_flow = Flow("orders.notifications")
+router.route_task(process_order, broker=events_broker, flow=orders_flow)
+router.route_task(
+    notify_order,
+    broker=outbound_broker,
+    flow=notifications_flow,
+)
+router.subscribe(events_broker, orders_flow, process_order)
+
+# Only these brokers receive listeners. The outbound broker belongs to the
+# same Router, so this explicit multi-broker worker starts its client lifecycle
+# for routed child sends without consuming from it.
+worker_brokers = (commands_broker, events_broker)

docs/examples/router/multiple_brokers.py

@@ -0,0 +1,98 @@
+"""Route one task through several brokers with a shared router."""
+
+import asyncio
+
+from taskiq import Flow, InMemoryBroker, TaskiqRoute, TaskiqRouter
+
+router = TaskiqRouter()
+
+default_email_flow = Flow("emails.default")
+priority_email_flow = Flow("emails.priority")
+bulk_email_flow = Flow("emails.bulk")
+
+default_broker = InMemoryBroker(
+    router=router,
+    broker_name="default",
+    default_flow=default_email_flow,
+    await_inplace=True,
+)
+priority_broker = InMemoryBroker(
+    router=router,
+    broker_name="priority",
+    default_flow=priority_email_flow,
+    await_inplace=True,
+)
+
+
+@default_broker.task(task_name="examples.send_email", domain="notifications")
+async def send_email(user_id: int, template: str) -> str:
+    """Pretend to render and send an email."""
+    return f"{template} email sent to user {user_id}"
+
+
+priority_route = router.route_task(
+    send_email,
+    broker=priority_broker,
+    flow=priority_email_flow,
+)
+priority_subscription = router.subscribe(
+    priority_broker,
+    priority_email_flow,
+    send_email,
+)
+
+
+def _format_route(task_name: str, route: TaskiqRoute) -> str:
+    """Return a readable route diagnostic for the example output."""
+    flow_name = route.flow.name if route.flow is not None else "<default>"
+    return f"{task_name} -> broker={route.broker_name}, flow={flow_name}"
+
+
+def _format_listen_plan() -> str:
+    """Return flows that the priority broker should subscribe to."""
+    flow_names = ", ".join(flow.name for flow in priority_broker.get_subscribed_flows())
+    return f"priority listens to: {flow_names}"
+
+
+async def _main() -> None:
+    await default_broker.startup()
+    await priority_broker.startup()
+    try:
+        direct_result = await send_email(7, "welcome")
+
+        declared_route = router.resolve_route(send_email)
+        assert declared_route == priority_route
+
+        routed_task = (
+            await send_email.kicker()
+            .with_route(declared_route)
+            .kiq(
+                7,
+                "welcome",
+            )
+        )
+        routed_result = await routed_task.wait_result(timeout=2)
+
+        bulk_route = router.resolve_route(
+            send_email,
+            broker=default_broker,
+            flow=bulk_email_flow,
+        )
+        bulk_task = await send_email.kicker().with_route(bulk_route).kiq(8, "digest")
+        bulk_result = await bulk_task.wait_result(timeout=2)
+
+        print(f"Direct call: {direct_result}")
+        print(f"Router rule: {_format_route(send_email.task_name, priority_route)}")
+        print(f"Subscription tasks: {sorted(priority_subscription.task_names)}")
+        print(_format_listen_plan())
+        print(f"Resolved route: {_format_route(send_email.task_name, declared_route)}")
+        print(f"Routed call: {routed_result.return_value}")
+        print(f"Override route: {_format_route(send_email.task_name, bulk_route)}")
+        print(f"Route override: {bulk_result.return_value}")
+    finally:
+        await priority_broker.shutdown()
+        await default_broker.shutdown()
+
+
+if __name__ == "__main__":
+    asyncio.run(_main())

docs/examples/router/shared_task_package.py

@@ -0,0 +1,100 @@
+"""Declare shared task definitions and bind them in the final application."""
+
+import asyncio
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from taskiq import (
+    AsyncTaskiqDecoratedTask,
+    Flow,
+    InMemoryBroker,
+    TaskiqRouter,
+    task_builder,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class BillingQueue:
+    """Broker-specific flow that follows the shared flow protocol."""
+
+    name: str
+    priority: int
+
+    def broker_options(self) -> Mapping[str, object]:
+        """Return options that a billing broker adapter can understand."""
+        return {
+            "priority": self.priority,
+        }
+
+
+class BillingTask(AsyncTaskiqDecoratedTask[Any, Any]):
+    """Custom task class shared by billing package tasks."""
+
+    def billing_name(self) -> str:
+        """Return a billing-specific task name."""
+        return self.task_name
+
+
+@task_builder("billing.calculate_total", base_cls=BillingTask, domain="billing")
+async def calculate_total(price: int, quantity: int) -> int:
+    """Package-level task definition that is not bound to any broker."""
+    return price * quantity
+
+
+router = TaskiqRouter()
+billing_flow = Flow("billing.tasks")
+priority_billing_flow = BillingQueue(name="billing.priority", priority=10)
+
+billing_broker = InMemoryBroker(
+    router=router,
+    broker_name="billing",
+    default_flow=billing_flow,
+    await_inplace=True,
+)
+
+registered_calculate_total = router.register_task(
+    calculate_total,
+    broker=billing_broker,
+    flow=billing_flow,
+)
+router.subscribe(
+    billing_broker,
+    billing_flow,
+    registered_calculate_total,
+)
+
+
+async def _main() -> None:
+    await billing_broker.startup()
+    try:
+        direct_result = await calculate_total.call(19, 3)
+
+        priority_route = router.resolve_route(
+            registered_calculate_total,
+            broker=billing_broker,
+            flow=priority_billing_flow,
+        )
+        prepared_task = (
+            registered_calculate_total.kicker()
+            .with_route(
+                priority_route,
+            )
+            .prepare(19, 3)
+        )
+
+        queued_task = await prepared_task.kiq()
+        queued_result = await queued_task.wait_result(timeout=2)
+
+        print(f"Shared task direct call: {direct_result}")
+        print(f"Registered task class: {registered_calculate_total.billing_name()}")
+        listen_flow = billing_broker.get_subscribed_flows()[0]
+        print(f"Registered listen flow: {listen_flow.name}")
+        print(f"Prepared message: {prepared_task.message.task_name}")
+        print(f"Registered queued call: {queued_result.return_value}")
+    finally:
+        await billing_broker.shutdown()
+
+
+if __name__ == "__main__":
+    asyncio.run(_main())

docs/guide/architecture-overview.md

@@ -82,6 +82,28 @@ asyncio.run(main())
 
 ```
 
+## Router and flows
+
+Taskiq can use a `TaskiqRouter` to keep routing rules outside of broker
+implementations. Brokers remain transport adapters, while the router owns task
+registration, route resolution and flow subscriptions.
+This section describes the `experiment/separate_broker` branch contract. The
+old `@broker.task(...)`, `.kiq()`, labels, scheduler and result backend behavior
+remain compatible, while router/flow APIs are additive review material for the
+branch.
+
+`Flow` is a transport-neutral delivery address. Broker packages may provide
+their own flow classes for queue, topic, subject or stream options, as long as
+they implement the same flow protocol. The router deduplicates subscriptions by
+flow name and rejects same-name flows with incompatible broker options.
+
+Routing and subscribing are separate responsibilities. `route_task(...)`
+chooses the outbound broker and flow for task invocations. `subscribe(...)`
+adds flows to a broker listen plan for flow-aware broker adapters. Worker task
+lookup still uses `task_name`; flow does not select the Python task.
+
+Read more in the [Routing and flows](./routing-and-flows.md) section.
+
 ## Messages
 
 Every message has labels. You can define labels

docs/guide/cli.md

@@ -18,6 +18,27 @@ Like this:
 taskiq worker mybroker:broker_var my_project.module1 my_project.module2
 ```
 
+### Multiple brokers
+
+The broker import target may expose a module-level sequence of broker instances
+that use one shared Router:
+
+```python
+# my_project/worker.py
+worker_brokers = (rabbit_broker, kafka_broker)
+```
+
+```bash
+taskiq worker my_project.worker:worker_brokers my_project.tasks
+```
+
+The sequence is explicit listener configuration. Router brokers outside the
+sequence are started in client mode for routed sends and are not consumed by
+this worker. A single broker target keeps its existing lifecycle. Execution,
+prefetch and tasks-per-child limits apply to the whole worker process, not once
+per listener. See [Routing and flows](./routing-and-flows.md) for validation,
+lifecycle and adapter compatibility details.
+
 ### Sync function
 
 Taskiq can run synchronous functions. However, since it operates asynchronously, it executes them in a separate thread or process. By default, **ThreadPoolExecutor** is used. But if you're planning to use Taskiq for heavy computations, such as neural network model training or other CPU-intensive tasks, you may want to use **ProcessPoolExecutor** instead.
@@ -146,7 +167,14 @@ kill -HUP <main pid>
 If you send `SIGINT` or `SIGKILL` to the main process by pressing <kbd>Ctrl</kbd>+<kbd>C</kbd> or using the `kill` command, it will initiate the shutdown process.
 By default, it will stop fetching new messages immediately after receiving the signal but will wait for the completion of all currently executing tasks.
 
-If you don't want to wait too long for tasks to complete each time you shut down the worker, you can either send termination signals three times to the main process to perform a hard kill or configure the `--wait-tasks-timeout` to set a hard time limit for shutting down.
+If you don't want to wait indefinitely for tasks to complete during worker shutdown, you can either send termination signals three times to the main process to perform a hard kill or configure `--wait-tasks-timeout` as the graceful callback boundary.
+
+When `--wait-tasks-timeout` is omitted, Taskiq keeps the legacy behavior and
+waits for the complete listener shutdown, including running callbacks, without
+a time limit. When the task timeout is configured, `--shutdown-timeout` is also
+the post-drain listener cleanup allowance. Every managed broker then receives
+its own `--shutdown-timeout` deadline, so total multi-broker cleanup can take
+longer than that value.
 
 ::: tip Cool tip
 The number of signals before a hard kill can be configured with the `--hardkill-count` CLI argument.
@@ -170,8 +198,8 @@ The number of signals before a hard kill can be configured with the `--hardkill-
 * `--ack-type` - Type of acknowledgement. This parameter is used to set when to acknowledge the task. Possible values are `when_received`, `when_executed`, `when_saved`, `manual`. Default is `when_saved`.
 * `--max-tasks-per-child` - maximum number of tasks to be executed by a single worker process before restart.
 * `--max-fails` - Maximum number of child process exits.
-* `--shutdown-timeout` - maximum amount of time for graceful broker's shutdown in seconds (default 5).
-* `--wait-tasks-timeout` - if cannot read new messages from the broker or maximum number of tasks is reached, worker will wait for all current tasks to finish. This parameter sets the maximum amount of time to wait until shutdown.
+* `--shutdown-timeout` - maximum graceful shutdown time for each managed broker in seconds (default 5). With `--wait-tasks-timeout`, the same value is also the listener's post-drain cleanup allowance.
+* `--wait-tasks-timeout` - maximum graceful wait for active task callbacks before Taskiq cancels and awaits them. When omitted, callback and listener shutdown remains unbounded for backward compatibility.
 * `--hardkill-count` - Number of termination signals to the main process before performing a hardkill.
 
 ## Scheduler