diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 87b2c5be..913b15f0 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -58,6 +58,7 @@ jobs: - amgi-paho-mqtt - amgi-redis - amgi-sqs-event-source-mapping + - amgi-stomp - amgi-types - asyncfast - asyncfast-avro diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index a5ea9a09..da72361d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -11,7 +11,7 @@ repos: rev: v1.19.1 hooks: - id: mypy - additional_dependencies: ["pydantic==2.12.5", "nats-py>=2.10.0", "types-docutils", "types-PyYAML", "sphinx>=7.4.7", "starlette==1.2.1"] + additional_dependencies: ["pydantic==2.12.5", "nats-py>=2.10.0", "stompman>=3.16.0", "types-docutils", "types-PyYAML", "sphinx>=7.4.7", "starlette==1.2.1"] - repo: https://github.com/PyCQA/autoflake rev: v2.3.3 hooks: diff --git a/.python-versions b/.python-versions index f1d9c785..d4c556aa 100644 --- a/.python-versions +++ b/.python-versions @@ -1,5 +1,5 @@ -3.10 3.11 3.12 3.13 3.14 +3.10 diff --git a/README.md b/README.md index 255207f2..650c4b27 100644 --- a/README.md +++ b/README.md @@ -206,6 +206,10 @@ The following AMGI servers are currently available: - **[amgi-redis](https://pypi.org/project/amgi-redis/)** - [Redis](https://redis.io/) server implementation +### STOMP + +- **[amgi-stomp](https://pypi.org/project/amgi-stomp/)** - [STOMP](https://stomp.github.io/) server implementation + ### SQS - **[amgi-aiobotocore](https://pypi.org/project/amgi-aiobotocore/)** - Contains a [SQS](https://aws.amazon.com/sqs/) diff --git a/docs/specifications/bindings.rst b/docs/specifications/bindings.rst index b754edea..865136fc 100644 --- a/docs/specifications/bindings.rst +++ b/docs/specifications/bindings.rst @@ -18,6 +18,17 @@ The Kafka binding object has the following key: - ``message["bindings"]["keys"]`` (:py:obj:`~typing.NotRequired` [ :py:obj:`~typing.Optional` [:py:obj:`bytes` ]]) - The :py:obj:`bytes` of the Kafka key, this can be also be :py:obj:`None`. If missing, it defaults to :py:obj:`None`. +******** + STOMP +******** + +The STOMP binding object has the following keys: + +- ``message["bindings"]["stomp"]["message_id"]`` - The :py:obj:`str` message id of the STOMP message, assigned by the + broker. Useful for deduplication. +- ``message["bindings"]["stomp"]["subscription"]`` - The :py:obj:`str` id of the STOMP subscription the message was + received on. + .. _asyncapi: https://www.asyncapi.com/en .. _message bindings object: https://www.asyncapi.com/docs/reference/specification/v3.0.0#messageBindingsObject diff --git a/example/README.md b/example/README.md index 01818578..2305c2f3 100644 --- a/example/README.md +++ b/example/README.md @@ -146,3 +146,23 @@ Connect the app via AMQP with: ```commandline asyncfast run amgi-aio-pika main:app input_channel ``` + +## STOMP + +Run the STOMP compose file with: + +```commandline +docker compose --file stomp/docker-compose.yaml up --detach +``` + +This includes: + +- An ActiveMQ Artemis broker with STOMP enabled, running on `localhost:61613` +- The Artemis web console ([`http://localhost:8161/console`](http://localhost:8161/console)) logging in with + `artemis`/`artemis` + +Connect the app via STOMP with: + +```commandline +asyncfast run amgi-stomp main:app input_channel --login artemis --passcode artemis +``` diff --git a/example/pyproject.toml b/example/pyproject.toml index 804a7584..f3ebf4fd 100644 --- a/example/pyproject.toml +++ b/example/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "amgi-aiokafka", "amgi-nats", "amgi-paho-mqtt", + "amgi-stomp", "asyncfast", ] @@ -25,4 +26,5 @@ sources.amgi-aio-pika = { workspace = true } sources.amgi-aiokafka = { workspace = true } sources.amgi-nats = { workspace = true } sources.amgi-paho-mqtt = { workspace = true } +sources.amgi-stomp = { workspace = true } sources.asyncfast = { workspace = true } diff --git a/example/stomp/docker-compose.yaml b/example/stomp/docker-compose.yaml new file mode 100644 index 00000000..eb5bfe0c --- /dev/null +++ b/example/stomp/docker-compose.yaml @@ -0,0 +1,7 @@ +services: + activemq-artemis: + image: ghcr.io/asyncfast/apache/activemq-artemis:2.44.0 + restart: unless-stopped + ports: + - "61613:61613" + - "8161:8161" diff --git a/packages/amgi-stomp/LICENSE b/packages/amgi-stomp/LICENSE new file mode 100644 index 00000000..5da9ba59 --- /dev/null +++ b/packages/amgi-stomp/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2025 Jack Burridge + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/packages/amgi-stomp/README.md b/packages/amgi-stomp/README.md new file mode 100644 index 00000000..63462186 --- /dev/null +++ b/packages/amgi-stomp/README.md @@ -0,0 +1,72 @@ +# amgi-stomp + +amgi-stomp is an [AMGI](https://amgi.readthedocs.io/en/latest/) compatible server to run AMGI applications against +[STOMP](https://stomp.github.io/) brokers using [stompman](https://pypi.org/project/stompman/). + +amgi-stomp requires Python 3.11 or greater, as this is the minimum version supported by `stompman`. + +## Installation + +``` +pip install amgi-stomp==0.47.0 +``` + +## Example + +This example uses [AsyncFast](https://pypi.org/project/asyncfast/): + +```python +from dataclasses import dataclass + +from amgi_stomp import run +from asyncfast import AsyncFast + +app = AsyncFast() + + +@dataclass +class Order: + item_ids: list[str] + + +@app.channel("order-queue") +async def order_queue(order: Order) -> None: + # Makes an order + ... + + +if __name__ == "__main__": + run(app, "order-queue") +``` + +Or the application could be run via the commandline: + +```commandline +asyncfast run amgi-stomp main:app order-queue +``` + +Messages are consumed with `client-individual` acknowledgement mode. Sending `message.ack` or `message.nack` events +acknowledges or rejects the current STOMP message. Whether messages that are never acknowledged or are nacked are +redelivered, dead-lettered, or discarded depends on the broker's configuration. + +If an application raises an exception while handling a message, the exception is logged and the message is nacked; what the +broker does with a nacked message (redeliver, dead-letter, or discard) is broker-specific. + +The STOMP protocol metadata of a received message is available in the AMGI message scope: +`message["bindings"]["stomp"]["message_id"]` and `message["bindings"]["stomp"]["subscription"]`. When using AsyncFast, +these can be received directly with the `asyncfast.StompMessageId` and `asyncfast.StompSubscription` bindings. + +By default, messages are sent without waiting for a broker receipt. Pass a `receipt_timeout` in seconds to `run`, +`Server`, or `MessageSend` to wait for a receipt confirming the broker processed the SEND frame, at the cost of a broker round-trip. A receipt does +not by itself make a message durable; persistence is broker-specific. Pass +`ssl=True` or an `ssl.SSLContext` via the `ssl` parameter of `run` or `Server` to connect over TLS. The `heartbeat`, +`connect_retry_attempts`, and `connect_retry_interval` parameters expose stompman's heartbeat and connection retry +options without changing its defaults. + +## Contact + +For questions or suggestions, please contact [jack.burridge@mail.com](mailto:jack.burridge@mail.com). + +## License + +Copyright 2025 AMGI diff --git a/packages/amgi-stomp/pyproject.toml b/packages/amgi-stomp/pyproject.toml new file mode 100644 index 00000000..67352aca --- /dev/null +++ b/packages/amgi-stomp/pyproject.toml @@ -0,0 +1,48 @@ +[build-system] +build-backend = "uv_build" +requires = [ + "uv-build>=0.8.14,<0.9.0", +] + +[project] +name = "amgi-stomp" +version = "0.49.0" +description = "AMGI STOMP Server" +readme = "README.md" +license = "MIT" +license-files = [ + "LICENSE", +] +authors = [ + { name = "jack.burridge", email = "jack.burridge@mail.com" }, +] +requires-python = ">=3.10" +classifiers = [ + "Programming Language :: Python :: 3 :: Only", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", +] +dependencies = [ + "amgi-common==0.49.0", + "amgi-types==0.49.0", + "stompman>=3.16.0; python_full_version>='3.11'", +] +entry-points.amgi_server.amgi-stomp = "amgi_stomp:_run_cli" + +[dependency-groups] +dev = [ + "pytest>=8.4.1", + "pytest-asyncio>=1.3.0", + "pytest-cov>=7.0.0", + "pytest-timeout>=2.4.0", + "test-utils", + "testcontainers>=4.13.0", +] + +[tool.uv] +sources.amgi-common.workspace = true +sources.amgi-types.workspace = true +sources.test-utils.workspace = true diff --git a/packages/amgi-stomp/src/amgi_stomp/__init__.py b/packages/amgi-stomp/src/amgi_stomp/__init__.py new file mode 100644 index 00000000..87f0b9ca --- /dev/null +++ b/packages/amgi-stomp/src/amgi_stomp/__init__.py @@ -0,0 +1,381 @@ +import asyncio +import logging +import sys +from collections.abc import Awaitable +from collections.abc import Callable +from collections.abc import Sequence +from contextlib import suppress +from functools import partial +from ssl import SSLContext +from types import TracebackType +from typing import Any +from typing import AsyncContextManager +from typing import cast +from typing import Literal + +if sys.version_info < (3, 11): # pragma: no cover + raise RuntimeError("amgi-stomp requires Python 3.11 or greater") + +from typing import Self + +import stompman +from amgi_common import Lifespan +from amgi_common import server_serve +from amgi_types import AMGIApplication +from amgi_types import AMGIReceiveEvent +from amgi_types import AMGISendEvent +from amgi_types import MessageScope +from amgi_types import MessageSendEvent + +_MessageSendT = Callable[[MessageSendEvent], Awaitable[None]] +_MessageSendManagerT = AsyncContextManager[_MessageSendT] +_SendT = Callable[[AMGISendEvent], Awaitable[None]] + +_FRAME_METADATA_HEADERS = frozenset( + {"destination", "message-id", "subscription", "ack", "content-length"} +) +_SHUTDOWN_TIMEOUT = 10 + +logger = logging.getLogger("amgi-stomp.error") + + +def run( + app: AMGIApplication, + *destinations: str, + host: str = "localhost", + port: int = 61613, + login: str = "guest", + passcode: str = "guest", + ssl: Literal[True] | SSLContext | None = None, + heartbeat: tuple[int, int] = (1000, 1000), + connect_retry_attempts: int = 3, + connect_retry_interval: int = 1, + receipt_timeout: float | None = None, + message_send: _MessageSendManagerT | None = None, +) -> None: + server = Server( + app, + *destinations, + host=host, + port=port, + login=login, + passcode=passcode, + ssl=ssl, + heartbeat=heartbeat, + connect_retry_attempts=connect_retry_attempts, + connect_retry_interval=connect_retry_interval, + receipt_timeout=receipt_timeout, + message_send=message_send, + ) + server_serve(server) + + +def _run_cli( + app: AMGIApplication, + destinations: list[str], + host: str = "localhost", + port: int = 61613, + login: str = "guest", + passcode: str = "guest", +) -> None: + run(app, *destinations, host=host, port=port, login=login, passcode=passcode) + + +async def _receive() -> AMGIReceiveEvent: + raise RuntimeError("Receive should not be called") + + +def _encode_headers(headers: dict[str, str] | None) -> list[tuple[bytes, bytes]]: + if not headers: + return [] + return [(key.encode(), value.encode()) for key, value in headers.items()] + + +def _decode_headers(headers: Sequence[tuple[bytes, bytes]]) -> dict[str, str]: + return {key.decode(): value.decode() for key, value in headers} + + +def _create_client( + connection_parameters: stompman.ConnectionParameters, + *, + ssl: Literal[True] | SSLContext | None, + heartbeat: tuple[int, int], + connect_retry_attempts: int, + connect_retry_interval: int, +) -> stompman.Client: + return stompman.Client( + servers=[connection_parameters], + ssl=ssl, + heartbeat=stompman.Heartbeat(*heartbeat), + connect_retry_attempts=connect_retry_attempts, + connect_retry_interval=connect_retry_interval, + ) + + +class _Send: + def __init__( + self, + frame: stompman.AckableMessageFrame, + message_send: _MessageSendT, + ) -> None: + self._frame = frame + self._message_send = message_send + + async def __call__(self, event: AMGISendEvent) -> None: + if event["type"] == "message.ack": + await self._frame.ack() + elif event["type"] == "message.nack": + await self._frame.nack() + elif event["type"] == "message.send": + await self._message_send(event) + + +class MessageSend: + def __init__( + self, + host: str, + port: int, + login: str, + passcode: str, + ssl: Literal[True] | SSLContext | None = None, + heartbeat: tuple[int, int] = (1000, 1000), + connect_retry_attempts: int = 3, + connect_retry_interval: int = 1, + receipt_timeout: float | None = None, + ) -> None: + self._client: stompman.Client | None = None + self._connection_parameters = stompman.ConnectionParameters( + host=host, + port=port, + login=login, + passcode=passcode, + ) + self._ssl = ssl + self._heartbeat = heartbeat + self._connect_retry_attempts = connect_retry_attempts + self._connect_retry_interval = connect_retry_interval + self._receipt_timeout = receipt_timeout + + async def __aenter__(self) -> Self: + self._client = _create_client( + self._connection_parameters, + ssl=self._ssl, + heartbeat=self._heartbeat, + connect_retry_attempts=self._connect_retry_attempts, + connect_retry_interval=self._connect_retry_interval, + ) + await self._client.__aenter__() + return self + + async def __call__(self, event: MessageSendEvent) -> None: + if self._client is None: + raise RuntimeError("MessageSend not initialized") + + await self._client.send( + event.get("payload") or b"", + event["address"], + headers=_decode_headers(event["headers"]), + receipt_timeout=self._receipt_timeout, + ) + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + if self._client is not None: + await self._client.__aexit__(exc_type, exc_val, exc_tb) + + +class Server: + def __init__( + self, + app: AMGIApplication, + *destinations: str, + host: str, + port: int, + login: str, + passcode: str, + ssl: Literal[True] | SSLContext | None = None, + heartbeat: tuple[int, int] = (1000, 1000), + connect_retry_attempts: int = 3, + connect_retry_interval: int = 1, + receipt_timeout: float | None = None, + message_send: _MessageSendManagerT | None = None, + ) -> None: + self._app = app + self._destinations = destinations + self._connection_parameters = stompman.ConnectionParameters( + host=host, + port=port, + login=login, + passcode=passcode, + ) + self._ssl = ssl + self._heartbeat = heartbeat + self._connect_retry_attempts = connect_retry_attempts + self._connect_retry_interval = connect_retry_interval + self._message_send = message_send or MessageSend( + host, + port, + login, + passcode, + ssl=ssl, + heartbeat=heartbeat, + connect_retry_attempts=connect_retry_attempts, + connect_retry_interval=connect_retry_interval, + receipt_timeout=receipt_timeout, + ) + self._stopped = asyncio.Event() + self._tasks = set[asyncio.Task[None]]() + self._accepting = True + + async def serve(self) -> None: + client = _create_client( + self._connection_parameters, + ssl=self._ssl, + heartbeat=self._heartbeat, + connect_retry_attempts=self._connect_retry_attempts, + connect_retry_interval=self._connect_retry_interval, + ) + await client.__aenter__() + try: + await self._serve(client) + except BaseException as exc: + # A connection whose socket has stalled must not block shutdown, so + # closing it is bounded and best-effort. The in-flight exception is + # passed through so stompman does not wait for handlers to drain + # when serve() is already unwinding with an error + with suppress(asyncio.TimeoutError): + await asyncio.wait_for( + client.__aexit__(type(exc), exc, exc.__traceback__), + _SHUTDOWN_TIMEOUT, + ) + raise + with suppress(asyncio.TimeoutError): + await asyncio.wait_for( + client.__aexit__(None, None, None), _SHUTDOWN_TIMEOUT + ) + + async def _serve(self, client: stompman.Client) -> None: + message_send = await self._message_send.__aenter__() + try: + async with Lifespan(self._app) as state: + subscriptions = [ + await client.subscribe_with_manual_ack( + destination, + partial( + self._handle_message, + message_send=message_send, + state=state, + ), + ) + for destination in self._destinations + ] + await self._stopped.wait() + self._accepting = False + await self._drain_tasks() + for subscription in subscriptions: + with suppress(asyncio.TimeoutError): + await asyncio.wait_for( + subscription.unsubscribe(), _SHUTDOWN_TIMEOUT + ) + except BaseException as exc: + if not await self._exit_message_send(exc): + raise + else: + await self._exit_message_send(None) + + async def _drain_tasks(self) -> None: + """Wait for in-flight message handlers to complete before unsubscribing. + + stompman refuses to acknowledge messages on an unsubscribed subscription, + so unsubscribing first would leave messages that were processed, but never + acknowledged, to be redelivered after a restart. Handlers are drained + while the subscription is still active. New callbacks are no longer + accepted by this point, so the set of tasks is stable. Tasks still + running after the shutdown timeout are cancelled, leaving their + messages unacknowledged so the broker may redeliver them. + """ + if not self._tasks: + return + _, pending = await asyncio.wait(list(self._tasks), timeout=_SHUTDOWN_TIMEOUT) + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + + async def _exit_message_send(self, exc: BaseException | None) -> bool: + """Exit the message send manager, bounded so a stalled socket cannot + block shutdown. Returns whether the manager suppressed the exception.""" + try: + if exc is None: + return bool( + await asyncio.wait_for( + self._message_send.__aexit__(None, None, None), + _SHUTDOWN_TIMEOUT, + ) + ) + return bool( + await asyncio.wait_for( + self._message_send.__aexit__(type(exc), exc, exc.__traceback__), + _SHUTDOWN_TIMEOUT, + ) + ) + except asyncio.TimeoutError: + return False + + async def _handle_message( + self, + frame: stompman.AckableMessageFrame, + *, + message_send: _MessageSendT, + state: dict[str, Any], + ) -> None: + if not self._accepting: + # Left unacknowledged, so the broker may redeliver it + return + task = asyncio.get_running_loop().create_task( + self._process_message(frame, message_send, state) + ) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + async def _process_message( + self, + frame: stompman.AckableMessageFrame, + message_send: _MessageSendT, + state: dict[str, Any], + ) -> None: + scope: MessageScope = { + "type": "message", + "amgi": {"version": "2.0", "spec_version": "2.0"}, + "address": frame.headers["destination"], + "headers": _encode_headers( + { + key: value + for key, value in cast("dict[str, str]", frame.headers).items() + if key not in _FRAME_METADATA_HEADERS + } + ), + "payload": frame.body, + "bindings": { + "stomp": { + "message_id": frame.headers["message-id"], + "subscription": frame.headers["subscription"], + } + }, + "state": state.copy(), + } + try: + await self._app(scope, _receive, _Send(frame, message_send)) + except Exception: + logger.exception("Application raised an exception") + # Whether the message is redelivered depends on the broker's + # policy. The subscription may already be inactive during + # shutdown, in which case stompman only warns + with suppress(Exception): + await frame.nack() + + def stop(self) -> None: + self._stopped.set() diff --git a/packages/amgi-stomp/src/amgi_stomp/py.typed b/packages/amgi-stomp/src/amgi_stomp/py.typed new file mode 100644 index 00000000..e69de29b diff --git a/packages/amgi-stomp/tests_amgi_stomp/test_stomp_message.py b/packages/amgi-stomp/tests_amgi_stomp/test_stomp_message.py new file mode 100644 index 00000000..a2d98aa5 --- /dev/null +++ b/packages/amgi-stomp/tests_amgi_stomp/test_stomp_message.py @@ -0,0 +1,80 @@ +from types import SimpleNamespace +from typing import cast +from unittest.mock import AsyncMock + +import pytest +from amgi_stomp import _decode_headers +from amgi_stomp import _encode_headers +from amgi_stomp import _Send +from amgi_stomp import MessageSend +from stompman import AckableMessageFrame +from stompman import Client + + +async def test_send_ack() -> None: + frame = SimpleNamespace(ack=AsyncMock(), nack=AsyncMock()) + message_send = AsyncMock() + + await _Send(cast(AckableMessageFrame, frame), message_send)({"type": "message.ack"}) + + frame.ack.assert_awaited_once_with() + frame.nack.assert_not_awaited() + message_send.assert_not_awaited() + + +async def test_send_nack() -> None: + frame = SimpleNamespace(ack=AsyncMock(), nack=AsyncMock()) + message_send = AsyncMock() + + await _Send(cast(AckableMessageFrame, frame), message_send)( + {"type": "message.nack", "message": "reject"} + ) + + frame.nack.assert_awaited_once_with() + frame.ack.assert_not_awaited() + message_send.assert_not_awaited() + + +async def test_send_message_send() -> None: + frame = SimpleNamespace(ack=AsyncMock(), nack=AsyncMock()) + message_send = AsyncMock() + + await _Send(cast(AckableMessageFrame, frame), message_send)( + {"type": "message.send", "address": "destination", "headers": []} + ) + + message_send.assert_awaited_once_with( + {"type": "message.send", "address": "destination", "headers": []} + ) + frame.ack.assert_not_awaited() + frame.nack.assert_not_awaited() + + +async def test_message_send_requires_context_manager() -> None: + message_send = MessageSend("localhost", 61613, "guest", "guest") + + with pytest.raises(RuntimeError, match="MessageSend not initialized"): + await message_send( + {"type": "message.send", "address": "destination", "headers": []} + ) + + +async def test_message_send_closes_client() -> None: + message_send = MessageSend("localhost", 61613, "guest", "guest") + client_exit = AsyncMock() + message_send._client = cast(Client, SimpleNamespace(__aexit__=client_exit)) + + await message_send.__aexit__(None, None, None) + + client_exit.assert_awaited_once_with(None, None, None) + + +def test_encode_headers_handles_empty_and_values() -> None: + assert _encode_headers(None) == [] + assert _encode_headers({}) == [] + assert _encode_headers({"name": "value"}) == [(b"name", b"value")] + + +def test_decode_headers_handles_empty_and_values() -> None: + assert _decode_headers([]) == {} + assert _decode_headers([(b"name", b"value")]) == {"name": "value"} diff --git a/packages/amgi-stomp/tests_amgi_stomp/test_stomp_message_integration.py b/packages/amgi-stomp/tests_amgi_stomp/test_stomp_message_integration.py new file mode 100644 index 00000000..12d0d6fe --- /dev/null +++ b/packages/amgi-stomp/tests_amgi_stomp/test_stomp_message_integration.py @@ -0,0 +1,457 @@ +import asyncio +import time +from collections.abc import AsyncGenerator +from collections.abc import Generator +from types import TracebackType +from typing import Any +from unittest.mock import ANY +from uuid import uuid4 + +import pytest +from amgi_common import LifespanFailureError +from amgi_stomp import _MessageSendT +from amgi_stomp import _run_cli +from amgi_stomp import run +from amgi_stomp import Server +from amgi_types import AMGIReceiveCallable +from amgi_types import AMGISendCallable +from amgi_types import MessageSendEvent +from amgi_types import Scope +from stompman import AckableMessageFrame +from stompman import Client +from stompman import ConnectionParameters +from stompman import MessageFrame +from test_utils import assert_run_can_terminate +from test_utils import MockApp +from testcontainers.core.container import DockerContainer + +_LOGIN = "artemis" +_PASSCODE = "artemis" + + +async def _wait_for_stomp(connection_parameters: ConnectionParameters) -> None: + deadline = time.monotonic() + 60 + while True: + try: + async with Client( + servers=[connection_parameters], + connect_retry_attempts=1, + connect_retry_interval=0, + ): + return + except Exception: # pragma: no cover + if time.monotonic() > deadline: + raise + await asyncio.sleep(0.5) + + +async def _send( + client: Client, body: bytes, destination: str, headers: dict[str, str] | None = None +) -> None: + await client.send(body, destination, headers=headers) + + +@pytest.fixture(scope="module") +def artemis_container() -> Generator[DockerContainer, None, None]: + with DockerContainer( + image="ghcr.io/asyncfast/apache/activemq-artemis:2.44.0" + ).with_exposed_ports(61613) as container: + asyncio.run( + _wait_for_stomp( + ConnectionParameters( + host=container.get_container_host_ip(), + port=int(container.get_exposed_port(61613)), + login=_LOGIN, + passcode=_PASSCODE, + ) + ) + ) + yield container + + +@pytest.fixture +def destination() -> str: + return f"/queue/queue-{uuid4()}" + + +@pytest.fixture +def connection_parameters( + artemis_container: DockerContainer, +) -> ConnectionParameters: + return ConnectionParameters( + host=artemis_container.get_container_host_ip(), + port=int(artemis_container.get_exposed_port(61613)), + login=_LOGIN, + passcode=_PASSCODE, + ) + + +@pytest.fixture +async def client( + connection_parameters: ConnectionParameters, +) -> AsyncGenerator[Client, None]: + async with Client(servers=[connection_parameters]) as stomp_client: + yield stomp_client + + +class _TestServer(Server): + """A server that reports once the broker has confirmed its subscriptions + + Artemis drops a message sent to a queue that does not exist yet, so a test + must not send before the server has subscribed + """ + + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self._subscribed = asyncio.Event() + + async def _serve(self, client: Client) -> None: + server = self + + class ConfirmedClient: + async def subscribe_with_manual_ack(self, *args: Any, **kwargs: Any) -> Any: + subscription = await client.subscribe_with_manual_ack( + *args, receipt_timeout=5, **kwargs + ) + server._subscribed.set() + return subscription + + await super()._serve(ConfirmedClient()) # type: ignore[arg-type] + + async def wait_subscribed(self) -> None: + await self._subscribed.wait() + + +def _make_server( + app: MockApp, + destination: str, + connection_parameters: ConnectionParameters, + receipt_timeout: float | None = None, +) -> _TestServer: + return _TestServer( + app, + destination, + host=connection_parameters.host, + port=connection_parameters.port, + login=connection_parameters.login, + passcode=connection_parameters.passcode, + receipt_timeout=receipt_timeout, + ) + + +@pytest.fixture +async def app( + connection_parameters: ConnectionParameters, destination: str +) -> AsyncGenerator[MockApp, None]: + app = MockApp() + server = _make_server(app, destination, connection_parameters) + async with app.lifespan(server=server): + await server.wait_subscribed() + yield app + + +@pytest.fixture +async def app_with_state( + connection_parameters: ConnectionParameters, destination: str +) -> AsyncGenerator[tuple[MockApp, dict[str, Any]], None]: + app = MockApp() + state = {"item": uuid4()} + server = _make_server(app, destination, connection_parameters) + async with app.lifespan(state, server=server): + await server.wait_subscribed() + yield app, state + + +@pytest.mark.integration +async def test_message(app: MockApp, destination: str, client: Client) -> None: + await _send(client, b"value", destination, headers={"custom": "value"}) + + async with app.call() as (scope, receive, send): + assert scope == { + "type": "message", + "amgi": {"version": "2.0", "spec_version": "2.0"}, + "address": destination, + "headers": [ + (b"expires", b"0"), + (b"redelivered", b"false"), + (b"priority", b"4"), + (b"persistent", b"false"), + (b"timestamp", ANY), + (b"custom", b"value"), + ], + "payload": b"value", + "bindings": {"stomp": {"message_id": ANY, "subscription": ANY}}, + "state": {}, + } + await send({"type": "message.ack"}) + + +@pytest.mark.integration +async def test_message_send(app: MockApp, destination: str, client: Client) -> None: + send_destination = f"/queue/send-{uuid4()}" + messages: asyncio.Queue[MessageFrame] = asyncio.Queue() + + async def handler(frame: AckableMessageFrame) -> None: + await messages.put(frame) + + subscription = await client.subscribe_with_manual_ack(send_destination, handler) + + await _send(client, b"", destination) + + async with app.call() as (scope, receive, send): + await send( + { + "type": "message.send", + "address": send_destination, + "headers": [(b"key", b"value")], + "payload": b"test", + } + ) + await send({"type": "message.ack"}) + + frame = await asyncio.wait_for(messages.get(), 5) + assert frame.body == b"test" + assert frame.headers["destination"] == send_destination + assert frame.headers.get("key") == "value" + + await subscription.unsubscribe() + + +@pytest.mark.integration +async def test_message_send_defaults_empty_payload_and_headers( + app: MockApp, destination: str, client: Client +) -> None: + send_destination = f"/queue/send-{uuid4()}" + messages: asyncio.Queue[MessageFrame] = asyncio.Queue() + + async def handler(frame: AckableMessageFrame) -> None: + await messages.put(frame) + + subscription = await client.subscribe_with_manual_ack(send_destination, handler) + + await _send(client, b"", destination) + + async with app.call() as (scope, receive, send): + await send({"type": "message.send", "address": send_destination, "headers": []}) + await send({"type": "message.ack"}) + + frame = await asyncio.wait_for(messages.get(), 5) + assert frame.body == b"" + assert frame.headers.get("key") is None + + await subscription.unsubscribe() + + +@pytest.mark.integration +@pytest.mark.timeout(30) +async def test_message_send_receipt( + connection_parameters: ConnectionParameters, destination: str, client: Client +) -> None: + send_destination = f"/queue/receipt-{uuid4()}" + messages: asyncio.Queue[MessageFrame] = asyncio.Queue() + + async def handler(frame: AckableMessageFrame) -> None: + await messages.put(frame) + + subscription = await client.subscribe_with_manual_ack(send_destination, handler) + + receipt_app = MockApp() + server = _make_server( + receipt_app, destination, connection_parameters, receipt_timeout=5 + ) + async with receipt_app.lifespan(server=server): + await server.wait_subscribed() + await _send(client, b"", destination) + + async with receipt_app.call() as (scope, receive, send): + await send( + { + "type": "message.send", + "address": send_destination, + "headers": [], + "payload": b"receipt", + } + ) + await send({"type": "message.ack"}) + + frame = await asyncio.wait_for(messages.get(), 5) + assert frame.body == b"receipt" + + await subscription.unsubscribe() + + +@pytest.mark.integration +async def test_message_ack(app: MockApp, destination: str, client: Client) -> None: + await _send(client, b"value", destination) + + async with app.call() as (scope, receive, send): + await send({"type": "message.ack"}) + + messages: asyncio.Queue[MessageFrame] = asyncio.Queue() + + async def handler(frame: AckableMessageFrame) -> None: + # Only reached if the broker wrongly redelivers an acknowledged message + await messages.put(frame) # pragma: no cover + + subscription = await client.subscribe_with_manual_ack(destination, handler) + + with pytest.raises(TimeoutError): + await asyncio.wait_for(messages.get(), 1) + + await subscription.unsubscribe() + + +@pytest.mark.integration +async def test_message_nack(app: MockApp, destination: str, client: Client) -> None: + await _send(client, b"value", destination) + + async with app.call() as (scope, receive, send): + assert scope["type"] == "message" + assert scope["payload"] == b"value" + await send({"type": "message.nack", "message": "reject"}) + + # The broker does not redeliver the nacked message, but the subscription stays + # healthy and processes the next message + await _send(client, b"second", destination) + + async with app.call() as (scope, receive, send): + assert scope["type"] == "message" + assert scope["payload"] == b"second" + await send({"type": "message.ack"}) + + +@pytest.mark.integration +async def test_lifespan( + app_with_state: tuple[MockApp, dict[str, Any]], + destination: str, + client: Client, +) -> None: + app, state = app_with_state + + await _send(client, b"", destination) + + async with app.call() as (scope, receive, send): + assert scope["state"] == state + await send({"type": "message.ack"}) + + +@pytest.mark.integration +async def test_message_receive_not_callable( + app: MockApp, destination: str, client: Client +) -> None: + await _send(client, b"test", destination) + + async with app.call() as (scope, receive, send): + with pytest.raises(RuntimeError, match="Receive should not be called"): + await receive() + await send({"type": "message.ack"}) + + +@pytest.mark.integration +def test_run(destination: str, connection_parameters: ConnectionParameters) -> None: + assert_run_can_terminate( + run, + destination, + host=connection_parameters.host, + port=connection_parameters.port, + login=connection_parameters.login, + passcode=connection_parameters.passcode, + ) + + +@pytest.mark.integration +def test_run_cli(destination: str, connection_parameters: ConnectionParameters) -> None: + assert_run_can_terminate( + _run_cli, + [destination], + host=connection_parameters.host, + port=connection_parameters.port, + login=connection_parameters.login, + passcode=connection_parameters.passcode, + ) + + +async def _failing_lifespan_app( + scope: Scope, receive: AMGIReceiveCallable, send: AMGISendCallable +) -> None: + assert scope["type"] == "lifespan" + event = await receive() + assert event == {"type": "lifespan.startup"} + await send({"type": "lifespan.startup.failed", "message": "startup failed"}) + + +class _StubMessageSend: + def __init__(self, suppress: bool = False, exit_delay: float = 0) -> None: + self._suppress = suppress + self._exit_delay = exit_delay + + async def __aenter__(self) -> _MessageSendT: + async def message_send(event: MessageSendEvent) -> None: + return None # pragma: no cover + + return message_send + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> bool: + await asyncio.sleep(self._exit_delay) + return self._suppress + + +@pytest.mark.integration +async def test_serve_lifespan_startup_failure( + connection_parameters: ConnectionParameters, destination: str +) -> None: + server = Server( + _failing_lifespan_app, + destination, + host=connection_parameters.host, + port=connection_parameters.port, + login=connection_parameters.login, + passcode=connection_parameters.passcode, + ) + serve_task = asyncio.create_task(server.serve()) + + with pytest.raises(LifespanFailureError): + await asyncio.wait_for(serve_task, 5) + + +@pytest.mark.integration +async def test_serve_message_send_suppresses_exception( + connection_parameters: ConnectionParameters, destination: str +) -> None: + server = Server( + _failing_lifespan_app, + destination, + host=connection_parameters.host, + port=connection_parameters.port, + login=connection_parameters.login, + passcode=connection_parameters.passcode, + message_send=_StubMessageSend(suppress=True), + ) + serve_task = asyncio.create_task(server.serve()) + + await asyncio.wait_for(serve_task, 5) + + +@pytest.mark.integration +@pytest.mark.timeout(30) +async def test_serve_message_send_exit_timeout( + connection_parameters: ConnectionParameters, destination: str +) -> None: + server = Server( + _failing_lifespan_app, + destination, + host=connection_parameters.host, + port=connection_parameters.port, + login=connection_parameters.login, + passcode=connection_parameters.passcode, + message_send=_StubMessageSend(exit_delay=60), + ) + serve_task = asyncio.create_task(server.serve()) + + with pytest.raises(LifespanFailureError): + await asyncio.wait_for(serve_task, 30) diff --git a/packages/amgi-stomp/tests_amgi_stomp/test_stomp_server.py b/packages/amgi-stomp/tests_amgi_stomp/test_stomp_server.py new file mode 100644 index 00000000..9164ce9f --- /dev/null +++ b/packages/amgi-stomp/tests_amgi_stomp/test_stomp_server.py @@ -0,0 +1,368 @@ +import asyncio +import logging +from collections.abc import Callable +from collections.abc import Coroutine +from types import SimpleNamespace +from types import TracebackType +from typing import Any +from typing import cast +from unittest.mock import AsyncMock + +import pytest +import stompman +from amgi_stomp import _MessageSendT +from amgi_stomp import Server +from amgi_types import AMGIReceiveCallable +from amgi_types import AMGISendCallable +from amgi_types import MessageScope +from amgi_types import MessageSendEvent +from amgi_types import Scope +from stompman import AckableMessageFrame + + +class _StubMessageSendManager: + async def __aenter__(self) -> _MessageSendT: + async def message_send(event: MessageSendEvent) -> None: + return None # pragma: no cover + + return message_send + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> bool: + return False + + +class _StubSubscription: + def __init__(self, events: list[str]) -> None: + self._events = events + + async def unsubscribe(self) -> None: + self._events.append("unsubscribe") + + +class _StubClient: + def __init__(self, subscription: _StubSubscription, events: list[str]) -> None: + self._subscription = subscription + self._events = events + self.handler: ( + Callable[[AckableMessageFrame], Coroutine[Any, Any, None]] | None + ) = None + self.exit_args: ( + tuple[type[BaseException] | None, BaseException | None, Any] | None + ) = None + self.subscribed = asyncio.Event() + + async def __aenter__(self) -> "_StubClient": + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + self.exit_args = (exc_type, exc_val, exc_tb) + + async def subscribe_with_manual_ack( + self, + destination: str, + handler: Callable[[AckableMessageFrame], Coroutine[Any, Any, None]], + **kwargs: Any, + ) -> _StubSubscription: + self.handler = handler + self.subscribed.set() + return self._subscription + + +def _stub_frame( + events: list[str] | None = None, + ack: AsyncMock | None = None, + nack: AsyncMock | None = None, +) -> AckableMessageFrame: + async def _ack() -> None: + if events is not None: + events.append("ack") + + return cast( + AckableMessageFrame, + SimpleNamespace( + headers={ + "destination": "destination", + "message-id": "message-id-1", + "subscription": "subscription-1", + }, + body=b"value", + ack=ack or AsyncMock(side_effect=_ack), + nack=nack or AsyncMock(), + ), + ) + + +async def _lifespan_app_call( + scope: Scope, receive: AMGIReceiveCallable, send: AMGISendCallable +) -> None: + assert await receive() == {"type": "lifespan.startup"} + await send({"type": "lifespan.startup.complete"}) + assert await receive() == {"type": "lifespan.shutdown"} + await send({"type": "lifespan.shutdown.complete"}) + + +async def test_serve_acks_in_flight_message_before_unsubscribe( + monkeypatch: pytest.MonkeyPatch, +) -> None: + events: list[str] = [] + message_acked = asyncio.Event() + release_message = asyncio.Event() + + async def app( + scope: Scope, receive: AMGIReceiveCallable, send: AMGISendCallable + ) -> None: + if scope["type"] == "lifespan": + await _lifespan_app_call(scope, receive, send) + return + events.append("message-start") + await send({"type": "message.ack"}) + events.append("message-acked") + message_acked.set() + await release_message.wait() + events.append("message-end") + + stub_client = _StubClient(_StubSubscription(events), events) + monkeypatch.setattr(stompman, "Client", lambda **kwargs: stub_client) + + server = Server( + app, + "destination", + host="localhost", + port=61613, + login="guest", + passcode="guest", + message_send=_StubMessageSendManager(), + ) + serve_task = asyncio.create_task(server.serve()) + await asyncio.wait_for(stub_client.subscribed.wait(), 5) + + assert stub_client.handler is not None + await stub_client.handler(_stub_frame(events)) + await asyncio.wait_for(message_acked.wait(), 5) + + server.stop() + # serve() must now be draining the in-flight handler, which is still blocked + # on release_message, so the subscription cannot have been unsubscribed yet + for _ in range(10): + await asyncio.sleep(0) + assert "unsubscribe" not in events + + release_message.set() + await asyncio.wait_for(serve_task, 5) + + assert events.index("ack") < events.index("unsubscribe") + assert events.index("message-end") < events.index("unsubscribe") + assert stub_client.exit_args == (None, None, None) + + +async def test_process_message_nacks_when_application_raises( + caplog: pytest.LogCaptureFixture, +) -> None: + async def app( + scope: Scope, receive: AMGIReceiveCallable, send: AMGISendCallable + ) -> None: + raise RuntimeError("application error") + + server = Server( + app, + "destination", + host="localhost", + port=61613, + login="guest", + passcode="guest", + ) + ack = AsyncMock() + nack = AsyncMock() + frame = _stub_frame(ack=ack, nack=nack) + + with caplog.at_level(logging.ERROR, logger="amgi-stomp.error"): + await server._process_message(frame, AsyncMock(), {}) + + nack.assert_awaited_once_with() + ack.assert_not_awaited() + assert "application error" in caplog.text + + +async def test_process_message_suppresses_nack_failure() -> None: + async def app( + scope: Scope, receive: AMGIReceiveCallable, send: AMGISendCallable + ) -> None: + raise RuntimeError("application error") + + server = Server( + app, + "destination", + host="localhost", + port=61613, + login="guest", + passcode="guest", + ) + nack = AsyncMock(side_effect=RuntimeError("nack failed")) + frame = _stub_frame(nack=nack) + + await server._process_message(frame, AsyncMock(), {}) + + nack.assert_awaited_once_with() + + +async def test_process_message_propagates_cancellation() -> None: + async def app( + scope: Scope, receive: AMGIReceiveCallable, send: AMGISendCallable + ) -> None: + raise asyncio.CancelledError + + server = Server( + app, + "destination", + host="localhost", + port=61613, + login="guest", + passcode="guest", + ) + ack = AsyncMock() + nack = AsyncMock() + frame = _stub_frame(ack=ack, nack=nack) + + with pytest.raises(asyncio.CancelledError): + await server._process_message(frame, AsyncMock(), {}) + + nack.assert_not_awaited() + ack.assert_not_awaited() + + +async def test_process_message_scope_bindings() -> None: + scopes: list[MessageScope] = [] + + async def app( + scope: Scope, receive: AMGIReceiveCallable, send: AMGISendCallable + ) -> None: + scopes.append(cast(MessageScope, scope)) + + server = Server( + app, + "destination", + host="localhost", + port=61613, + login="guest", + passcode="guest", + ) + frame = cast( + AckableMessageFrame, + SimpleNamespace( + headers={ + "destination": "destination", + "message-id": "message-id-1", + "subscription": "subscription-1", + "custom": "value", + }, + body=b"value", + ack=AsyncMock(), + nack=AsyncMock(), + ), + ) + + await server._process_message(frame, AsyncMock(), {}) + + assert scopes[0]["bindings"] == { + "stomp": {"message_id": "message-id-1", "subscription": "subscription-1"} + } + assert dict(scopes[0]["headers"]) == {b"custom": b"value"} + + +async def test_serve_ignores_messages_after_stop( + monkeypatch: pytest.MonkeyPatch, +) -> None: + events: list[str] = [] + handled: list[str] = [] + + async def app( + scope: Scope, receive: AMGIReceiveCallable, send: AMGISendCallable + ) -> None: + if scope["type"] == "lifespan": + await _lifespan_app_call(scope, receive, send) + return + handled.append("message") # pragma: no cover + + stub_client = _StubClient(_StubSubscription(events), events) + monkeypatch.setattr(stompman, "Client", lambda **kwargs: stub_client) + + server = Server( + app, + "destination", + host="localhost", + port=61613, + login="guest", + passcode="guest", + message_send=_StubMessageSendManager(), + ) + serve_task = asyncio.create_task(server.serve()) + await asyncio.wait_for(stub_client.subscribed.wait(), 5) + + server.stop() + for _ in range(10): + await asyncio.sleep(0) + assert stub_client.handler is not None + ack = AsyncMock() + await stub_client.handler(_stub_frame(ack=ack)) + await asyncio.wait_for(serve_task, 5) + + assert handled == [] + ack.assert_not_awaited() + + +async def test_serve_cancels_handlers_after_shutdown_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import amgi_stomp + + monkeypatch.setattr(amgi_stomp, "_SHUTDOWN_TIMEOUT", 0.05) + events: list[str] = [] + started = asyncio.Event() + cancelled = asyncio.Event() + + async def app( + scope: Scope, receive: AMGIReceiveCallable, send: AMGISendCallable + ) -> None: + if scope["type"] == "lifespan": + await _lifespan_app_call(scope, receive, send) + return + started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelled.set() + raise + + stub_client = _StubClient(_StubSubscription(events), events) + monkeypatch.setattr(stompman, "Client", lambda **kwargs: stub_client) + + server = Server( + app, + "destination", + host="localhost", + port=61613, + login="guest", + passcode="guest", + message_send=_StubMessageSendManager(), + ) + serve_task = asyncio.create_task(server.serve()) + await asyncio.wait_for(stub_client.subscribed.wait(), 5) + assert stub_client.handler is not None + await stub_client.handler(_stub_frame(events)) + await asyncio.wait_for(started.wait(), 5) + + server.stop() + await asyncio.wait_for(serve_task, 5) + + assert cancelled.is_set() + assert "unsubscribe" in events diff --git a/packages/asyncfast/src/asyncfast/__init__.py b/packages/asyncfast/src/asyncfast/__init__.py index e18a89d9..7ea2ee24 100644 --- a/packages/asyncfast/src/asyncfast/__init__.py +++ b/packages/asyncfast/src/asyncfast/__init__.py @@ -8,6 +8,8 @@ from asyncfast._message import Message from asyncfast._utils import ChannelNotFoundError from asyncfast._utils import InvalidChannelDefinitionError +from asyncfast.bindings import StompMessageId +from asyncfast.bindings import StompSubscription __all__ = [ "AsyncFast", @@ -20,4 +22,6 @@ "Payload", "Message", "ChannelNotFoundError", + "StompMessageId", + "StompSubscription", ] diff --git a/packages/asyncfast/src/asyncfast/bindings.py b/packages/asyncfast/src/asyncfast/bindings.py index 94efb0a3..72c81b81 100644 --- a/packages/asyncfast/src/asyncfast/bindings.py +++ b/packages/asyncfast/src/asyncfast/bindings.py @@ -38,3 +38,13 @@ class SqsMessageDeduplicationId(Binding): # type: ignore[misc] class SqsMessageGroupId(Binding): # type: ignore[misc] __protocol__ = "sqs" __field_name__ = "message_group_id" + + +class StompMessageId(Binding): # type: ignore[misc] + __protocol__ = "stomp" + __field_name__ = "message_id" + + +class StompSubscription(Binding): # type: ignore[misc] + __protocol__ = "stomp" + __field_name__ = "subscription" diff --git a/packages/asyncfast/tests_asyncfast/test_message.py b/packages/asyncfast/tests_asyncfast/test_message.py index 8d7fd3b7..560fff6c 100644 --- a/packages/asyncfast/tests_asyncfast/test_message.py +++ b/packages/asyncfast/tests_asyncfast/test_message.py @@ -21,6 +21,8 @@ from asyncfast import Payload from asyncfast.bindings import KafkaKey from asyncfast.bindings import SqsMessageGroupId +from asyncfast.bindings import StompMessageId +from asyncfast.bindings import StompSubscription from pydantic import BaseModel @@ -765,6 +767,89 @@ async def topic_handler(group_id: Annotated[int, SqsMessageGroupId()]) -> None: test_mock.assert_called_once_with(1234) +async def test_message_binding_stomp_message_id() -> None: + app = AsyncFast() + + test_mock = Mock() + + @app.channel("topic") + async def topic_handler(message_id: Annotated[str, StompMessageId()]) -> None: + test_mock(message_id) + + message_scope: MessageScope = { + "type": "message", + "amgi": {"version": "2.0", "spec_version": "2.0"}, + "address": "topic", + "headers": [], + "bindings": {"stomp": {"message_id": "message-id-1"}}, + } + await app( + message_scope, + AsyncMock(), + AsyncMock(), + ) + + test_mock.assert_called_once_with("message-id-1") + + +async def test_message_binding_stomp_subscription() -> None: + app = AsyncFast() + + test_mock = Mock() + + @app.channel("topic") + async def topic_handler(subscription: Annotated[str, StompSubscription()]) -> None: + test_mock(subscription) + + message_scope: MessageScope = { + "type": "message", + "amgi": {"version": "2.0", "spec_version": "2.0"}, + "address": "topic", + "headers": [], + "bindings": {"stomp": {"subscription": "subscription-1"}}, + } + await app( + message_scope, + AsyncMock(), + AsyncMock(), + ) + + test_mock.assert_called_once_with("subscription-1") + + +@pytest.mark.parametrize( + ["bindings", "expected_value"], + (({}, None), ({"stomp": {"message_id": "message-id-1"}}, "message-id-1")), +) +async def test_message_binding_default_stomp_message_id( + bindings: dict[str, Any], expected_value: Optional[str] +) -> None: + app = AsyncFast() + + test_mock = Mock() + + @app.channel("topic") + async def topic_handler( + message_id: Annotated[Optional[str], StompMessageId()] = None, + ) -> None: + test_mock(message_id) + + message_scope: MessageScope = { + "type": "message", + "amgi": {"version": "2.0", "spec_version": "2.0"}, + "address": "topic", + "headers": [], + "bindings": bindings, + } + await app( + message_scope, + AsyncMock(), + AsyncMock(), + ) + + test_mock.assert_called_once_with(expected_value) + + @pytest.mark.parametrize( ["bindings", "expected_key"], (({}, None), ({"kafka": {"key": b"1234"}}, 1234)) ) diff --git a/tox.ini b/tox.ini index 39110a84..d5c2a6ba 100644 --- a/tox.ini +++ b/tox.ini @@ -6,8 +6,10 @@ env_list = py313-asyncfast-pydantic2{8-12} clean py3{10-14}-{amgi-aio-pika, amgi-aiobotocore, amgi-aiokafka, amgi-cloudevents, amgi-common, amgi-kafka-event-source-mapping, amgi-nats, amgi-paho-mqtt, amgi-redis, amgi-sqs-event-source-mapping, asyncfast-avro, asyncfast-cli, pytest-amgi} + py3{11-14}-amgi-stomp py3{10-12}-asyncfast-pydantic2{0-12} py3{10-14}-{amgi-aio-pika, amgi-aiobotocore, amgi-aiokafka, amgi-cloudevents, amgi-common, amgi-kafka-event-source-mapping, amgi-nats, amgi-paho-mqtt, amgi-redis, amgi-sqs-event-source-mapping, amgi-types, asyncfast, asyncfast-avro, asyncfast-cli, pytest-amgi}-import + py3{11-14}-amgi-stomp-import [testenv] runner = uv-venv-lock-runner @@ -48,6 +50,12 @@ commands = labels = clean uv_sync_flags = --all-packages +[testenv:py3{11-14}-amgi-stomp] +commands = + {[testenv]commands} packages/amgi-stomp +labels = amgi-stomp +uv_sync_flags = --package=amgi-stomp + [testenv:py3{10-12}-asyncfast-pydantic2{0-12}] commands_pre = {[pydantic2]commands_pre} @@ -56,6 +64,14 @@ commands = labels = asyncfast uv_sync_flags = --package=asyncfast +[testenv:py3{11-14}-amgi-stomp-import] +commands = + python -c "import amgi_stomp" +labels = amgi-stomp +uv_sync_flags = + --package=amgi-stomp + --no-dev + [testenv:py3{10-14}-amgi-aio-pika] commands = {[testenv]commands} packages/amgi-aio-pika diff --git a/uv.lock b/uv.lock index 5a541b64..10cd5a5f 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 5 requires-python = ">=3.10" resolution-markers = [ "python_full_version >= '3.15'", @@ -22,6 +22,7 @@ members = [ "amgi-paho-mqtt", "amgi-redis", "amgi-sqs-event-source-mapping", + "amgi-stomp", "amgi-types", "asyncfast", "asyncfast-avro", @@ -794,6 +795,43 @@ dev = [ { name = "test-utils", editable = "packages/test-utils" }, ] +[[package]] +name = "amgi-stomp" +version = "0.49.0" +source = { editable = "packages/amgi-stomp" } +dependencies = [ + { name = "amgi-common" }, + { name = "amgi-types" }, + { name = "stompman", marker = "python_full_version >= '3.11'" }, +] + +[package.dev-dependencies] +dev = [ + { name = "pytest" }, + { name = "pytest-asyncio" }, + { name = "pytest-cov" }, + { name = "pytest-timeout" }, + { name = "test-utils" }, + { name = "testcontainers" }, +] + +[package.metadata] +requires-dist = [ + { name = "amgi-common", editable = "packages/amgi-common" }, + { name = "amgi-types", editable = "packages/amgi-types" }, + { name = "stompman", marker = "python_full_version >= '3.11'", specifier = ">=3.16.0" }, +] + +[package.metadata.requires-dev] +dev = [ + { name = "pytest", specifier = ">=8.4.1" }, + { name = "pytest-asyncio", specifier = ">=1.3.0" }, + { name = "pytest-cov", specifier = ">=7.0.0" }, + { name = "pytest-timeout", specifier = ">=2.4.0" }, + { name = "test-utils", editable = "packages/test-utils" }, + { name = "testcontainers", specifier = ">=4.13.0" }, +] + [[package]] name = "amgi-types" version = "0.49.0" @@ -1687,6 +1725,7 @@ dependencies = [ { name = "amgi-aiokafka" }, { name = "amgi-nats" }, { name = "amgi-paho-mqtt" }, + { name = "amgi-stomp" }, { name = "asyncfast" }, ] @@ -1696,6 +1735,7 @@ requires-dist = [ { name = "amgi-aiokafka", editable = "packages/amgi-aiokafka" }, { name = "amgi-nats", editable = "packages/amgi-nats" }, { name = "amgi-paho-mqtt", editable = "packages/amgi-paho-mqtt" }, + { name = "amgi-stomp", editable = "packages/amgi-stomp" }, { name = "asyncfast", editable = "packages/asyncfast" }, ] @@ -4252,6 +4292,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ec/bb/2799cc2ede3ed41131f8975621e7213dfc7ef4acbbaadfa440f32500c370/starlette-1.3.1-py3-none-any.whl", hash = "sha256:c7372aae11c3c3f26a42df7bd626cec2f47d03483d261d369516a615a53714c6", size = 73632, upload-time = "2026-06-12T09:23:10.017Z" }, ] +[[package]] +name = "stompman" +version = "3.16.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d0/50/ae169a5a09729adedea5bf1a1a8bd939493e6a5c1cd7f8544750d4b8f4a6/stompman-3.16.0.tar.gz", hash = "sha256:700476a8c9428f8c3cb1bc0211a6e80c7248db6dca09992e6707350c2a5332d3", size = 21841, upload-time = "2026-09-17T14:51:49.579Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e6/0e/1cc4ec247e460a7b469a2d734f1c518b5770a888a72ac19279c080efee24/stompman-3.16.0-py3-none-any.whl", hash = "sha256:8d8b755043e31024ee7f1d5dfe773ba61cd410dc43dd4ea861fa26e57556e93b", size = 29647, upload-time = "2026-09-17T14:51:48.227Z" }, +] + [[package]] name = "sympy" version = "1.14.0"