Merge branch 'main' into fix-acks-on-activemq-classic

Author: vrslev

README.md

@@ -127,7 +127,7 @@ stompman takes care of cleaning up resources automatically. When you leave the c
 Also, I want to pointed out that:
 
 - Protocol parsing is inspired by [aiostomp](https://github.com/pedrokiefer/aiostomp/blob/3449dcb53f43e5956ccc7662bb5b7d76bc6ef36b/aiostomp/protocol.py) (meaning: consumed by me and refactored from).
-- stompman is tested and used with [Artemis ActiveMQ](https://activemq.apache.org/components/artemis/).
+- stompman is tested and used with [ActiveMQ Artemis](https://activemq.apache.org/components/artemis/) and [ActiveMQ Classic](https://activemq.apache.org/components/classic/).
 - Specification says that headers in CONNECT and CONNECTED frames shouldn't be escaped for backwards compatibility. stompman escapes headers in CONNECT frame (outcoming), but does not unescape headers in CONNECTED (outcoming).
 
 ### Examples

docker-compose.yml

@@ -5,16 +5,22 @@ services:
       args:
         PYTHON_IMAGE: ${PYTHON_IMAGE:-python:3.13-slim-bullseye}
     depends_on:
-      artemis:
+      activemq-artemis:
+        condition: service_started
+      activemq-classic:
         condition: service_started
-    environment:
-      ARTEMIS_HOST: artemis
 
-  artemis:
+  activemq-artemis:
     image: apache/activemq-artemis:2.37.0-alpine
     environment:
       ARTEMIS_USER: admin
       ARTEMIS_PASSWORD: ":=123"
     ports:
       - 8161:8161
       - 61616:61616
+
+  activemq-classic:
+    image: apache/activemq-classic:6.1.2
+    environment:
+      ACTIVEMQ_CONNECTION_USER: admin
+      ACTIVEMQ_CONNECTION_PASSWORD: ":=123"

examples/consumer.py

@@ -1,14 +1,8 @@
 import asyncio
-import os
 
 import stompman
 
-server = stompman.ConnectionParameters(
-    host=os.environ.get("ARTEMIS_HOST", "0.0.0.0"),  # noqa: S104
-    port=61616,
-    login="admin",
-    passcode=":=123",
-)
+server = stompman.ConnectionParameters(host="0.0.0.0", port=61616, login="admin", passcode=":=123")  # noqa: S104
 
 
 async def handle_message(message_frame: stompman.MessageFrame) -> None:

examples/producer.py

@@ -1,14 +1,8 @@
 import asyncio
-import os
 
 import stompman
 
-server = stompman.ConnectionParameters(
-    host=os.environ.get("ARTEMIS_HOST", "0.0.0.0"),  # noqa: S104
-    port=61616,
-    login="admin",
-    passcode=":=123",
-)
+server = stompman.ConnectionParameters(host="0.0.0.0", port=61616, login="admin", passcode=":=123")  # noqa: S104
 
 
 async def main() -> None:

stompman/client.py

@@ -1,12 +1,12 @@
 import asyncio
 import inspect
-from collections.abc import AsyncGenerator, Awaitable, Callable, Coroutine
+from collections.abc import AsyncGenerator, Awaitable, Callable
 from contextlib import AsyncExitStack, asynccontextmanager
 from dataclasses import dataclass, field
 from functools import partial
 from ssl import SSLContext
 from types import TracebackType
-from typing import ClassVar, Literal, Self
+from typing import Any, ClassVar, Literal, Self
 
 from stompman.config import ConnectionParameters, Heartbeat
 from stompman.connection import AbstractConnection, Connection
@@ -30,8 +30,8 @@ class Client:
     PROTOCOL_VERSION: ClassVar = "1.2"  # https://stomp.github.io/stomp-specification-1.2.html
 
     servers: list[ConnectionParameters] = field(kw_only=False)
-    on_error_frame: Callable[[ErrorFrame], None] | None = None
-    on_heartbeat: Callable[[], None] | Callable[[], Awaitable[None]] | None = None
+    on_error_frame: Callable[[ErrorFrame], Any] | None = None
+    on_heartbeat: Callable[[], Any] | Callable[[], Awaitable[Any]] | None = None
 
     heartbeat: Heartbeat = field(default=Heartbeat(1000, 1000))
     ssl: Literal[True] | SSLContext | None = None
@@ -146,11 +146,11 @@ async def begin(self) -> AsyncGenerator[Transaction, None]:
     async def subscribe(
         self,
         destination: str,
-        handler: Callable[[MessageFrame], Coroutine[None, None, None]],
+        handler: Callable[[MessageFrame], Awaitable[Any]],
         *,
         ack: AckMode = "client-individual",
         headers: dict[str, str] | None = None,
-        on_suppressed_exception: Callable[[Exception, MessageFrame], None],
+        on_suppressed_exception: Callable[[Exception, MessageFrame], Any],
         suppressed_exception_classes: tuple[type[Exception], ...] = (Exception,),
     ) -> "Subscription":
         subscription = Subscription(

stompman/connection_lifespan.py

@@ -2,7 +2,7 @@
 from collections.abc import Callable
 from contextlib import suppress
 from dataclasses import dataclass
-from typing import Protocol
+from typing import Any, Protocol
 from uuid import uuid4
 
 from stompman.config import ConnectionParameters, Heartbeat
@@ -37,7 +37,7 @@ class ConnectionLifespan(AbstractConnectionLifespan):
     disconnect_confirmation_timeout: int
     active_subscriptions: ActiveSubscriptions
     active_transactions: ActiveTransactions
-    set_heartbeat_interval: Callable[[float], None]
+    set_heartbeat_interval: Callable[[float], Any]
 
     async def _establish_connection(self) -> StompProtocolConnectionIssue | None:
         await self.connection.write_frame(

stompman/subscription.py

@@ -1,5 +1,6 @@
-from collections.abc import Callable, Coroutine
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
+from typing import Any
 from uuid import uuid4
 
 from stompman.connection import AbstractConnection
@@ -21,9 +22,9 @@ class Subscription:
     id: str = field(default_factory=lambda: _make_subscription_id(), init=False)  # noqa: PLW0108
     destination: str
     headers: dict[str, str] | None
-    handler: Callable[[MessageFrame], Coroutine[None, None, None]]
+    handler: Callable[[MessageFrame], Awaitable[Any]]
     ack: AckMode
-    on_suppressed_exception: Callable[[Exception, MessageFrame], None]
+    on_suppressed_exception: Callable[[Exception, MessageFrame], Any]
     suppressed_exception_classes: tuple[type[Exception], ...]
     _connection_manager: ConnectionManager
     _active_subscriptions: ActiveSubscriptions

tests/integration.py

@@ -1,9 +1,8 @@
 import asyncio
-import os
 from collections.abc import AsyncGenerator, Callable
 from contextlib import asynccontextmanager
 from itertools import starmap
-from typing import Final
+from typing import Final, cast
 from uuid import uuid4
 
 import pytest
@@ -21,23 +20,29 @@
     parse_header,
 )
 
-pytestmark = pytest.mark.anyio
+DESTINATION: Final = "DLQ"
+
 
-CONNECTION_PARAMETERS: Final = stompman.ConnectionParameters(
-    host=os.environ["ARTEMIS_HOST"], port=61616, login="admin", passcode=":=123"
+@pytest.fixture(
+    params=[
+        stompman.ConnectionParameters(host="activemq-artemis", port=61616, login="admin", passcode=":=123"),
+        stompman.ConnectionParameters(host="activemq-classic", port=61613, login="admin", passcode=":=123"),
+    ]
 )
-DESTINATION: Final = "DLQ"
+def connection_parameters(request: pytest.FixtureRequest) -> stompman.ConnectionParameters:
+    return cast(stompman.ConnectionParameters, request.param)
 
 
 @asynccontextmanager
-async def create_client() -> AsyncGenerator[stompman.Client, None]:
+async def create_client(connection_parameters: stompman.ConnectionParameters) -> AsyncGenerator[stompman.Client, None]:
     async with stompman.Client(
-        servers=[CONNECTION_PARAMETERS], read_timeout=10, connection_confirmation_timeout=10
+        servers=[connection_parameters], read_timeout=10, connection_confirmation_timeout=10
     ) as client:
         yield client
 
 
-async def test_ok() -> None:
+@pytest.mark.anyio
+async def test_ok(connection_parameters: stompman.ConnectionParameters) -> None:
     async def produce() -> None:
         for message in messages[200:]:
             await producer.send(body=message, destination=DESTINATION, headers={"hello": "from outside transaction"})
@@ -65,7 +70,11 @@ async def handle_message(frame: stompman.MessageFrame) -> None:  # noqa: RUF029
 
     messages = [str(uuid4()).encode() for _ in range(10000)]
 
-    async with create_client() as consumer, create_client() as producer, asyncio.TaskGroup() as task_group:
+    async with (
+        create_client(connection_parameters) as consumer,
+        create_client(connection_parameters) as producer,
+        asyncio.TaskGroup() as task_group,
+    ):
         task_group.create_task(consume())
         task_group.create_task(produce())