Address CodeRabbit review: codec and connection hardening

- Pass content_type to _load_bs4 so XML content uses the xml parser
  instead of silently applying html.parser to text/xml
- Wrap encode() in try/except to maintain EncodeError contract,
  matching the existing pattern in decode()
- Extract _serialize and _compress helpers to reduce encode complexity
- Guard against missing umsgpack with explicit EncodeError
- Catch ConnectionWrongStateError alongside ConnectionClosed in on_open
- Treat AMQP reply codes 403, 405, 406 as terminal channel errors to
  prevent infinite reopen loops on configuration/permission failures

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: gmr

rejected/codecs.py

@@ -119,7 +119,7 @@ async def decode(
             if content_type == 'text/csv':
                 return _load_csv(body)
             if bs4 and content_type in BS4_MIME_TYPES:
-                return _load_bs4(body)
+                return _load_bs4(body, content_type)
             if content_type in YAML_MIME_TYPES:
                 return yaml.safe_load(body)
         except DecodeError:
@@ -142,34 +142,46 @@ async def encode(
         str/bytes), then content_encoding compression.
 
         """
-        if not isinstance(body, (str, bytes)):
-            if content_type == AVRO_DATUM_MIME_TYPE:
-                body = await self._encode_avro(body, message_type)
-            elif content_type == 'application/json':
-                body = json.dumps(body, ensure_ascii=True).encode('utf-8')
-            elif umsgpack and content_type == 'application/msgpack':
-                body = umsgpack.packb(body)
-            elif content_type == 'application/x-plist':
-                body = plistlib.dumps(body)
-            elif content_type == 'text/csv':
-                body = _dump_csv(body)
-            elif (
-                bs4
-                and isinstance(body, bs4.BeautifulSoup)
-                and content_type in BS4_MIME_TYPES
-            ):
-                body = str(body)
-            elif content_type in YAML_MIME_TYPES:
-                body = yaml.dump(body)
-
-        if content_encoding:
-            if isinstance(body, str):
-                body = body.encode('utf-8')
-            if content_encoding == 'gzip':
-                body = gzip.compress(body)
-            elif content_encoding == 'bzip2':
-                body = bz2.compress(body)
+        try:
+            if not isinstance(body, (str, bytes)):
+                body = await self._serialize(body, content_type, message_type)
+            if content_encoding:
+                body = _compress(body, content_encoding)
+        except EncodeError:
+            raise
+        except Exception as error:
+            raise EncodeError(str(error)) from error
+        return body
 
+    async def _serialize(
+        self,
+        body: typing.Any,
+        content_type: str | None,
+        message_type: str | None,
+    ) -> typing.Any:
+        """Serialize a non-string/bytes body by content type."""
+        if content_type == AVRO_DATUM_MIME_TYPE:
+            return await self._encode_avro(body, message_type)
+        if content_type == 'application/json':
+            return json.dumps(body, ensure_ascii=True).encode('utf-8')
+        if content_type == 'application/msgpack':
+            if umsgpack is None:
+                raise EncodeError(
+                    'umsgpack is required for MessagePack encoding'
+                )
+            return umsgpack.packb(body)
+        if content_type == 'application/x-plist':
+            return plistlib.dumps(body)
+        if content_type == 'text/csv':
+            return _dump_csv(body)
+        if (
+            bs4
+            and isinstance(body, bs4.BeautifulSoup)
+            and content_type in BS4_MIME_TYPES
+        ):
+            return str(body)
+        if content_type in YAML_MIME_TYPES:
+            return yaml.dump(body)
         return body
 
     async def _encode_avro(
@@ -282,6 +294,17 @@ async def _load_http_schema(
 # --- Internal helpers (stateless, sync) ---
 
 
+def _compress(body: typing.Any, content_encoding: str) -> bytes:
+    """Apply content-encoding compression to a body."""
+    if isinstance(body, str):
+        body = body.encode('utf-8')
+    if content_encoding == 'gzip':
+        return gzip.compress(body)
+    if content_encoding == 'bzip2':
+        return bz2.compress(body)
+    return body
+
+
 def _load_json(value: bytes | str) -> typing.Any:
     """Deserialize a JSON value."""
     if isinstance(value, bytes):
@@ -310,13 +333,14 @@ def _load_csv(value: bytes | str) -> csv.DictReader[str]:
     return csv.DictReader(csv_buffer, dialect=dialect)
 
 
-def _load_bs4(value: bytes | str) -> typing.Any:
+def _load_bs4(value: bytes | str, content_type: str) -> typing.Any:
     """Parse HTML or XML into a BeautifulSoup object."""
     if not bs4:
         raise DecodeError('BeautifulSoup4 is not enabled')
     if isinstance(value, bytes):
         value = value.decode('utf-8')
-    return bs4.BeautifulSoup(value, 'html.parser')
+    parser = 'xml' if content_type == 'text/xml' else 'html.parser'
+    return bs4.BeautifulSoup(value, parser)
 
 
 def _dump_csv(value: list[list[typing.Any]]) -> str:

rejected/connection.py

@@ -104,7 +104,10 @@ def on_open(
             return
         try:
             self.connection.channel(on_open_callback=self.on_channel_open)
-        except pika.exceptions.ConnectionClosed:
+        except (
+            pika.exceptions.ConnectionWrongStateError,
+            pika.exceptions.ConnectionClosed,
+        ):
             LOGGER.warning('Channel open on closed connection')
             self.set_state(self.STATE_CLOSED)
             self.callbacks.on_closed(self.name)
@@ -202,7 +205,8 @@ def on_channel_closed(
             reply_code = 0
             reply_text = str(closing_reason) or 'unknown'
 
-        if reply_code <= 0 or reply_code == 404:
+        terminal_codes = {0, 403, 404, 405, 406}
+        if reply_code <= 0 or reply_code in terminal_codes:
             LOGGER.error(
                 'Channel Error (%r): %s', reply_code, reply_text or 'unknown'
             )