Remove read buffer based on message frame buffer to be able to read heartbeats and system frames—reverts #135

README.md

@@ -40,7 +40,6 @@ async with stompman.Client(
     disconnect_confirmation_timeout=2,
     write_retry_attempts=3,
     check_server_alive_interval_factor=3,
-    max_concurrent_consumed_messages=10,
 ) as client:
     ...
 ```

packages/faststream-stomp/test_faststream_stomp/test_integration.py

@@ -1,5 +1,7 @@
 import asyncio
 import logging
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Annotated
 from unittest import mock
@@ -27,6 +29,15 @@ def broker(first_server_connection_parameters: stompman.ConnectionParameters) ->
     return faststream_stomp.StompBroker(stompman.Client([first_server_connection_parameters]))
 
 
+@asynccontextmanager
+async def run_faststream_app(application: FastStream) -> AsyncGenerator[None, None]:
+    async with asyncio.TaskGroup() as task_group:
+        run_task = task_group.create_task(application.run())
+        yield
+        application.exit()
+        await asyncio.wait([run_task])
+
+
 async def test_simple(faker: faker.Faker, broker: faststream_stomp.StompBroker) -> None:
     app = FastStream(broker)
     expected_body, destination = faker.pystr(), faker.pystr()
@@ -43,10 +54,8 @@ async def _() -> None:
         await broker.connect()
         await publisher.publish(expected_body.encode(), correlation_id=gen_cor_id())
 
-    async with asyncio.timeout(10), asyncio.TaskGroup() as task_group:
-        run_task = task_group.create_task(app.run())
+    async with asyncio.timeout(10), run_faststream_app(app):
         await event.wait()
-        run_task.cancel()
 
 
 async def test_republish(faker: faker.Faker, broker: faststream_stomp.StompBroker) -> None:
@@ -71,10 +80,8 @@ async def _() -> None:
         await broker.connect()
         await first_publisher.publish(expected_body.encode())
 
-    async with asyncio.timeout(10), asyncio.TaskGroup() as task_group:
-        run_task = task_group.create_task(app.run())
+    async with asyncio.timeout(10), run_faststream_app(app):
         await event.wait()
-        run_task.cancel()
 
 
 async def test_router(faker: faker.Faker, broker: faststream_stomp.StompBroker) -> None:
@@ -96,10 +103,8 @@ async def _() -> None:
         await broker.connect()
         await publisher.publish(expected_body)
 
-    async with asyncio.timeout(1), asyncio.TaskGroup() as task_group:
-        run_task = task_group.create_task(app.run())
+    async with asyncio.timeout(1), run_faststream_app(app):
         await event.wait()
-        run_task.cancel()
 
 
 async def test_broker_close(broker: faststream_stomp.StompBroker) -> None:

packages/stompman/stompman/client.py

@@ -42,17 +42,15 @@ class Client:
     disconnect_confirmation_timeout: int = 2
     check_server_alive_interval_factor: int = 3
     """Client will check if server alive `server heartbeat interval` times `interval factor`"""
-    max_concurrent_consumed_messages: int = 10
 
     connection_class: type[AbstractConnection] = Connection
 
     _connection_manager: ConnectionManager = field(init=False)
-    _active_subscriptions: ActiveSubscriptions = field(default_factory=ActiveSubscriptions, init=False, repr=False)
+    _active_subscriptions: ActiveSubscriptions = field(default_factory=ActiveSubscriptions, init=False)
     _active_transactions: set[Transaction] = field(default_factory=set, init=False)
     _exit_stack: AsyncExitStack = field(default_factory=AsyncExitStack, init=False)
     _listen_task: asyncio.Task[None] = field(init=False, repr=False)
     _task_group: asyncio.TaskGroup = field(init=False, repr=False)
-    _message_frame_semaphore: asyncio.Semaphore = field(init=False, repr=False)
 
     def __post_init__(self) -> None:
         self._connection_manager = ConnectionManager(
@@ -75,7 +73,6 @@ def __post_init__(self) -> None:
             check_server_alive_interval_factor=self.check_server_alive_interval_factor,
             ssl=self.ssl,
         )
-        self._message_frame_semaphore = asyncio.Semaphore(self.max_concurrent_consumed_messages)
 
     async def __aenter__(self) -> Self:
         self._task_group = await self._exit_stack.enter_async_context(asyncio.TaskGroup())
@@ -99,17 +96,16 @@ async def _listen_to_frames(self) -> None:
             async for frame in self._connection_manager.read_frames_reconnecting():
                 match frame:
                     case MessageFrame():
-                        if not (subscription := self._active_subscriptions.get_by_id(frame.headers["subscription"])):
-                            continue
-                        await self._message_frame_semaphore.acquire()
-                        created_task = task_group.create_task(
-                            subscription._run_handler(frame=frame)  # noqa: SLF001
-                            if isinstance(subscription, AutoAckSubscription)
-                            else subscription.handler(
-                                AckableMessageFrame(headers=frame.headers, body=frame.body, _subscription=subscription)
+                        if subscription := self._active_subscriptions.get_by_id(frame.headers["subscription"]):
+                            task_group.create_task(
+                                subscription._run_handler(frame=frame)  # noqa: SLF001
+                                if isinstance(subscription, AutoAckSubscription)
+                                else subscription.handler(
+                                    AckableMessageFrame(
+                                        headers=frame.headers, body=frame.body, _subscription=subscription
+                                    )
+                                )
                             )
-                        )
-                        created_task.add_done_callback(lambda _: self._message_frame_semaphore.release())
                     case ErrorFrame():
                         if self.on_error_frame:
                             self.on_error_frame(frame)