Merge main

Author: vrslev

.github/dependabot.yml

@@ -4,8 +4,16 @@ updates:
     directory: /
     schedule:
       interval: monthly
+    groups:
+      github-actions:
+        patterns:
+          - "*"
 
   - package-ecosystem: pip
     directory: /
     schedule:
       interval: monthly
+    groups:
+      pip:
+        patterns:
+          - "*"

.github/workflows/publish.yml

@@ -9,7 +9,7 @@ jobs:
   publish:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
       - uses: extractions/setup-just@v3
       - uses: astral-sh/setup-uv@v6
         with:

.github/workflows/test.yml

@@ -15,7 +15,7 @@ jobs:
   check-types:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
       - uses: extractions/setup-just@v3
       - uses: astral-sh/setup-uv@v6
         with:
@@ -25,7 +25,7 @@ jobs:
   lint:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
       - uses: extractions/setup-just@v3
       - uses: astral-sh/setup-uv@v6
         with:
@@ -42,7 +42,7 @@ jobs:
           - "3.12"
           - "3.13"
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
       - uses: extractions/setup-just@v3
       - uses: astral-sh/setup-uv@v6
         with:

packages/stompman/stompman/client.py

@@ -20,6 +20,7 @@
     ReceiptFrame,
     SendFrame,
 )
+from stompman.logger import LOGGER
 from stompman.subscription import AckableMessageFrame, ActiveSubscriptions, AutoAckSubscription, ManualAckSubscription
 from stompman.transaction import Transaction
 
@@ -29,7 +30,9 @@ 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], Any] | None = None
+    on_error_frame: Callable[[ErrorFrame], Any] | None = lambda error_frame: LOGGER.error(
+        "received error frame: %s", error_frame
+    )
 
     heartbeat: Heartbeat = field(default=Heartbeat(1000, 1000))
     ssl: Literal[True] | SSLContext | None = None

packages/stompman/stompman/connection.py

@@ -37,7 +37,7 @@ def _reraise_connection_lost(*causes: type[Exception]) -> Generator[None, None,
     try:
         yield
     except causes as exception:
-        raise ConnectionLostError from exception
+        raise ConnectionLostError(reason=exception) from exception
 
 
 @dataclass(kw_only=True, slots=True)
@@ -85,10 +85,8 @@ async def write_frame(self, frame: AnyClientFrame) -> None:
             await self.writer.drain()
 
     async def _read_non_empty_bytes(self, max_chunk_size: int) -> bytes:
-        while (  # noqa: ASYNC110
-            chunk := await self.reader.read(max_chunk_size)
-        ) == b"":  # pragma: no cover (it definitely happens)
-            await asyncio.sleep(0)
+        if (chunk := await self.reader.read(max_chunk_size)) == b"":
+            raise ConnectionLostError(reason="eof")
         return chunk
 
     async def read_frames(self) -> AsyncGenerator[AnyServerFrame, None]:

packages/stompman/stompman/connection_manager.py

@@ -55,21 +55,18 @@ class ConnectionManager:
     _reconnect_lock: asyncio.Lock = field(init=False, default_factory=asyncio.Lock)
     _task_group: asyncio.TaskGroup = field(init=False, default_factory=asyncio.TaskGroup)
     _send_heartbeat_task: asyncio.Task[None] = field(init=False, repr=False)
-    _check_server_heartbeat_task: asyncio.Task[None] = field(init=False, repr=False)
 
     async def __aenter__(self) -> Self:
         await self._task_group.__aenter__()
         self._send_heartbeat_task = self._task_group.create_task(asyncio.sleep(0))
-        self._check_server_heartbeat_task = self._task_group.create_task(asyncio.sleep(0))
         self._active_connection_state = await self._get_active_connection_state(is_initial_call=True)
         return self
 
     async def __aexit__(
         self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None
     ) -> None:
         self._send_heartbeat_task.cancel()
-        self._check_server_heartbeat_task.cancel()
-        await asyncio.wait([self._send_heartbeat_task, self._check_server_heartbeat_task])
+        await asyncio.wait([self._send_heartbeat_task])
         await self._task_group.__aexit__(exc_type, exc_value, traceback)
 
         if not self._active_connection_state:
@@ -82,29 +79,16 @@ async def __aexit__(
 
     def _restart_heartbeat_tasks(self, server_heartbeat: Heartbeat) -> None:
         self._send_heartbeat_task.cancel()
-        self._check_server_heartbeat_task.cancel()
         self._send_heartbeat_task = self._task_group.create_task(
             self._send_heartbeats_forever(server_heartbeat.want_to_receive_interval_ms)
         )
-        self._check_server_heartbeat_task = self._task_group.create_task(
-            self._check_server_heartbeat_forever(server_heartbeat.will_send_interval_ms)
-        )
 
     async def _send_heartbeats_forever(self, send_heartbeat_interval_ms: int) -> None:
         send_heartbeat_interval_seconds = send_heartbeat_interval_ms / 1000
         while True:
             await self.write_heartbeat_reconnecting()
             await asyncio.sleep(send_heartbeat_interval_seconds)
 
-    async def _check_server_heartbeat_forever(self, receive_heartbeat_interval_ms: int) -> None:
-        receive_heartbeat_interval_seconds = receive_heartbeat_interval_ms / 1000
-        while True:
-            await asyncio.sleep(receive_heartbeat_interval_seconds * self.check_server_alive_interval_factor)
-            if not self._active_connection_state:
-                continue
-            if not self._active_connection_state.is_alive(self.check_server_alive_interval_factor):
-                self._clear_active_connection_state()
-
     async def _create_connection_to_one_server(
         self, server: ConnectionParameters
     ) -> tuple[AbstractConnection, ConnectionParameters] | None:
@@ -168,7 +152,7 @@ async def _get_active_connection_state(self, *, is_initial_call: bool = False) -
                     self._active_connection_state = connection_result
                     if not is_initial_call:
                         LOGGER.warning(
-                            "reconnected after failure connection failure. connection_parameters: %s",
+                            "reconnected after connection failure. connection_parameters: %s",
                             connection_result.lifespan.connection_parameters,
                         )
                     return connection_result
@@ -178,11 +162,12 @@ async def _get_active_connection_state(self, *, is_initial_call: bool = False) -
 
         raise FailedAllConnectAttemptsError(retry_attempts=self.connect_retry_attempts, issues=connection_issues)
 
-    def _clear_active_connection_state(self) -> None:
+    def _clear_active_connection_state(self, error_reason: ConnectionLostError) -> None:
         if not self._active_connection_state:
             return
         LOGGER.warning(
-            "connection lost. connection_parameters: %s",
+            "connection lost. reason: %r, connection_parameters: %s",
+            error_reason.reason,
             self._active_connection_state.lifespan.connection_parameters,
         )
         self._active_connection_state = None
@@ -192,8 +177,8 @@ async def write_heartbeat_reconnecting(self) -> None:
             connection_state = await self._get_active_connection_state()
             try:
                 return connection_state.connection.write_heartbeat()
-            except ConnectionLostError:
-                self._clear_active_connection_state()
+            except ConnectionLostError as error:
+                self._clear_active_connection_state(error)
 
         raise FailedAllWriteAttemptsError(retry_attempts=self.write_retry_attempts)
 
@@ -202,8 +187,8 @@ async def write_frame_reconnecting(self, frame: AnyClientFrame) -> None:
             connection_state = await self._get_active_connection_state()
             try:
                 return await connection_state.connection.write_frame(frame)
-            except ConnectionLostError:
-                self._clear_active_connection_state()
+            except ConnectionLostError as error:
+                self._clear_active_connection_state(error)
 
         raise FailedAllWriteAttemptsError(retry_attempts=self.write_retry_attempts)
 
@@ -213,15 +198,15 @@ async def read_frames_reconnecting(self) -> AsyncGenerator[AnyServerFrame, None]
             try:
                 async for frame in connection_state.connection.read_frames():
                     yield frame
-            except ConnectionLostError:
-                self._clear_active_connection_state()
+            except ConnectionLostError as error:
+                self._clear_active_connection_state(error)
 
     async def maybe_write_frame(self, frame: AnyClientFrame) -> bool:
         if not self._active_connection_state:
             return False
         try:
             await self._active_connection_state.connection.write_frame(frame)
-        except ConnectionLostError:
-            self._clear_active_connection_state()
+        except ConnectionLostError as error:
+            self._clear_active_connection_state(error)
             return False
         return True

packages/stompman/stompman/errors.py

@@ -14,6 +14,8 @@ def __str__(self) -> str:
 class ConnectionLostError(Error):
     """Raised in stompman.AbstractConnection—and handled in stompman.ConnectionManager, therefore is private."""
 
+    reason: Exception | str
+
 
 @dataclass(frozen=True, kw_only=True, slots=True)
 class ConnectionConfirmationTimeout:

packages/stompman/stompman/serde.py

@@ -1,8 +1,7 @@
 import struct
-from collections import deque
 from collections.abc import Iterator
 from contextlib import suppress
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Any, Final, cast
 
 from stompman.frames import (
@@ -141,53 +140,83 @@ def make_frame_from_parts(*, command: bytes, headers: dict[str, str], body: byte
     return frame_type(headers=headers_, body=body) if frame_type in FRAMES_WITH_BODY else frame_type(headers=headers_)  # type: ignore[call-arg]
 
 
-def parse_lines_into_frame(lines: deque[bytearray]) -> AnyClientFrame | AnyServerFrame:
-    command = bytes(lines.popleft())
-    headers = {}
-
-    while line := lines.popleft():
-        header = parse_header(line)
-        if header and header[0] not in headers:
-            headers[header[0]] = header[1]
-    body = bytes(lines.popleft()) if lines else b""
-    return make_frame_from_parts(command=command, headers=headers, body=body)
-
-
-@dataclass(kw_only=True, slots=True)
+@dataclass(kw_only=True, slots=True, init=False)
 class FrameParser:
-    _lines: deque[bytearray] = field(default_factory=deque, init=False)
-    _current_line: bytearray = field(default_factory=bytearray, init=False)
-    _previous_byte: bytes = field(default=b"", init=False)
-    _headers_processed: bool = field(default=False, init=False)
+    _current_buf: bytearray
+    _previous_byte: bytes | None
+    _headers_processed: bool
+    _command: bytes | None
+    _headers: dict[str, str]
+    _content_length: int | None
+
+    def __init__(self) -> None:
+        self._previous_byte = None
+        self._reset()
 
     def _reset(self) -> None:
+        self._current_buf = bytearray()
         self._headers_processed = False
-        self._lines.clear()
-        self._current_line = bytearray()
+        self._command = None
+        self._headers = {}
+        self._content_length = None
+
+    def _handle_null_byte(self) -> Iterator[AnyClientFrame | AnyServerFrame]:
+        if not self._command or not self._headers_processed:
+            self._reset()
+            return
+        if self._content_length is not None and self._content_length != len(self._current_buf):
+            self._current_buf += NULL
+            return
+        yield make_frame_from_parts(command=self._command, headers=self._headers, body=bytes(self._current_buf))
+        self._reset()
+
+    def _handle_newline_byte(self) -> Iterator[HeartbeatFrame]:
+        if not self._current_buf and not self._command:
+            yield HeartbeatFrame()
+            return
+        if self._previous_byte == CARRIAGE:
+            self._current_buf.pop()
+        self._headers_processed = not self._current_buf  # extra empty line after headers
+
+        if self._command:
+            self._process_header()
+        else:
+            self._process_command()
+
+    def _process_command(self) -> None:
+        current_buf_bytes = bytes(self._current_buf)
+        if current_buf_bytes not in COMMANDS_TO_FRAMES:
+            self._reset()
+        else:
+            self._command = current_buf_bytes
+            self._current_buf = bytearray()
+
+    def _process_header(self) -> None:
+        header = parse_header(self._current_buf)
+        if not header:
+            self._current_buf = bytearray()
+            return
+        header_key, header_value = header
+        if header_key not in self._headers:
+            self._headers[header_key] = header_value
+            if header_key.lower() == "content-length":
+                with suppress(ValueError):
+                    self._content_length = int(header_value)
+        self._current_buf = bytearray()
+
+    def _handle_body_byte(self, byte: bytes) -> None:
+        if self._content_length is None or self._content_length != len(self._current_buf):
+            self._current_buf += byte
 
     def parse_frames_from_chunk(self, chunk: bytes) -> Iterator[AnyClientFrame | AnyServerFrame]:
         for byte in iter_bytes(chunk):
             if byte == NULL:
-                if self._headers_processed:
-                    self._lines.append(self._current_line)
-                    yield parse_lines_into_frame(self._lines)
-                self._reset()
-
-            elif not self._headers_processed and byte == NEWLINE:
-                if self._current_line or self._lines:
-                    if self._previous_byte == CARRIAGE:
-                        self._current_line.pop()
-                    self._headers_processed = not self._current_line  # extra empty line after headers
-
-                    if not self._lines and bytes(self._current_line) not in COMMANDS_TO_FRAMES:
-                        self._reset()
-                    else:
-                        self._lines.append(self._current_line)
-                        self._current_line = bytearray()
-                else:
-                    yield HeartbeatFrame()
-
+                yield from self._handle_null_byte()
+            elif self._headers_processed:
+                self._handle_body_byte(byte)
+            elif byte == NEWLINE:
+                yield from self._handle_newline_byte()
             else:
-                self._current_line += byte
+                self._current_buf += byte
 
             self._previous_byte = byte

packages/stompman/test_stompman/test_connection_lifespan.py

@@ -148,7 +148,7 @@ async def mock_sleep(delay: float) -> None:
     async with EnrichedClient(connection_class=connection_class):
         await real_sleep(0)
 
-    assert sleep_calls == [0, 0, 1, 3, 1, 3, 1, 3]
+    assert sleep_calls == [0, 1, 1, 1]
     assert write_heartbeat_mock.mock_calls == [mock.call(), mock.call(), mock.call(), mock.call()]