refactor consumer option (#81)

This PR refactors consumer options by introducing an abstract ConsumerOptions class that provides a common interface for queue-specific consumer configurations. The refactoring generalizes the consumer interface to support different queue types while maintaining backward compatibility for stream queues.

Adds an abstract ConsumerOptions base class with validation and filter set methods
Refactors StreamConsumerOptions to inherit from the new abstract class and implement validation logic
Updates consumer creation interface to use the generalized consumer_options parameter
---------

Signed-off-by: Gabriele Santomaggio <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: Gsantomaggio

examples/streams/example_with_streams.py

@@ -104,7 +104,7 @@ def main() -> None:
         message_handler=MyMessageHandler(),
         # can be first, last, next or an offset long
         # you can also specify stream filters with methods: apply_filters and filter_match_unfiltered
-        stream_consumer_options=StreamConsumerOptions(
+        consumer_options=StreamConsumerOptions(
             offset_specification=OffsetSpecification.first
         ),
     )

examples/streams_with_filters/example_streams_with_filters.py

@@ -91,7 +91,7 @@ def main() -> None:
         message_handler=MyMessageHandler(),
         # the consumer will only receive messages with filter value banana and subject yellow
         # and application property from = italy
-        stream_consumer_options=StreamConsumerOptions(
+        consumer_options=StreamConsumerOptions(
             offset_specification=OffsetSpecification.first,
             filter_options=StreamFilterOptions(
                 values=["banana"],

examples/streams_with_sql_filters/example_streams_with_sql_filters.py

@@ -88,7 +88,7 @@ def main() -> None:
     consumer = consumer_connection.consumer(
         addr_queue,
         message_handler=MyMessageHandler(),
-        stream_consumer_options=StreamConsumerOptions(
+        consumer_options=StreamConsumerOptions(
             offset_specification=OffsetSpecification.first,
             filter_options=StreamFilterOptions(sql=sql),
         ),

rabbitmq_amqp_python_client/__init__.py

@@ -6,6 +6,7 @@
 from .connection import Connection
 from .consumer import Consumer
 from .entities import (
+    ConsumerOptions,
     ExchangeCustomSpecification,
     ExchangeSpecification,
     ExchangeToExchangeBindingSpecification,
@@ -89,6 +90,7 @@
     "ConnectionClosed",
     "StreamConsumerOptions",
     "StreamFilterOptions",
+    "ConsumerOptions",
     "MessageProperties",
     "OffsetSpecification",
     "OutcomeState",

rabbitmq_amqp_python_client/connection.py

@@ -16,9 +16,9 @@
 from .address_helper import validate_address
 from .consumer import Consumer
 from .entities import (
+    ConsumerOptions,
     OAuth2Options,
     RecoveryConfiguration,
-    StreamConsumerOptions,
 )
 from .exceptions import (
     ArgumentOutOfRangeException,
@@ -211,15 +211,15 @@ def _validate_server_properties(self) -> None:
 
         logger.debug(f"Connected to RabbitMQ server version {server_version}")
 
-    def _is_server_version_gte_4_2_0(self) -> bool:
+    def _is_server_version_gte(self, target_version: str) -> bool:
         """
-        Check if the server version is greater than or equal to 4.2.0.
+        Check if the server version is greater than or equal to version.
 
         This is an internal method that can be used to conditionally enable
-        features that require RabbitMQ 4.2.0 or higher.
+        features that require RabbitMQ version or higher.
 
         Returns:
-            bool: True if server version >= 4.2.0, False otherwise
+            bool: True if server version >= version, False otherwise
 
         Raises:
             ValidationCodeException: If connection is not established or
@@ -237,7 +237,12 @@ def _is_server_version_gte_4_2_0(self) -> bool:
             raise ValidationCodeException("Server version not provided")
 
         try:
-            return version.parse(str(server_version)) >= version.parse("4.2.0")
+            srv = version.parse(str(server_version))
+            trg = version.parse(target_version)
+            # compare the version even if it contains pre-release or build metadata
+            return (
+                version.parse("{}.{}.{}".format(srv.major, srv.minor, srv.micro)) >= trg
+            )
         except Exception as e:
             raise ValidationCodeException(
                 f"Failed to parse server version '{server_version}': {e}"
@@ -376,7 +381,7 @@ def consumer(
         self,
         destination: str,
         message_handler: Optional[MessagingHandler] = None,
-        stream_consumer_options: Optional[StreamConsumerOptions] = None,
+        consumer_options: Optional[ConsumerOptions] = None,
         credit: Optional[int] = None,
     ) -> Consumer:
         """
@@ -385,7 +390,7 @@ def consumer(
         Args:
             destination: The address to consume from
             message_handler: Optional handler for processing messages
-            stream_consumer_options: Optional configuration for stream consumption
+            consumer_options: Optional configuration for queue consumption. Each queue has its own consumer options.
             credit: Optional credit value for flow control
 
         Returns:
@@ -398,8 +403,16 @@ def consumer(
             raise ArgumentOutOfRangeException(
                 "destination address must start with /queues or /exchanges"
             )
+        if consumer_options is not None:
+            consumer_options.validate(
+                {
+                    "4.0.0": self._is_server_version_gte("4.0.0"),
+                    "4.1.0": self._is_server_version_gte("4.1.0"),
+                    "4.2.0": self._is_server_version_gte("4.2.0"),
+                }
+            )
         consumer = Consumer(
-            self._conn, destination, message_handler, stream_consumer_options, credit
+            self._conn, destination, message_handler, consumer_options, credit
         )
         self._consumers.append(consumer)
         return consumer

rabbitmq_amqp_python_client/consumer.py

@@ -2,7 +2,7 @@
 from typing import Literal, Optional, Union, cast
 
 from .amqp_consumer_handler import AMQPMessagingHandler
-from .entities import StreamConsumerOptions
+from .entities import ConsumerOptions
 from .options import (
     ReceiverOptionUnsettled,
     ReceiverOptionUnsettledWithFilters,
@@ -38,7 +38,7 @@ def __init__(
         conn: BlockingConnection,
         addr: str,
         handler: Optional[AMQPMessagingHandler] = None,
-        stream_options: Optional[StreamConsumerOptions] = None,
+        stream_options: Optional[ConsumerOptions] = None,
         credit: Optional[int] = None,
     ):
         """

rabbitmq_amqp_python_client/entities.py

@@ -153,6 +153,14 @@ class ExchangeToExchangeBindingSpecification:
     binding_key: Optional[str] = None
 
 
+class ConsumerOptions:
+    def validate(self, versions: Dict[str, bool]) -> None:
+        raise NotImplementedError("Subclasses should implement this method")
+
+    def filter_set(self) -> Dict[symbol, Described]:
+        raise NotImplementedError("Subclasses should implement this method")
+
+
 @dataclass
 class MessageProperties:
     """
@@ -215,7 +223,7 @@ def __init__(
         self.sql = sql
 
 
-class StreamConsumerOptions:
+class StreamConsumerOptions(ConsumerOptions):
     """
     Configuration options for stream queues.
 
@@ -237,6 +245,7 @@ def __init__(
     ):
 
         self._filter_set: Dict[symbol, Described] = {}
+        self._filter_option = filter_options
 
         if offset_specification is None and filter_options is None:
             raise ValidationCodeException(
@@ -329,7 +338,6 @@ def _filter_message_properties(
     def _filter_application_properties(
         self, application_properties: Optional[dict[str, Any]]
     ) -> None:
-        app_prop = {}
         if application_properties is not None:
             app_prop = application_properties.copy()
 
@@ -356,6 +364,41 @@ def filter_set(self) -> Dict[symbol, Described]:
         """
         return self._filter_set
 
+    def validate(self, versions: Dict[str, bool]) -> None:
+        """
+        Validates stream filter options against supported RabbitMQ server versions.
+
+        Args:
+            versions: Dictionary mapping version strings to boolean indicating support.
+
+        Raises:
+            ValidationCodeException: If a filter option requires a higher RabbitMQ version.
+        """
+        if self._filter_option is None:
+            return
+        if self._filter_option.values and not versions.get("4.1.0", False):
+            raise ValidationCodeException(
+                "Stream filter by values requires RabbitMQ 4.1.0 or higher"
+            )
+        if self._filter_option.match_unfiltered and not versions.get("4.1.0", False):
+            raise ValidationCodeException(
+                "Stream filter by match_unfiltered requires RabbitMQ 4.1.0 or higher"
+            )
+        if self._filter_option.sql and not versions.get("4.2.0", False):
+            raise ValidationCodeException(
+                "Stream filter by SQL requires RabbitMQ 4.2.0 or higher"
+            )
+        if self._filter_option.message_properties and not versions.get("4.1.0", False):
+            raise ValidationCodeException(
+                "Stream filter by message_properties requires RabbitMQ 4.1.0 or higher"
+            )
+        if self._filter_option.application_properties and not versions.get(
+            "4.1.0", False
+        ):
+            raise ValidationCodeException(
+                "Stream filter by application_properties requires RabbitMQ 4.1.0 or higher"
+            )
+
 
 @dataclass
 class RecoveryConfiguration:

rabbitmq_amqp_python_client/options.py

@@ -1,4 +1,4 @@
-from .entities import StreamConsumerOptions
+from .entities import ConsumerOptions
 from .qpid.proton._data import (  # noqa: E402
     PropertyDict,
     symbol,
@@ -68,8 +68,8 @@ def test(self, link: Link) -> bool:
 
 
 class ReceiverOptionUnsettledWithFilters(Filter):  # type: ignore
-    def __init__(self, addr: str, filter_options: StreamConsumerOptions):
-        super().__init__(filter_options.filter_set())
+    def __init__(self, addr: str, consumer_options: ConsumerOptions):
+        super().__init__(consumer_options.filter_set())
         self._addr = addr
 
     def apply(self, link: Link) -> None:

tests/test_server_validation.py

@@ -306,7 +306,7 @@ def test_is_server_version_gte_4_2_0_exact_version(self):
         mock_blocking_conn.conn = mock_proton_conn
         self.connection._conn = mock_blocking_conn
 
-        result = self.connection._is_server_version_gte_4_2_0()
+        result = self.connection._is_server_version_gte("4.2.0")
         assert result is True
 
     def test_is_server_version_gte_4_2_0_higher_versions(self):
@@ -322,7 +322,7 @@ def test_is_server_version_gte_4_2_0_higher_versions(self):
             mock_blocking_conn.conn = mock_proton_conn
             self.connection._conn = mock_blocking_conn
 
-            result = self.connection._is_server_version_gte_4_2_0()
+            result = self.connection._is_server_version_gte("4.2.0")
             assert result is True, f"Version {version_str} should return True"
 
     def test_is_server_version_gte_4_2_0_lower_versions(self):
@@ -338,15 +338,15 @@ def test_is_server_version_gte_4_2_0_lower_versions(self):
             mock_blocking_conn.conn = mock_proton_conn
             self.connection._conn = mock_blocking_conn
 
-            result = self.connection._is_server_version_gte_4_2_0()
+            result = self.connection._is_server_version_gte("4.2.0")
             assert result is False, f"Version {version_str} should return False"
 
     def test_is_server_version_gte_4_2_0_no_connection(self):
         """Test when connection is None."""
         self.connection._conn = None
 
         with pytest.raises(ValidationCodeException) as exc_info:
-            self.connection._is_server_version_gte_4_2_0()
+            self.connection._is_server_version_gte("4.2.0")
 
         assert "Connection not established" in str(exc_info.value)
 
@@ -357,7 +357,7 @@ def test_is_server_version_gte_4_2_0_no_proton_connection(self):
         self.connection._conn = mock_blocking_conn
 
         with pytest.raises(ValidationCodeException) as exc_info:
-            self.connection._is_server_version_gte_4_2_0()
+            self.connection._is_server_version_gte("4.2.0")
 
         assert "Connection not established" in str(exc_info.value)
 
@@ -370,7 +370,7 @@ def test_is_server_version_gte_4_2_0_no_remote_properties(self):
         self.connection._conn = mock_blocking_conn
 
         with pytest.raises(ValidationCodeException) as exc_info:
-            self.connection._is_server_version_gte_4_2_0()
+            self.connection._is_server_version_gte("4.2.0")
 
         assert "No remote properties received from server" in str(exc_info.value)
 
@@ -388,7 +388,7 @@ def test_is_server_version_gte_4_2_0_missing_version(self):
         self.connection._conn = mock_blocking_conn
 
         with pytest.raises(ValidationCodeException) as exc_info:
-            self.connection._is_server_version_gte_4_2_0()
+            self.connection._is_server_version_gte("4.2.0")
 
         assert "Server version not provided" in str(exc_info.value)
 
@@ -406,7 +406,7 @@ def test_is_server_version_gte_4_2_0_invalid_version_format(self):
             self.connection._conn = mock_blocking_conn
 
             with pytest.raises(ValidationCodeException) as exc_info:
-                self.connection._is_server_version_gte_4_2_0()
+                self.connection._is_server_version_gte("4.2.0")
 
             error_msg = str(exc_info.value)
             assert "Failed to parse server version" in error_msg
@@ -419,7 +419,10 @@ def test_is_server_version_gte_4_2_0_edge_cases(self):
             ("4.2.0", True),  # Exact match
             ("4.2.0.0", True),  # With extra zeroes
             ("v4.2.0", True),  # With v prefix
-            ("4.2.0-rc1", False),  # Pre-release should be less than 4.2.0
+            (
+                "4.2.0-rc1",
+                True,
+            ),  # Pre-release should be less than 4.2.0 but accepted it equal
         ]
 
         for version_str, expected in test_cases:
@@ -433,12 +436,12 @@ def test_is_server_version_gte_4_2_0_edge_cases(self):
 
             if version_str == "4.2.0-rc1":
                 # Pre-release versions should be handled correctly
-                result = self.connection._is_server_version_gte_4_2_0()
+                result = self.connection._is_server_version_gte("4.2.0")
                 assert (
                     result == expected
                 ), f"Version {version_str} should return {expected}"
             else:
-                result = self.connection._is_server_version_gte_4_2_0()
+                result = self.connection._is_server_version_gte("4.2.0")
                 assert (
                     result == expected
                 ), f"Version {version_str} should return {expected}"