[Service Bus] Exception Handling review

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -7,6 +7,10 @@
 * Added method `get_topic_sender` in `ServiceBusClient` to get a `ServiceBusSender` for a topic.
 * Added method `get_subscription_receiver` in `ServiceBusClient` to get a `ServiceBusReceiver` for a subscription under specific topic.
 * `ServiceBusSender.send()` can now send a list of messages in one call, if they fit into a single batch.  If they do not fit a `ValueError` is thrown.
+* `BatchMessage.add()` and `ServiceBusSender.send()` raises `MessageContentTooLarge`, which is a subclass of `ValueError` if the content is over-sized.
+* `ServiceBusReceiver.receive()` raises `ValueError` if the max_batch_size is greater than the prefetch of `ServiceBusClient`.
+* Added exception classes `MessageError`, `MessageContentTooLarge`, `ServiceBusAuthenticationError`.
+* Removed exception class `InvalidHandlerState`.
 
 **BugFixes**
 

sdk/servicebus/azure-servicebus/azure/servicebus/_base_handler.py

@@ -20,7 +20,6 @@
 
 from ._common._configuration import Configuration
 from .exceptions import (
-    InvalidHandlerState,
     ServiceBusError,
     ServiceBusAuthorizationError,
     _create_servicebus_exception
@@ -236,9 +235,6 @@ def _do_retryable_operation(self, operation, timeout=None, **kwargs):
 
     def _mgmt_request_response(self, mgmt_operation, message, callback, keep_alive_associated_link=True, **kwargs):
         self._open()
-        if not self._running:
-            raise InvalidHandlerState("Client connection is closed.")
-
         application_properties = {}
         # Some mgmt calls do not support an associated link name (such as list_sessions).  Most do, so on by default.
         if keep_alive_associated_link:

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

@@ -43,8 +43,8 @@
     MessageAlreadySettled,
     MessageLockExpired,
     SessionLockExpired,
-    MessageSettleFailed
-)
+    MessageSettleFailed,
+    MessageContentTooLarge)
 from .utils import utc_from_timestamp, utc_now
 
 _LOGGER = logging.getLogger(__name__)
@@ -340,8 +340,8 @@ def add(self, message):
         )
 
         if size_after_add > self.max_size_in_bytes:
-            raise ValueError(
-                "EventDataBatch has reached its size limit: {}".format(
+            raise MessageContentTooLarge(
+                "BatchMessage has reached its size limit: {}".format(
                     self.max_size_in_bytes
                 )
             )
@@ -453,7 +453,7 @@ def __init__(self, message, mode=ReceiveSettleMode.PeekLock, **kwargs):
         self._is_deferred_message = kwargs.get("is_deferred_message", False)
         self.auto_renew_error = None
 
-    def _is_live(self, action):
+    def _check_live(self, action):
         # pylint: disable=no-member
         if not self._receiver or not self._receiver._running:  # pylint: disable=protected-access
             raise MessageSettleFailed(action, "Orphan message had no open connection.")
@@ -611,7 +611,7 @@ def complete(self):
         :raises: ~azure.servicebus.exceptions.MessageSettleFailed if message settle operation fails.
         """
         # pylint: disable=protected-access
-        self._is_live(MESSAGE_COMPLETE)
+        self._check_live(MESSAGE_COMPLETE)
         self._settle_message(MESSAGE_COMPLETE)
         self._settled = True
 
@@ -632,7 +632,7 @@ def dead_letter(self, reason=None, description=None):
         :raises: ~azure.servicebus.exceptions.MessageSettleFailed if message settle operation fails.
         """
         # pylint: disable=protected-access
-        self._is_live(MESSAGE_DEAD_LETTER)
+        self._check_live(MESSAGE_DEAD_LETTER)
 
         details = {
             MGMT_REQUEST_DEAD_LETTER_REASON: str(reason) if reason else "",
@@ -654,7 +654,7 @@ def abandon(self):
         :raises: ~azure.servicebus.exceptions.MessageSettleFailed if message settle operation fails.
         """
         # pylint: disable=protected-access
-        self._is_live(MESSAGE_ABANDON)
+        self._check_live(MESSAGE_ABANDON)
         self._settle_message(MESSAGE_ABANDON)
         self._settled = True
 
@@ -671,7 +671,7 @@ def defer(self):
         :raises: ~azure.servicebus.exceptions.SessionLockExpired if session lock has already expired.
         :raises: ~azure.servicebus.exceptions.MessageSettleFailed if message settle operation fails.
         """
-        self._is_live(MESSAGE_DEFER)
+        self._check_live(MESSAGE_DEFER)
         self._settle_message(MESSAGE_DEFER)
         self._settled = True
 
@@ -696,7 +696,7 @@ def renew_lock(self):
                 raise TypeError("Session messages cannot be renewed. Please renew the Session lock instead.")
         except AttributeError:
             pass
-        self._is_live(MESSAGE_RENEW_LOCK)
+        self._check_live(MESSAGE_RENEW_LOCK)
         token = self.lock_token
         if not token:
             raise ValueError("Unable to renew lock - no lock token found.")

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

@@ -45,8 +45,8 @@ def _build_message(self, received, message_type=ReceivedMessage):
         self._last_received_sequenced_number = message.sequence_number
         return message
 
-    def _can_run(self):
-        return True
+    def _check_live(self):
+        """check whether the receiver is alive"""
 
     def _get_source(self):
         return self._entity_uri
@@ -78,7 +78,7 @@ def _on_attach(self, source, target, properties, error):  # pylint: disable=unus
             self._session_id = session_filter.decode(self._config.encoding)
             self._session._session_id = self._session_id
 
-    def _can_run(self):
+    def _check_live(self):
         if self._session and self._session.expired:
             raise SessionLockExpired(inner_exception=self._session.auto_renew_error)
 

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

@@ -130,7 +130,7 @@ def __iter__(self):
         return self
 
     def __next__(self):
-        self._can_run()
+        self._check_live()
         while True:
             try:
                 return self._do_retryable_operation(self._iter_next)
@@ -305,7 +305,10 @@ def receive(self, max_batch_size=None, max_wait_time=None):
                 :caption: Receive messages from ServiceBus.
 
         """
-        self._can_run()
+        self._check_live()
+        if max_batch_size and self._config.prefetch < max_batch_size:
+            raise ValueError("max_batch_size should be less than or equal to prefetch of ServiceBusReceiver, or you "
+                             "could set a larger prefetch value when you're constructing the ServiceBusReceiver.")
         return self._do_retryable_operation(
             self._receive,
             max_batch_size=max_batch_size,
@@ -334,7 +337,7 @@ def receive_deferred_messages(self, sequence_numbers):
                 :caption: Receive deferred messages from ServiceBus.
 
         """
-        self._can_run()
+        self._check_live()
         if not sequence_numbers:
             raise ValueError("At least one sequence number must be specified.")
         self._open()
@@ -381,7 +384,7 @@ def peek(self, message_count=1, sequence_number=None):
                 :caption: Look at pending messages in the queue.
 
         """
-        self._can_run()
+        self._check_live()
         if not sequence_number:
             sequence_number = self._last_received_sequenced_number or 1
         if int(message_count) < 1:

sdk/servicebus/azure-servicebus/azure/servicebus/_servicebus_sender.py

@@ -14,10 +14,9 @@
 from ._common import mgmt_handlers
 from ._common.message import Message, BatchMessage
 from .exceptions import (
-    MessageSendFailed,
     OperationTimeoutError,
-    _ServiceBusErrorPolicy
-)
+    _ServiceBusErrorPolicy,
+    )
 from ._common.utils import create_authentication
 from ._common.constants import (
     REQUEST_RESPONSE_CANCEL_SCHEDULED_MESSAGE_OPERATION,
@@ -183,10 +182,7 @@ def _open(self):
     def _send(self, message, timeout=None, last_exception=None):
         self._open()
         self._set_msg_timeout(timeout, last_exception)
-        try:
-            self._handler.send_message(message.message)
-        except Exception as e:
-            raise MessageSendFailed(e)
+        self._handler.send_message(message.message)
 
     def _schedule(self, message, schedule_time_utc):
         # type: (Union[Message, BatchMessage], datetime.datetime) -> List[int]
@@ -319,8 +315,10 @@ def send(self, message):
             batch = self.create_batch()
             batch._from_list(message)
             message = batch
-        except TypeError: # Message was not a list or generator.
+        except TypeError:  # Message was not a list or generator.
             pass
+        if isinstance(message, BatchMessage) and len(message) == 0:
+            raise ValueError("A BatchMessage or list of Message must have at least one Message")
 
         self._do_retryable_operation(
             self._send,

sdk/servicebus/azure-servicebus/azure/servicebus/_servicebus_session.py

@@ -53,7 +53,7 @@ def __init__(self, session_id, receiver, encoding="UTF-8"):
         self._locked_until_utc = None
         self.auto_renew_error = None
 
-    def _can_run(self):
+    def _check_live(self):
         if self.expired:
             raise SessionLockExpired(inner_exception=self.auto_renew_error)
 
@@ -74,7 +74,7 @@ def get_session_state(self):
                 :dedent: 4
                 :caption: Get the session state
         """
-        self._can_run()
+        self._check_live()
         response = self._receiver._mgmt_request_response_with_retry(  # pylint: disable=protected-access
             REQUEST_RESPONSE_GET_SESSION_STATE_OPERATION,
             {MGMT_REQUEST_SESSION_ID: self.session_id},
@@ -101,7 +101,7 @@ def set_session_state(self, state):
                 :dedent: 4
                 :caption: Set the session state
         """
-        self._can_run()
+        self._check_live()
         state = state.encode(self._encoding) if isinstance(state, six.text_type) else state
         return self._receiver._mgmt_request_response_with_retry(  # pylint: disable=protected-access
             REQUEST_RESPONSE_SET_SESSION_STATE_OPERATION,
@@ -128,7 +128,7 @@ def renew_lock(self):
                 :dedent: 4
                 :caption: Renew the session lock before it expires
         """
-        self._can_run()
+        self._check_live()
         expiry = self._receiver._mgmt_request_response_with_retry(  # pylint: disable=protected-access
             REQUEST_RESPONSE_RENEW_SESSION_LOCK_OPERATION,
             {MGMT_REQUEST_SESSION_ID: self.session_id},

sdk/servicebus/azure-servicebus/azure/servicebus/_servicebus_session_receiver.py

@@ -155,4 +155,4 @@ def from_connection_string(
                 :caption: Create a new instance of the ServiceBusReceiver from connection string.
 
         """
-        return super(ServiceBusSessionReceiver, self).from_connection_string(conn_str, **kwargs)
+        return super(ServiceBusSessionReceiver, cls).from_connection_string(conn_str, **kwargs)

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

@@ -68,7 +68,7 @@ async def complete(self):
         :raises: ~azure.servicebus.exceptions.MessageSettleFailed if message settle operation fails.
         """
         # pylint: disable=protected-access
-        self._is_live(MESSAGE_COMPLETE)
+        self._check_live(MESSAGE_COMPLETE)
         await self._settle_message(MESSAGE_COMPLETE)
         self._settled = True
 
@@ -88,7 +88,7 @@ async def dead_letter(self, reason=None, description=None):
         :raises: ~azure.servicebus.exceptions.MessageSettleFailed if message settle operation fails.
         """
         # pylint: disable=protected-access
-        self._is_live(MESSAGE_DEAD_LETTER)
+        self._check_live(MESSAGE_DEAD_LETTER)
         await self._settle_message(MESSAGE_DEAD_LETTER)
         self._settled = True
 
@@ -102,7 +102,7 @@ async def abandon(self):
         :raises: ~azure.servicebus.exceptions.MessageSettleFailed if message settle operation fails.
         """
         # pylint: disable=protected-access
-        self._is_live(MESSAGE_ABANDON)
+        self._check_live(MESSAGE_ABANDON)
         await self._settle_message(MESSAGE_ABANDON)
         self._settled = True
 
@@ -116,7 +116,7 @@ async def defer(self):
         :raises: ~azure.servicebus.exceptions.MessageSettleFailed if message settle operation fails.
         """
         # pylint: disable=protected-access
-        self._is_live(MESSAGE_DEFER)
+        self._check_live(MESSAGE_DEFER)
         await self._settle_message(MESSAGE_DEFER)
         self._settled = True
 
@@ -142,7 +142,7 @@ async def renew_lock(self):
                 raise TypeError("Session messages cannot be renewed. Please renew the Session lock instead.")
         except AttributeError:
             pass
-        self._is_live(MESSAGE_RENEW_LOCK)
+        self._check_live(MESSAGE_RENEW_LOCK)
         token = self.lock_token
         if not token:
             raise ValueError("Unable to renew lock - no lock token found.")

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

@@ -16,7 +16,6 @@
     ASSOCIATEDLINKPROPERTYNAME
 )
 from ..exceptions import (
-    InvalidHandlerState,
     ServiceBusError,
     _create_servicebus_exception
 )
@@ -139,8 +138,6 @@ async def _do_retryable_operation(self, operation, timeout=None, **kwargs):
 
     async def _mgmt_request_response(self, mgmt_operation, message, callback, keep_alive_associated_link=True, **kwargs):
         await self._open()
-        if not self._running:
-            raise InvalidHandlerState("Client connection is closed.")
 
         application_properties = {}
         # Some mgmt calls do not support an associated link name (such as list_sessions).  Most do, so on by default.

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

@@ -125,7 +125,7 @@ def __init__(
         self._connection = kwargs.get("connection")
 
     async def __anext__(self):
-        self._can_run()
+        self._check_live()
         while True:
             try:
                 return await self._do_retryable_operation(self._iter_next)
@@ -170,7 +170,7 @@ async def _open(self):
                 await asyncio.sleep(0.05)
             self._running = True
         except:
-            self.close()
+            await self.close()
             raise
 
     async def _receive(self, max_batch_size=None, timeout=None):
@@ -295,7 +295,10 @@ async def receive(self, max_batch_size=None, max_wait_time=None):
                 :caption: Receive messages from ServiceBus.
 
         """
-        self._can_run()
+        self._check_live()
+        if max_batch_size and self._config.prefetch < max_batch_size:
+            raise ValueError("max_batch_size should be less than or equal to prefetch of ServiceBusReceiver, or you "
+                             "could set a larger prefetch value when you're constructing the ServiceBusReceiver.")
         return await self._do_retryable_operation(
             self._receive,
             max_batch_size=max_batch_size,
@@ -324,7 +327,7 @@ async def receive_deferred_messages(self, sequence_numbers):
                 :caption: Receive deferred messages from ServiceBus.
 
         """
-        self._can_run()
+        self._check_live()
         if not sequence_numbers:
             raise ValueError("At least one sequence number must be specified.")
         await self._open()
@@ -369,7 +372,7 @@ async def peek(self, message_count=1, sequence_number=0):
                 :dedent: 4
                 :caption: Peek messages in the queue.
         """
-        self._can_run()
+        self._check_live()
         if not sequence_number:
             sequence_number = self._last_received_sequenced_number or 1
         if int(message_count) < 1:

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

@@ -12,9 +12,6 @@
 from .._common.message import Message, BatchMessage
 from .._servicebus_sender import SenderMixin
 from ._base_handler_async import BaseHandlerAsync
-from ..exceptions import (
-    MessageSendFailed
-)
 from .._common.constants import (
     REQUEST_RESPONSE_SCHEDULE_MESSAGE_OPERATION,
     REQUEST_RESPONSE_CANCEL_SCHEDULED_MESSAGE_OPERATION,
@@ -128,16 +125,13 @@ async def _open(self):
             self._max_message_size_on_link = self._handler.message_handler._link.peer_max_message_size \
                                              or uamqp.constants.MAX_MESSAGE_LENGTH_BYTES
         except:
-            self.close()
+            await self.close()
             raise
 
     async def _send(self, message, timeout=None, last_exception=None):
         await self._open()
         self._set_msg_timeout(timeout, last_exception)
-        try:
-            await self._handler.send_message_async(message.message)
-        except Exception as e:
-            raise MessageSendFailed(e)
+        await self._handler.send_message_async(message.message)
 
     async def _schedule(self, message, schedule_time_utc):
         # type: (Union[Message, BatchMessage], datetime.datetime) -> List[int]
@@ -267,8 +261,10 @@ async def send(self, message):
             batch = await self.create_batch()
             batch._from_list(message)
             message = batch
-        except TypeError: # Message was not a list or generator.
+        except TypeError:  # Message was not a list or generator.
             pass
+        if isinstance(message, BatchMessage) and len(message) == 0:
+            raise ValueError("A BatchMessage or list of Message must have at least one Message")
 
         await self._do_retryable_operation(
             self._send,