Add Azure Service Bus Queue and Subscription triggers for async message processing

docs/spelling_wordlist.txt

@@ -1623,6 +1623,8 @@ serializer
 serializers
 serverless
 ServiceAccount
+servicebus
+ServiceBusReceivedMessage
 ServicePrincipalCredentials
 ServiceResource
 SES

providers/microsoft/azure/docs/connections/message_bus.rst

@@ -0,0 +1,99 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+.. _howto/trigger:azure_service_bus_queue:
+
+Microsoft Azure Service Bus Queue Trigger
+=========================================
+
+The Microsoft Azure Service Bus Queue trigger enables you to monitor Azure Service Bus queues for new messages and trigger DAG runs when messages arrive.
+
+Authenticating to Azure
+-----------------------
+
+The trigger uses the connection you have configured in Airflow to connect to Azure.
+Please refer to the :ref:`howto/connection:azure_service_bus` documentation for details on how to configure your connection.
+
+Using the Trigger
+-----------------
+
+This example shows how to use the ``AzureServiceBusQueueTrigger``.
+
+.. code-block:: python
+
+      from airflow.providers.microsoft.azure.triggers.message_bus import AzureServiceBusQueueTrigger
+
+      trigger = AzureServiceBusQueueTrigger(
+          queues=["my_queue"],
+          azure_service_bus_conn_id="azure_service_bus_default",
+          max_message_count=1,
+          poll_interval=5.0,
+      )
+
+      # The trigger will fire when a message is available in the queue.
+
+.. _howto/trigger:azure_service_bus_subscription:
+
+Microsoft Azure Service Bus Subscription Trigger
+================================================
+
+The Microsoft Azure Service Bus Subscription trigger enables you to monitor Azure Service Bus topic subscriptions for new messages and trigger DAG runs when messages arrive.
+
+Authenticating to Azure
+-----------------------
+
+The trigger uses the connection you have configured in Airflow to connect to Azure.
+Please refer to the :ref:`howto/connection:azure_service_bus` documentation for details on how to configure your connection.
+
+Using the Trigger
+-----------------
+
+This example shows how to use the ``AzureServiceBusSubscriptionTrigger``.
+
+.. code-block:: python
+
+      from airflow.providers.microsoft.azure.triggers.message_bus import AzureServiceBusSubscriptionTrigger
+
+      trigger = AzureServiceBusSubscriptionTrigger(
+          topics=["my_topic"],
+          subscription_name="my_subscription",
+          azure_service_bus_conn_id="azure_service_bus_default",
+          max_message_count=1,
+          poll_interval=5.0,
+      )
+
+      # The trigger will fire when a message is available in the topic subscription.
+
+Azure Service Bus Message Queue
+===============================
+
+Azure Service Bus Queue Provider
+--------------------------------
+
+Implemented by :class:`~airflow.providers.microsoft.azure.triggers.message_bus.AzureServiceBusQueueTrigger`
+
+The Azure Service Bus Queue Provider is a message queue provider that uses Azure Service Bus queues.
+It allows you to monitor Azure Service Bus queues for new messages and trigger DAG runs when messages arrive.
+
+
+Azure Service Bus Subscription Provider
+---------------------------------------
+
+Implemented by :class:`~airflow.providers.microsoft.azure.triggers.message_bus.AzureServiceBusSubscriptionTrigger`
+
+The Azure Service Bus Subscription Provider is a message queue provider that uses Azure Service Bus topic subscriptions.
+It allows you to monitor Azure Service Bus topic subscriptions for new messages and trigger DAG runs when messages arrive.

providers/microsoft/azure/provider.yaml

@@ -294,6 +294,9 @@ triggers:
   - integration-name: Microsoft Power BI
     python-modules:
       - airflow.providers.microsoft.azure.triggers.powerbi
+  - integration-name: Microsoft Azure Service Bus
+    python-modules:
+      - airflow.providers.microsoft.azure.triggers.message_bus
 
 transfers:
   - source-integration-name: Local

providers/microsoft/azure/src/airflow/providers/microsoft/azure/get_provider_info.py

@@ -281,6 +281,10 @@ def get_provider_info():
                 "integration-name": "Microsoft Power BI",
                 "python-modules": ["airflow.providers.microsoft.azure.triggers.powerbi"],
             },
+            {
+                "integration-name": "Microsoft Azure Service Bus",
+                "python-modules": ["airflow.providers.microsoft.azure.triggers.message_bus"],
+            },
         ],
         "transfers": [
             {

providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/asb.py

@@ -606,6 +606,60 @@ def receive_subscription_message(
             for msg in received_msgs:
                 self._process_message(msg, context, message_callback, subscription_receiver)
 
+    def read_message(
+        self,
+        queue_name: str,
+        max_wait_time: float | None = None,
+    ) -> ServiceBusReceivedMessage | None:
+        """
+        Read a single message from a Service Bus queue without callback processing.
+
+        :param queue_name: The name of the queue to read from.
+        :param max_wait_time: Maximum time to wait for messages (seconds).
+        :return: The received message or None if no message is available.
+        """
+        with (
+            self.get_conn() as service_bus_client,
+            service_bus_client.get_queue_receiver(queue_name=queue_name) as receiver,
+            receiver,
+        ):
+            received_msgs = receiver.receive_messages(max_message_count=1, max_wait_time=max_wait_time)
+            if received_msgs:
+                msg = received_msgs[0]
+                receiver.complete_message(msg)
+                return msg
+            return None
+
+    def read_subscription_message(
+        self,
+        topic_name: str,
+        subscription_name: str,
+        max_wait_time: float | None = None,
+    ) -> ServiceBusReceivedMessage | None:
+        """
+        Read a single message from a Service Bus topic subscription without callback processing.
+
+        :param topic_name: The name of the topic.
+        :param subscription_name: The name of the subscription.
+        :param max_wait_time: Maximum time to wait for messages (seconds).
+        :return: The received message or None if no message is available.
+        """
+        with (
+            self.get_conn() as service_bus_client,
+            service_bus_client.get_subscription_receiver(
+                topic_name, subscription_name
+            ) as subscription_receiver,
+            subscription_receiver,
+        ):
+            received_msgs = subscription_receiver.receive_messages(
+                max_message_count=1, max_wait_time=max_wait_time
+            )
+            if received_msgs:
+                msg = received_msgs[0]
+                subscription_receiver.complete_message(msg)
+                return msg
+            return None
+
     def _process_message(
         self,
         msg: ServiceBusReceivedMessage,