[ServiceBus] receive related methods/constructor raise error if given max_wait_time <= 0 (Azure#14925)

* raise error if max_wait_time <= 0

* add test for subscription receiver

* Update sdk/servicebus/azure-servicebus/CHANGELOG.md

Author: yunhaoling

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -37,6 +37,8 @@ now raise more concrete exception other than `MessageSettleFailed` and `ServiceB
 * Exceptions `MessageSendFailed`, `MessageSettleFailed` and `MessageLockExpired`
  now inherit from `azure.servicebus.exceptions.MessageError`.
 * `get_state` in `ServiceBusSession` now returns `bytes` instead of a `string`.
+* `ServiceBusReceiver.receive_messages/get_streaming_message_iter` and
+ `ServiceBusClient.get_<queue/subscription>_receiver` now raises ValueError if the given `max_wait_time` is less than 0.
 * Message settlement methods are moved from `ServiceBusMessage` to `ServiceBusReceiver`:
   - Use `ServiceBusReceiver.complete_message` instead of `ServiceBusReceivedMessage.complete` to complete a message.
   - Use `ServiceBusReceiver.abandon_message` instead of `ServiceBusReceivedMessage.abandon` to abandon a message.

sdk/servicebus/azure-servicebus/azure/servicebus/_common/receiver_mixins.py

@@ -69,7 +69,10 @@ def _populate_attributes(self, **kwargs):
         # In large max_message_count case, like 5000, the pull receive would always return hundreds of messages limited
         # by the perf and time.
         self._further_pull_receive_timeout_ms = 200
-        self._max_wait_time = kwargs.get("max_wait_time", None)
+        max_wait_time = kwargs.get("max_wait_time", None)
+        if max_wait_time is not None and max_wait_time <= 0:
+            raise ValueError("The max_wait_time must be greater than 0.")
+        self._max_wait_time = max_wait_time
 
     def _build_message(self, received, message_type=ServiceBusReceivedMessage):
         message = message_type(message=received, receive_mode=self._receive_mode, receiver=self)

sdk/servicebus/azure-servicebus/azure/servicebus/_servicebus_client.py

@@ -221,8 +221,8 @@ def get_queue_receiver(self, queue_name, **kwargs):
          will be immediately removed from the queue, and cannot be subsequently rejected or re-received if
          the client fails to process the message. The default receive_mode is PeekLock.
         :paramtype receive_mode: ~azure.servicebus.ReceiveMode
-        :keyword float max_wait_time: The timeout in seconds between received messages after which the receiver will
-         automatically stop receiving. The default value is 0, meaning no timeout.
+        :keyword Optional[float] max_wait_time: The timeout in seconds between received messages after which the
+         receiver will automatically stop receiving. The default value is None, meaning no timeout.
         :keyword int prefetch_count: The maximum number of messages to cache with each request to the service.
          This setting is only for advanced performance tuning. Increasing this value will improve message throughput
          performance but increase the chance that messages will expire while they are cached if they're not
@@ -327,8 +327,8 @@ def get_subscription_receiver(self, topic_name, subscription_name, **kwargs):
          will be immediately removed from the subscription, and cannot be subsequently rejected or re-received if
          the client fails to process the message. The default receive_mode is PeekLock.
         :paramtype receive_mode: ~azure.servicebus.ReceiveMode
-        :keyword float max_wait_time: The timeout in seconds between received messages after which the receiver will
-         automatically stop receiving. The default value is 0, meaning no timeout.
+        :keyword Optional[float] max_wait_time: The timeout in seconds between received messages after which the
+         receiver will automatically stop receiving. The default value is None, meaning no timeout.
         :keyword int prefetch_count: The maximum number of messages to cache with each request to the service.
          This setting is only for advanced performance tuning. Increasing this value will improve message throughput
          performance but increase the chance that messages will expire while they are cached if they're not

sdk/servicebus/azure-servicebus/azure/servicebus/_servicebus_receiver.py

@@ -77,8 +77,8 @@ class ServiceBusReceiver(BaseHandler, ReceiverMixin):  # pylint: disable=too-man
      the client connects to.
     :keyword str subscription_name: The path of specific Service Bus Subscription under the
      specified Topic the client connects to.
-    :keyword float max_wait_time: The timeout in seconds between received messages after which the receiver will
-     automatically shutdown. The default value is 0, meaning no timeout.
+    :keyword Optional[float] max_wait_time: The timeout in seconds between received messages after which the
+     receiver will automatically stop receiving. The default value is None, meaning no timeout.
     :keyword receive_mode: The mode with which messages will be retrieved from the entity. The two options
      are PeekLock and ReceiveAndDelete. Messages received with PeekLock must be settled within a given
      lock period before they will be removed from the queue. Messages received with ReceiveAndDelete
@@ -381,15 +381,15 @@ def close(self):
         self._message_iter = None  # pylint: disable=attribute-defined-outside-init
 
     def get_streaming_message_iter(self, max_wait_time=None):
-        # type: (float) -> Iterator[ServiceBusReceivedMessage]
+        # type: (Optional[float]) -> Iterator[ServiceBusReceivedMessage]
         """Receive messages from an iterator indefinitely, or if a max_wait_time is specified, until
         such a timeout occurs.
 
         :param max_wait_time: Maximum time to wait in seconds for the next message to arrive.
          If no messages arrive, and no timeout is specified, this call will not return
          until the connection is closed. If specified, and no messages arrive for the
          timeout period, the iterator will stop.
-        :type max_wait_time: float
+        :type max_wait_time: Optional[float]
         :rtype: Iterator[ServiceBusReceivedMessage]
 
         .. admonition:: Example:
@@ -401,6 +401,8 @@ def get_streaming_message_iter(self, max_wait_time=None):
                 :dedent: 4
                 :caption: Receive indefinitely from an iterator in streaming fashion.
         """
+        if max_wait_time is not None and max_wait_time <= 0:
+            raise ValueError("The max_wait_time must be greater than 0.")
         return self._iter_contextual_wrapper(max_wait_time)
 
     @classmethod
@@ -426,8 +428,8 @@ def from_connection_string(
          if the client fails to process the message.
          The default mode is PeekLock.
         :paramtype receive_mode: ~azure.servicebus.ReceiveMode
-        :keyword float max_wait_time: The timeout in seconds between received messages after which the receiver will
-         automatically shutdown. The default value is 0, meaning no timeout.
+        :keyword Optional[float] max_wait_time: The timeout in seconds between received messages after which the
+         receiver will automatically stop receiving. The default value is None, meaning no timeout.
         :keyword bool logging_enable: Whether to output network trace logs to the logger. Default is `False`.
         :keyword transport_type: The type of transport protocol that will be used for communicating with
          the Service Bus service. Default is `TransportType.Amqp`.
@@ -484,9 +486,9 @@ def receive_messages(self, max_message_count=None, max_wait_time=None):
         return as soon as at least one message is received and there is a gap in incoming messages regardless
         of the specified batch size.
 
-        :param int max_message_count: Maximum number of messages in the batch. Actual number
+        :param Optional[int] max_message_count: Maximum number of messages in the batch. Actual number
          returned will depend on prefetch_count and incoming stream rate.
-        :param float max_wait_time: Maximum time to wait in seconds for the first message to arrive.
+        :param Optional[float] max_wait_time: Maximum time to wait in seconds for the first message to arrive.
          If no messages arrive, and no timeout is specified, this call will not return
          until the connection is closed. If specified, an no messages arrive within the
          timeout period, an empty list will be returned.
@@ -503,6 +505,8 @@ def receive_messages(self, max_message_count=None, max_wait_time=None):
                 :caption: Receive messages from ServiceBus.
 
         """
+        if max_wait_time is not None and max_wait_time <= 0:
+            raise ValueError("The max_wait_time must be greater than 0.")
         self._check_live()
         return self._do_retryable_operation(
             self._receive,

sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_client_async.py

@@ -214,8 +214,8 @@ def get_queue_receiver(self, queue_name: str, **kwargs: Any) -> ServiceBusReceiv
          will be immediately removed from the queue, and cannot be subsequently rejected or re-received if
          the client fails to process the message. The default mode is PeekLock.
         :paramtype receive_mode: ~azure.servicebus.ReceiveMode
-        :keyword float max_wait_time: The timeout in seconds between received messages after which the receiver will
-         automatically stop receiving. The default value is 0, meaning no timeout.
+        :keyword Optional[float] max_wait_time: The timeout in seconds between received messages after which the
+         receiver will automatically stop receiving. The default value is None, meaning no timeout.
         :keyword int prefetch_count: The maximum number of messages to cache with each request to the service.
          This setting is only for advanced performance tuning. Increasing this value will improve message throughput
          performance but increase the chance that messages will expire while they are cached if they're not
@@ -317,8 +317,8 @@ def get_subscription_receiver(self, topic_name: str, subscription_name: str, **k
          will be immediately removed from the subscription, and cannot be subsequently rejected or re-received if
          the client fails to process the message. The default mode is PeekLock.
         :paramtype receive_mode: ~azure.servicebus.ReceiveMode
-        :keyword float max_wait_time: The timeout in seconds between received messages after which the receiver will
-         automatically stop receiving. The default value is 0, meaning no timeout.
+        :keyword Optional[float] max_wait_time: The timeout in seconds between received messages after which the
+         receiver will automatically stop receiving. The default value is None, meaning no timeout.
         :keyword int prefetch_count: The maximum number of messages to cache with each request to the service.
          This setting is only for advanced performance tuning. Increasing this value will improve message throughput
          performance but increase the chance that messages will expire while they are cached if they're not

sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_receiver_async.py

@@ -81,8 +81,8 @@ class ServiceBusReceiver(collections.abc.AsyncIterator, BaseHandler, ReceiverMix
      if the client fails to process the message.
      The default mode is PeekLock.
     :paramtype receive_mode: ~azure.servicebus.ReceiveMode
-    :keyword float max_wait_time: The timeout in seconds between received messages after which the receiver will
-     automatically shutdown. The default value is 0, meaning no timeout.
+    :keyword Optional[float] max_wait_time: The timeout in seconds between received messages after which the receiver
+     will automatically stop receiving. The default value is None, meaning no timeout.
     :keyword bool logging_enable: Whether to output network trace logs to the logger. Default is `False`.
     :keyword transport_type: The type of transport protocol that will be used for communicating with
      the Service Bus service. Default is `TransportType.Amqp`.
@@ -402,6 +402,8 @@ def get_streaming_message_iter(
                 :dedent: 4
                 :caption: Receive indefinitely from an iterator in streaming fashion.
         """
+        if max_wait_time is not None and max_wait_time <= 0:
+            raise ValueError("The max_wait_time must be greater than 0.")
         return self._IterContextualWrapper(self, max_wait_time)
 
     @classmethod
@@ -425,8 +427,8 @@ def from_connection_string(
          if the client fails to process the message.
          The default mode is PeekLock.
         :paramtype receive_mode: ~azure.servicebus.ReceiveMode
-        :keyword float max_wait_time: The timeout in seconds between received messages after which the receiver will
-         automatically shutdown. The default value is 0, meaning no timeout.
+        :keyword Optional[float] max_wait_time: The timeout in seconds between received messages after which the
+         receiver will automatically stop receiving. The default value is None, meaning no timeout.
         :keyword bool logging_enable: Whether to output network trace logs to the logger. Default is `False`.
         :keyword transport_type: The type of transport protocol that will be used for communicating with
          the Service Bus service. Default is `TransportType.Amqp`.
@@ -504,6 +506,8 @@ async def receive_messages(
                 :caption: Receive messages from ServiceBus.
 
         """
+        if max_wait_time is not None and max_wait_time <= 0:
+            raise ValueError("The max_wait_time must be greater than 0.")
         self._check_live()
         return await self._do_retryable_operation(
             self._receive,

sdk/servicebus/azure-servicebus/tests/async_tests/test_queues_async.py

@@ -69,7 +69,17 @@ async def test_async_queue_by_queue_client_conn_str_receive_handler_peeklock(sel
             with pytest.raises(ServiceBusConnectionError):
                 await (sb_client.get_queue_receiver(servicebus_queue.name, session_id="test", max_wait_time=5))._open_with_retry()
 
+            with pytest.raises(ValueError):
+                sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=0)
+
             async with sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5) as receiver:
+
+                with pytest.raises(ValueError):
+                    await receiver.receive_messages(max_wait_time=0)
+
+                with pytest.raises(ValueError):
+                    await receiver.get_streaming_message_iter(max_wait_time=0)
+
                 count = 0
                 async for message in receiver:
                     print_message(_logger, message)

sdk/servicebus/azure-servicebus/tests/async_tests/test_subscriptions_async.py

@@ -47,11 +47,25 @@ async def test_subscription_by_subscription_client_conn_str_receive_basic(self,
                 message = ServiceBusMessage(b"Sample topic message")
                 await sender.send_messages(message)
 
+            with pytest.raises(ValueError):
+                sb_client.get_subscription_receiver(
+                    topic_name=servicebus_topic.name,
+                    subscription_name=servicebus_subscription.name,
+                    max_wait_time=0
+                )
+
             async with sb_client.get_subscription_receiver(
                     topic_name=servicebus_topic.name,
                     subscription_name=servicebus_subscription.name,
                     max_wait_time=5
             ) as receiver:
+
+                with pytest.raises(ValueError):
+                    await receiver.receive_messages(max_wait_time=-1)
+
+                with pytest.raises(ValueError):
+                    await receiver.get_streaming_message_iter(max_wait_time=0)
+
                 count = 0
                 async for message in receiver:
                     count += 1

sdk/servicebus/azure-servicebus/tests/test_queues.py

@@ -131,7 +131,17 @@ def test_queue_by_queue_client_conn_str_receive_handler_peeklock(self, servicebu
                     message.reply_to = 'reply_to'
                     sender.send_messages(message)
 
+            with pytest.raises(ValueError):
+                sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=0)
+
             receiver = sb_client.get_queue_receiver(servicebus_queue.name, max_wait_time=5)
+
+            with pytest.raises(ValueError):
+                receiver.receive_messages(max_wait_time=0)
+
+            with pytest.raises(ValueError):
+                receiver.get_streaming_message_iter(max_wait_time=0)
+
             count = 0
             for message in receiver:
                 print_message(_logger, message)

sdk/servicebus/azure-servicebus/tests/test_subscriptions.py

@@ -46,11 +46,25 @@ def test_subscription_by_subscription_client_conn_str_receive_basic(self, servic
                 message = ServiceBusMessage(b"Sample topic message")
                 sender.send_messages(message)
 
+            with pytest.raises(ValueError):
+                sb_client.get_subscription_receiver(
+                    topic_name=servicebus_topic.name,
+                    subscription_name=servicebus_subscription.name,
+                    max_wait_time=0
+                )
+
             with sb_client.get_subscription_receiver(
                     topic_name=servicebus_topic.name,
                     subscription_name=servicebus_subscription.name,
                     max_wait_time=5
             ) as receiver:
+
+                with pytest.raises(ValueError):
+                    receiver.receive_messages(max_wait_time=-1)
+
+                with pytest.raises(ValueError):
+                    receiver.get_streaming_message_iter(max_wait_time=0)
+
                 count = 0
                 for message in receiver:
                     count += 1