diff --git a/CONTEXT.md b/CONTEXT.md index e03373e..4fdbb54 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -18,7 +18,9 @@ _Avoid_: job, task, event **Queue**: The `queue` column value a subscriber filters on. Not a separate object; there is nothing to declare or create. -_Avoid_: topic, channel (reserve *channel* for the `outbox_` LISTEN/NOTIFY channel) +_Avoid_: topic, channel (reserve *channel* for the `outbox_
` LISTEN/NOTIFY channel). +One exception, upstream's and not ours: `channel_labels` and the AsyncAPI document it feeds +call a queue a channel. Say *AsyncAPI channel* there. **Lease**: A time-bounded claim on a row, held as the `(acquired_token, acquired_at)` pair. It expires on its diff --git a/docs/usage/subscriber.md b/docs/usage/subscriber.md index 24942e4..5071222 100644 --- a/docs/usage/subscriber.md +++ b/docs/usage/subscriber.md @@ -31,6 +31,10 @@ The subscriber claims rows from any of its queues in a single fetch. Its [connection budget](#connection-budget) is unchanged — `max_workers + 1` pool connections regardless of how many queues it serves. +In the AsyncAPI document it appears as one channel per queue +(`orders:Handle`, `refunds:Handle`), each addressed by that queue, rather +than one channel for the subscriber. + Do **not** register two subscribers on the **same** queue: they compete for the same rows, and registration emits a warning to that effect. To run more than one handler over a queue, attach them to a single subscriber; to scale @@ -115,6 +119,10 @@ The table above lists the outbox-specific knobs. The standard FastStream subscriber kwargs pass through unchanged too: `dependencies`, `parser`, `decoder`, and the AsyncAPI `title_` / `description_` / `include_in_schema`. +`title_` names both the AsyncAPI channel and its operation. On a subscriber +spanning several queues it prefixes each channel (`Ingest:orders`, +`Ingest:refunds`), since one title cannot name several channels on its own. + ## Slow handlers — dedicated queue When a handler's tail latency exceeds the subscriber's `lease_ttl_seconds`, diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 752ced4..cdee831 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -161,28 +161,34 @@ def _render_last_exception( class OutboxSubscriberSpecification(SubscriberSpecification["OutboxBrokerConfig", OutboxSubscriberSpecificationConfig]): @property def channel_labels(self) -> list[str]: - """One channel, keyed by every queue this subscriber drains.""" - return [",".join(self.config.queues)] + """The queues this subscriber drains, one per AsyncAPI channel. - @property - def name(self) -> str: - return f"{self.channel_labels[0]}:{self.call_name}" + Deduped through a dict rather than a set: set order varies per process and would + reach the document as the order of its channels. + """ + return list(dict.fromkeys(self.config.queues)) def get_schema(self) -> dict[str, SubscriberSpec]: - return { - self.name: SubscriberSpec( - address=self.channel_labels[0], + payloads = self.get_payloads() + labels = self.channel_labels + split = len(labels) > 1 + + schema = {} + for queue in labels: + key = self._channel_key(queue, split=split) + schema[key] = SubscriberSpec( + address=queue, description=self.description, operation=Operation( message=Message( - title=f"{self.name}:Message", - payload=resolve_payloads(self.get_payloads()), + title=f"{key}:Message", + payload=resolve_payloads(payloads), ), bindings=None, ), bindings=None, ) - } + return schema class OutboxSubscriber(TasksMixin, SubscriberUsecase[OutboxInnerMessage]): diff --git a/pyproject.toml b/pyproject.toml index 219e9bd..13606d0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,11 +29,7 @@ dependencies = [ [project.optional-dependencies] asyncpg = ["asyncpg>=0.29"] validate = ["alembic>=1.13"] -# Upper cap: fastapi 0.140 made Dependant a slotted dataclass, which breaks -# faststream's FastAPI integration (ag2ai/faststream#2959). faststream declares -# no fastapi dependency of its own, so this extra is the only place the pairing -# can be constrained. Lift the cap and raise the faststream floor once fixed. -fastapi = ["fastapi>=0.95,<0.140"] +fastapi = ["fastapi>=0.95"] prometheus = ["prometheus-client>=0.19"] opentelemetry = ["opentelemetry-api>=1.20", "opentelemetry-sdk>=1.20"] all = ["faststream-outbox[asyncpg,validate,fastapi,prometheus,opentelemetry]"] @@ -45,7 +41,7 @@ dev = [ "pytest-cov", "asyncpg>=0.29", "alembic>=1.13", - "fastapi>=0.95,<0.140", # see the cap note on the fastapi extra + "fastapi>=0.95", "faststream[kafka]>=0.7.6,<0.8", "httpx2>=2.2", "prometheus-client>=0.19", diff --git a/tests/test_unit.py b/tests/test_unit.py index 81633e6..3af2ac6 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -1731,7 +1731,14 @@ async def handler(body: str) -> None: ... assert route is not None -def test_subscriber_specification_name_lists_queues() -> None: +async def test_subscriber_specification_emits_one_channel_per_queue() -> None: + """INVARIANT: a multi-queue subscriber emits one channel per queue, each addressed by that queue. + + Folding them back into a single channel keyed by the joined queue list forces a synthetic + address: ``"orders,shipments"`` names nothing a consumer can subscribe to, and AsyncAPI offers + no way to read one address as several. The joined form predates ``SubscriberSpec.address``, + when the channel key was the only queue information the document carried. + """ metadata = MetaData() t = make_outbox_table(metadata) broker = OutboxBroker(outbox_table=t) @@ -1739,10 +1746,30 @@ def test_subscriber_specification_name_lists_queues() -> None: @broker.subscriber(["orders", "shipments"]) async def handle(body: str) -> None: ... - sub = next(iter(broker._subscribers)) # noqa: SLF001 - name = sub.specification.name - assert "orders" in name - assert "shipments" in name + async with TestOutboxBroker(broker): + sub = next(iter(broker._subscribers)) # noqa: SLF001 + schema = sub.specification.get_schema() + + assert {key: spec.address for key, spec in schema.items()} == { + "orders:Handle": "orders", + "shipments:Handle": "shipments", + } + + +async def test_subscriber_title_names_the_channel_not_only_the_operation() -> None: + """``title_`` names the channel as well, matching the publisher and every built-in broker.""" + metadata = MetaData() + t = make_outbox_table(metadata) + broker = OutboxBroker(outbox_table=t) + + @broker.subscriber("orders", title_="OrderIngest") + async def handle(body: str) -> None: ... + + async with TestOutboxBroker(broker): + spec = AsyncAPI(broker).to_specification().to_jsonable() + + assert list(spec["channels"]) == ["OrderIngest"] + assert spec["channels"]["OrderIngest"]["address"] == "orders" async def test_subscriber_specification_get_schema() -> None: