Add capability to send multiple events in one Send() call.

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -6,6 +6,7 @@
 
 * 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.
+* `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.
 
 **BugFixes**
 

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

@@ -291,14 +291,19 @@ def from_connection_string(
         return cls(**constructor_args)
 
     def send(self, message):
-        # type: (Union[Message, BatchMessage]) -> None
+        # type: (Union[Message, BatchMessage, List[Message]]) -> None
         """Sends message and blocks until acknowledgement is received or operation times out.
 
+        If a list of messages was provided, attempts to send them as a single batch, throwing a
+        `ValueError` if they cannot fit in a single batch.
+
         :param message: The ServiceBus message to be sent.
         :type message: ~azure.servicebus.Message or ~azure.servicebus.BatchMessage
         :rtype: None
-        :raises: ~azure.servicebus.common.errors.MessageSendFailed if the message fails to
-         send or ~azure.servicebus.common.errors.OperationTimeoutError if sending times out.
+        :raises: :class: ~azure.servicebus.common.errors.MessageSendFailed if the message fails to
+         send
+                :class: ~azure.servicebus.common.errors.OperationTimeoutError if sending times out.
+                :class: `ValueError` if list of messages is provided and cannot fit in a batch.
 
         .. admonition:: Example:
 
@@ -310,6 +315,14 @@ def send(self, message):
                 :caption: Send message.
 
         """
+        try:
+            batch = self.create_batch()
+            for each in message:
+                batch.add(each)
+            message = batch
+        except TypeError: # Message was not a list or generator.
+            pass
+
         self._do_retryable_operation(
             self._send,
             message=message,

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

@@ -239,14 +239,19 @@ def from_connection_string(
         return cls(**constructor_args)
 
     async def send(self, message):
-        # type: (Union[Message, BatchMessage]) -> None
+        # type: (Union[Message, BatchMessage, List[Message]]) -> None
         """Sends message and blocks until acknowledgement is received or operation times out.
 
+        If a list of messages was provided, attempts to send them as a single batch, throwing a
+        `ValueError` if they cannot fit in a single batch.
+
         :param message: The ServiceBus message to be sent.
         :type message: ~azure.servicebus.Message or ~azure.servicebus.BatchMessage
         :rtype: None
-        :raises: ~azure.servicebus.common.errors.MessageSendFailed if the message fails to
-         send or ~azure.servicebus.common.errors.OperationTimeoutError if sending times out.
+        :raises: :class: ~azure.servicebus.common.errors.MessageSendFailed if the message fails to
+         send
+                :class: ~azure.servicebus.common.errors.OperationTimeoutError if sending times out.
+                :class: `ValueError` if list of messages is provided and cannot fit in a batch.
 
         .. admonition:: Example:
 
@@ -258,6 +263,14 @@ async def send(self, message):
                 :caption: Send message.
 
         """
+        try:
+            batch = await self.create_batch()
+            for each in message:
+                batch.add(each)
+            message = batch
+        except TypeError: # Message was not a list or generator.
+            pass
+
         await self._do_retryable_operation(
             self._send,
             message=message,

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

@@ -66,6 +66,33 @@ async def test_async_queue_by_queue_client_conn_str_receive_handler_peeklock(sel
 
             assert count == 10
 
+
+    @pytest.mark.liveTest
+    @pytest.mark.live_test_only
+    @CachedResourceGroupPreparer(name_prefix='servicebustest')
+    @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
+    @ServiceBusQueuePreparer(name_prefix='servicebustest', dead_lettering_on_message_expiration=True)
+    async def test_async_queue_by_queue_client_send_multiple_messages(self, servicebus_namespace_connection_string, servicebus_queue, **kwargs):
+        async with ServiceBusClient.from_connection_string(
+            servicebus_namespace_connection_string, logging_enable=False) as sb_client:
+
+            async with sb_client.get_queue_sender(servicebus_queue.name) as sender:
+                messages = []
+                for i in range(10):
+                    message = Message("Handler message no. {}".format(i))
+                    messages.append(message)
+                await sender.send(messages)
+
+            async with sb_client.get_queue_receiver(servicebus_queue.name, idle_timeout=5) as receiver:
+                count = 0
+                async for message in receiver:
+                    print_message(_logger, message)
+                    count += 1
+                    await message.complete()
+
+                assert count == 10
+
+
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
     @CachedResourceGroupPreparer()
@@ -688,7 +715,6 @@ async def test_async_queue_by_queue_client_conn_str_receive_handler_with_autoloc
             await renewer.shutdown()
             assert len(messages) == 11
 
-    @pytest.mark.skip(reason='requires queuing messages')
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
     @CachedResourceGroupPreparer(name_prefix='servicebustest')
@@ -703,13 +729,18 @@ async def test_async_queue_by_servicebus_client_fail_send_messages(self, service
             async with sb_client.get_queue_sender(servicebus_queue.name) as sender:
                 with pytest.raises(MessageSendFailed):
                     await sender.send(Message(too_large))
-
-            async with sb_client.get_queue_sender(servicebus_queue.name) as sender:
-                sender.queue_message(Message(too_large))
-                results = await sender.send_pending_messages()
-                assert len(results) == 1
-                assert not results[0][0]
-                assert isinstance(results[0][1], MessageSendFailed)
+
+                half_too_large = "A" * int((1024 * 512) / 2)
+                with pytest.raises(ValueError):
+                    await sender.send([Message(half_too_large), Message(half_too_large)])
+
+            # TODO: Reenable this when queue_message exists.
+            #async with sb_client.get_queue_sender(servicebus_queue.name) as sender:
+            #    sender.queue_message(Message(too_large))
+            #    results = await sender.send_pending_messages()
+            #    assert len(results) == 1
+            #    assert not results[0][0]
+            #    assert isinstance(results[0][1], MessageSendFailed)
 
     @pytest.mark.liveTest
     @pytest.mark.live_test_only

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

@@ -114,6 +114,30 @@ def test_queue_by_queue_client_conn_str_receive_handler_peeklock(self, servicebu
 
             assert count == 10
 
+    @pytest.mark.liveTest
+    @pytest.mark.live_test_only
+    @CachedResourceGroupPreparer(name_prefix='servicebustest')
+    @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
+    @ServiceBusQueuePreparer(name_prefix='servicebustest', dead_lettering_on_message_expiration=True)
+    def test_queue_by_queue_client_send_multiple_messages(self, servicebus_namespace_connection_string, servicebus_queue, **kwargs):
+        with ServiceBusClient.from_connection_string(
+            servicebus_namespace_connection_string, logging_enable=False) as sb_client:
+
+            with sb_client.get_queue_sender(servicebus_queue.name) as sender:
+                messages = []
+                for i in range(10):
+                    message = Message("Handler message no. {}".format(i))
+                    messages.append(message)
+                sender.send(messages)
+
+            with sb_client.get_queue_receiver(servicebus_queue.name, idle_timeout=5) as receiver:
+                count = 0
+                for message in receiver:
+                    print_message(_logger, message)
+                    count += 1
+                    message.complete()
+
+                assert count == 10
 
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
@@ -673,34 +697,33 @@ def test_queue_by_servicebus_client_browse_empty_messages(self, servicebus_names
                 assert len(messages) == 0
 
 
-    @pytest.mark.skip(reason="Pending queue message")
     @pytest.mark.liveTest
     @pytest.mark.live_test_only
     @CachedResourceGroupPreparer(name_prefix='servicebustest')
     @CachedServiceBusNamespacePreparer(name_prefix='servicebustest')
     @CachedServiceBusQueuePreparer(name_prefix='servicebustest', dead_lettering_on_message_expiration=True)
-    def test_queue_by_servicebus_client_fail_send_messages(self, servicebus_namespace, servicebus_namespace_key_name, servicebus_namespace_primary_key, servicebus_queue, **kwargs):
+    def test_queue_by_servicebus_client_fail_send_messages(self, servicebus_namespace_connection_string, servicebus_queue, **kwargs):
 
         with ServiceBusClient.from_connection_string(
             servicebus_namespace_connection_string, logging_enable=False) as sb_client:
 
             too_large = "A" * 1024 * 512
-            with sb_client.get_queue_sender(servicebus_queue.name) as sender:
-                try:
-                    results = sender.send(Message(too_large))
-                except MessageSendFailed:
-                    pytest.skip("Open issue for uAMQP on OSX")
 
             with sb_client.get_queue_sender(servicebus_queue.name) as sender:
                 with pytest.raises(MessageSendFailed):
                     sender.send(Message(too_large))
-
-            with sb_client.get_queue_sender(servicebus_queue.name) as sender:
-                sender.queue_message(Message(too_large))
-                results = sender.send_pending_messages()
-                assert len(results) == 1
-                assert not results[0][0]
-                assert isinstance(results[0][1], MessageSendFailed)
+
+                half_too_large = "A" * int((1024 * 512) / 2)
+                with pytest.raises(ValueError):
+                    sender.send([Message(half_too_large), Message(half_too_large)])
+
+            # TODO: Reenable once queue_message is added.
+            #with sb_client.get_queue_sender(servicebus_queue.name) as sender:
+            #    sender.queue_message(Message(too_large))
+            #    results = sender.send_pending_messages()
+            #    assert len(results) == 1
+            #    assert not results[0][0]
+            #    assert isinstance(results[0][1], MessageSendFailed)
 
 
     @pytest.mark.liveTest