[ServiceBus] Support for scheduling and cancellation

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -6,6 +6,9 @@
 
 * 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.
+* Added support for scheduling messages and scheduled message cancellation.
+    - Use `ServiceBusSender.schedule` for scheduling messages.
+    - Use `ServiceBusSender.cancel_scheduled_messages` for scheduled messages cancellation.
 
 **BugFixes**
 

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

@@ -240,27 +240,39 @@ def time_to_live(self, value):
             self.header.time_to_live = int(value) * 1000
 
     @property
-    def body(self):
-        # type: () -> Union[bytes, Generator[bytes]]
-        """The body of the Message.
+    def scheduled_enqueue_time_utc(self):
+        # type: () -> Optional[datetime.datetime]
+        """Get or set the utc scheduled enqueue time to the message.
+        This property can be used for scheduling when sending a message through `ServiceBusSender.send` method.
+        Or you could use `ServiceBusSender.schedule` method to schedule messages which would return scheduled messages'
+        sequence numbers that can be used for future cancellation.
 
-        :rtype: bytes or generator[bytes]
+        :rtype: ~datetime.datetime
         """
-        return self.message.get_data()
+        if self.message.annotations:
+            timestamp = self.message.annotations.get(_X_OPT_SCHEDULED_ENQUEUE_TIME)
+            if timestamp:
+                in_seconds = timestamp/1000.0
+                return utc_from_timestamp(in_seconds)
+        return None
 
-    def schedule(self, schedule_time_utc):
+    @scheduled_enqueue_time_utc.setter
+    def scheduled_enqueue_time_utc(self, value):
         # type: (datetime.datetime) -> None
-        """Add a specific utc enqueue time to the message.
-
-        :param schedule_time_utc: The scheduled utc time to enqueue the message.
-        :type schedule_time_utc: ~datetime.datetime
-        :rtype: None
-        """
         if not self.properties.message_id:
             self.properties.message_id = str(uuid.uuid4())
         if not self.message.annotations:
             self.message.annotations = {}
-        self.message.annotations[types.AMQPSymbol(_X_OPT_SCHEDULED_ENQUEUE_TIME)] = schedule_time_utc
+        self.message.annotations[types.AMQPSymbol(_X_OPT_SCHEDULED_ENQUEUE_TIME)] = value
+
+    @property
+    def body(self):
+        # type: () -> Union[bytes, Generator[bytes]]
+        """The body of the Message.
+
+        :rtype: bytes or generator[bytes]
+        """
+        return self.message.get_data()
 
 
 class BatchMessage(object):
@@ -399,20 +411,6 @@ def enqueued_time_utc(self):
                 return utc_from_timestamp(in_seconds)
         return None
 
-    @property
-    def scheduled_enqueue_time_utc(self):
-        # type: () -> Optional[datetime.datetime]
-        """
-
-        :rtype: ~datetime.datetime
-        """
-        if self.message.annotations:
-            timestamp = self.message.annotations.get(_X_OPT_SCHEDULED_ENQUEUE_TIME)
-            if timestamp:
-                in_seconds = timestamp/1000.0
-                return utc_from_timestamp(in_seconds)
-        return None
-
     @property
     def sequence_number(self):
         # type: () -> Optional[int]

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

@@ -65,7 +65,7 @@ def _set_msg_timeout(self, timeout=None, last_exception=None):
     def _build_schedule_request(cls, schedule_time_utc, *messages):
         request_body = {MGMT_REQUEST_MESSAGES: []}
         for message in messages:
-            message.schedule(schedule_time_utc)
+            message.scheduled_enqueue_time_utc = schedule_time_utc
             message_data = {}
             message_data[MGMT_REQUEST_MESSAGE_ID] = message.properties.message_id
             if message.properties.group_id:
@@ -188,15 +188,15 @@ def _send(self, message, timeout=None, last_exception=None):
         except Exception as e:
             raise MessageSendFailed(e)
 
-    def _schedule(self, message, schedule_time_utc):
-        # type: (Union[Message, BatchMessage], datetime.datetime) -> List[int]
-        """Send Message or BatchMessage to be enqueued at a specific time.
+    def schedule(self, messages, schedule_time_utc):
+        # type: (Union[Message, List[Message]], datetime.datetime) -> List[int]
+        """Send Message or multiple Messages to be enqueued at a specific time.
         Returns a list of the sequence numbers of the enqueued messages.
-        :param message: The messages to schedule.
-        :type message: ~azure.servicebus.Message or ~azure.servicebus.BatchMessage
+        :param messages: The message or list of messages to schedule.
+        :type messages: ~azure.servicebus.Message or List[~azure.servicebus.Message]
         :param schedule_time_utc: The utc date and time to enqueue the messages.
         :type schedule_time_utc: ~datetime.datetime
-        :rtype: List[int]
+        :rtype: list[int]
 
         .. admonition:: Example:
 
@@ -209,24 +209,26 @@ def _schedule(self, message, schedule_time_utc):
         """
         # pylint: disable=protected-access
         self._open()
-        if isinstance(message, BatchMessage):
-            request_body = self._build_schedule_request(schedule_time_utc, *message._messages)
+        if isinstance(messages, Message):
+            request_body = self._build_schedule_request(schedule_time_utc, messages)
         else:
-            request_body = self._build_schedule_request(schedule_time_utc, message)
+            request_body = self._build_schedule_request(schedule_time_utc, *messages)
         return self._mgmt_request_response_with_retry(
             REQUEST_RESPONSE_SCHEDULE_MESSAGE_OPERATION,
             request_body,
             mgmt_handlers.schedule_op
         )
 
-    def _cancel_scheduled_messages(self, sequence_numbers):
+    def cancel_scheduled_messages(self, sequence_numbers):
         # type: (Union[int, List[int]]) -> None
         """
         Cancel one or more messages that have previously been scheduled and are still pending.
 
         :param sequence_numbers: The sequence numbers of the scheduled messages.
-        :type sequence_numbers: int or list[int]
+        :type sequence_numbers: int or List[int]
         :rtype: None
+        :raises: ~azure.servicebus.exceptions.ServiceBusError if messages cancellation failed due to message already
+         cancelled or enqueued.
 
         .. admonition:: Example:
 

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

@@ -139,15 +139,15 @@ async def _send(self, message, timeout=None, last_exception=None):
         except Exception as e:
             raise MessageSendFailed(e)
 
-    async def _schedule(self, message, schedule_time_utc):
-        # type: (Union[Message, BatchMessage], datetime.datetime) -> List[int]
-        """Send Message or BatchMessage to be enqueued at a specific time.
+    async def schedule(self, messages, schedule_time_utc):
+        # type: (Union[Message, List[Message]], datetime.datetime) -> List[int]
+        """Send Message or multiple Messages to be enqueued at a specific time.
         Returns a list of the sequence numbers of the enqueued messages.
-        :param message: The messages to schedule.
-        :type message: ~azure.servicebus.Message or ~azure.servicebus.BatchMessage
+        :param messages: The message or list of messages to schedule.
+        :type messages: ~azure.servicebus.Message or List[~azure.servicebus.Message]
         :param schedule_time_utc: The utc date and time to enqueue the messages.
         :type schedule_time_utc: ~datetime.datetime
-        :rtype: List[int]
+        :rtype: list[int]
 
         .. admonition:: Example:
 
@@ -160,24 +160,26 @@ async def _schedule(self, message, schedule_time_utc):
         """
         # pylint: disable=protected-access
         await self._open()
-        if isinstance(message, BatchMessage):
-            request_body = self._build_schedule_request(schedule_time_utc, *message._messages)
+        if isinstance(messages, Message):
+            request_body = self._build_schedule_request(schedule_time_utc, messages)
         else:
-            request_body = self._build_schedule_request(schedule_time_utc, message)
+            request_body = self._build_schedule_request(schedule_time_utc, *messages)
         return await self._mgmt_request_response_with_retry(
             REQUEST_RESPONSE_SCHEDULE_MESSAGE_OPERATION,
             request_body,
             mgmt_handlers.schedule_op
         )
 
-    async def _cancel_scheduled_messages(self, sequence_numbers):
+    async def cancel_scheduled_messages(self, sequence_numbers):
         # type: (Union[int, List[int]]) -> None
         """
         Cancel one or more messages that have previously been scheduled and are still pending.
 
-        :param sequence_numbers: he sequence numbers of the scheduled messages.
-        :type sequence_numbers: int or list[int]
+        :param sequence_numbers: The sequence numbers of the scheduled messages.
+        :type sequence_numbers: int or List[int]
         :rtype: None
+        :raises: ~azure.servicebus.exceptions.ServiceBusError if messages cancellation failed due to message already
+         cancelled or enqueued.
 
         .. admonition:: Example:
 

sdk/servicebus/azure-servicebus/samples/README.md

@@ -33,6 +33,9 @@ Both [sync version](./sync_samples) and [async version](./async_samples) of samp
 - [session_send_receive.py](./sync_samples/session_send_receive.py) ([async_version](./async_samples/session_send_receive_async.py)) - Examples to send messages to and receive messages from a session-enabled service bus queue:
     - Send messages to a session-enabled queue
     - Receive messages from session-enabled queue
+- [schedule_messages_and_cancellation](./sync_samples/schedule_messages_and_cancellation.py) ([async_version](./async_samples/schedule_messages_and_cancellation_async.py)) - Examples to schedule messages and cancel scheduled message:
+    - Schedule a single message or multiples messages to a queue
+    - Cancel scheduled messages from a queue
 - [client_identity_authentication.py](./sync_samples/client_identity_authentication.py) ([async_version](./async_samples/client_identity_authentication_async.py)) - Examples to authenticate the client by Azure Activate Directory
     - Authenticate and create the client utilizing the `azure.identity` library
 - [proxy.py](./sync_samples/proxy.py) ([async_version](./async_samples/proxy_async.py)) - Examples to send message behind a proxy:

sdk/servicebus/azure-servicebus/samples/async_samples/sample_code_servicebus_async.py

@@ -12,6 +12,7 @@
     - Receive and settle deferred messages
 """
 import os
+import datetime
 import asyncio
 from azure.servicebus.aio import ServiceBusClient
 from azure.servicebus import Message
@@ -96,7 +97,7 @@ async def example_create_servicebus_sender_async():
     topic_name = os.environ['SERVICE_BUS_TOPIC_NAME']
     servicebus_client = ServiceBusClient.from_connection_string(conn_str=servicebus_connection_str)
     async with servicebus_client:
-        queue_sender = servicebus_client.get_topic_sender(topic_name=topic_name)
+        topic_sender = servicebus_client.get_topic_sender(topic_name=topic_name)
     # [END create_topic_sender_from_sb_client_async]
 
     return queue_sender
@@ -271,8 +272,24 @@ async def example_session_ops_async():
                 break
 
 
+async def example_schedule_ops_async():
+    servicebus_sender = await example_create_servicebus_sender_async()
+    # [START scheduling_messages_async]
+    async with servicebus_sender:
+        scheduled_time_utc = datetime.datetime.utcnow() + datetime.timedelta(seconds=30)
+        scheduled_messages = [Message("Scheduled message") for _ in range(10)]
+        sequence_nums = await servicebus_sender.schedule(scheduled_messages, scheduled_time_utc)
+    # [END scheduling_messages_async]
+
+    # [START cancel_scheduled_messages_async]
+    async with servicebus_sender:
+        await servicebus_sender.cancel_scheduled_messages(sequence_nums)
+    # [END cancel_scheduled_messages_async]
+
+
 if __name__ == "__main__":
     loop = asyncio.get_event_loop()
     loop.run_until_complete(example_send_and_receive_async())
     loop.run_until_complete(example_receive_deferred_async())
+    loop.run_until_complete(example_schedule_ops_async())
     # loop.run_until_complete(example_session_ops_async())

sdk/servicebus/azure-servicebus/samples/async_samples/schedule_messages_and_cancellation_async.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python
+
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""
+Example to show scheduling messages to and cancelling messages from a Service Bus Queue asynchronously.
+"""
+
+# pylint: disable=C0111
+
+import os
+import asyncio
+import datetime
+from azure.servicebus.aio import ServiceBusClient
+from azure.servicebus import Message
+
+CONNECTION_STR = os.environ['SERVICE_BUS_CONNECTION_STR']
+QUEUE_NAME = os.environ["SERVICE_BUS_QUEUE_NAME"]
+
+
+async def schedule_single_message(sender):
+    message = Message("Message to be scheduled")
+    scheduled_time_utc = datetime.datetime.utcnow() + datetime.timedelta(seconds=30)
+    sequence_number = await sender.schedule(message, scheduled_time_utc)
+    return sequence_number
+
+
+async def schedule_multiple_messages(sender):
+    messages_to_schedule = []
+    for _ in range(10):
+        messages_to_schedule.append(Message("Message to be scheduled"))
+
+    scheduled_time_utc = datetime.datetime.utcnow() + datetime.timedelta(seconds=30)
+    sequence_numbers = await sender.schedule(messages_to_schedule, scheduled_time_utc)
+    return sequence_numbers
+
+
+async def main():
+    servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR, logging_enable=True)
+    async with servicebus_client:
+        sender = servicebus_client.get_queue_sender(queue_name=QUEUE_NAME)
+        async with sender:
+            sequence_number = await schedule_single_message(sender)
+            print("Single message is scheduled and sequence number is {}".format(sequence_number))
+            sequence_numbers = await schedule_multiple_messages(sender)
+            print("Multiple messages are scheduled and sequence numbers are {}".format(sequence_numbers))
+
+            await sender.cancel_scheduled_messages(sequence_number)
+            await sender.cancel_scheduled_messages(sequence_numbers)
+            print("All scheduled messages are cancelled.")
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(main())

sdk/servicebus/azure-servicebus/samples/sync_samples/sample_code_servicebus.py

@@ -93,7 +93,7 @@ def example_create_servicebus_sender_sync():
     topic_name = os.environ['SERVICE_BUS_TOPIC_NAME']
     servicebus_client = ServiceBusClient.from_connection_string(conn_str=servicebus_connection_str)
     with servicebus_client:
-        queue_sender = servicebus_client.get_topic_sender(topic_name=topic_name)
+        topic_sender = servicebus_client.get_topic_sender(topic_name=topic_name)
     # [END create_topic_sender_from_sb_client_sync]
 
     return queue_sender
@@ -295,6 +295,22 @@ def example_session_ops_sync():
                 break
 
 
+def example_schedule_ops_sync():
+    servicebus_sender = example_create_servicebus_sender_sync()
+    # [START scheduling_messages]
+    with servicebus_sender:
+        scheduled_time_utc = datetime.datetime.utcnow() + datetime.timedelta(seconds=30)
+        scheduled_messages = [Message("Scheduled message") for _ in range(10)]
+        sequence_nums = servicebus_sender.schedule(scheduled_messages, scheduled_time_utc)
+    # [END scheduling_messages]
+
+    # [START cancel_scheduled_messages]
+    with servicebus_sender:
+        servicebus_sender.cancel_scheduled_messages(sequence_nums)
+    # [END cancel_scheduled_messages]
+
+
 example_send_and_receive_sync()
 example_receive_deferred_sync()
+example_schedule_ops_sync()
 # example_session_ops_sync()

sdk/servicebus/azure-servicebus/samples/sync_samples/schedule_messages_and_cancellation.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python
+
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""
+Example to show scheduling messages to and cancelling messages from a Service Bus Queue.
+"""
+
+# pylint: disable=C0111
+
+import os
+import datetime
+from azure.servicebus import ServiceBusClient, Message
+
+CONNECTION_STR = os.environ['SERVICE_BUS_CONNECTION_STR']
+QUEUE_NAME = os.environ["SERVICE_BUS_QUEUE_NAME"]
+
+
+def schedule_single_message(sender):
+    message = Message("Message to be scheduled")
+    scheduled_time_utc = datetime.datetime.utcnow() + datetime.timedelta(seconds=30)
+    sequence_number = sender.schedule(message, scheduled_time_utc)
+    return sequence_number
+
+
+def schedule_multiple_messages(sender):
+    messages_to_schedule = []
+    for _ in range(10):
+        messages_to_schedule.append(Message("Message to be scheduled"))
+
+    scheduled_time_utc = datetime.datetime.utcnow() + datetime.timedelta(seconds=30)
+    sequence_numbers = sender.schedule(messages_to_schedule, scheduled_time_utc)
+    return sequence_numbers
+
+
+servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR, logging_enable=True)
+with servicebus_client:
+    sender = servicebus_client.get_queue_sender(queue_name=QUEUE_NAME)
+    with sender:
+        sequence_number = schedule_single_message(sender)
+        print("Single message is scheduled and sequence number is {}".format(sequence_number))
+        sequence_numbers = schedule_multiple_messages(sender)
+        print("Multiple messages are scheduled and sequence numbers are {}".format(sequence_numbers))
+
+        sender.cancel_scheduled_messages(sequence_number)
+        sender.cancel_scheduled_messages(sequence_numbers)
+        print("All scheduled messages are cancelled.")