azure - 使用 Azure 函数和 HTTP 触发器在 Azure 服务总线订阅中获取下一条消息

Question

我想创建一个满足以下要求的 Azure 函数：

• 触发 HTTP 请求
• 查看 Azure 服务总线订阅并根据 HTTP 请求中指定的一组筛选器获取下一条消息。

Answer 1

如果您使用的是 C#，则可以执行以下操作：

using System;
using System.IO;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Microsoft.Azure.ServiceBus;
using Microsoft.Azure.ServiceBus.Core;

namespace HttpTriggerSBRead
{
    public static class ReadSBOnHttpTrigger
    {
        const string ServiceBusConnectionString = "{service bus connection string}";
        const string TopicName = "{name of your topic}";
        const string SubscriptionName = "{name of your subscription}";

        [FunctionName("ReadSBOnHttpTrigger")]
        public static async Task<IActionResult> Run(
            [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post", Route = null)] HttpRequest req,
            ILogger log)
        {
            string filter = req.Query["filter"];

            string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
            dynamic data = JsonConvert.DeserializeObject(requestBody);
            filter = filter ?? data?.filter;

            SubscriptionClient sqlFilterOnlySubscriptionClient = new SubscriptionClient(ServiceBusConnectionString,
                TopicName, SubscriptionName);

            await sqlFilterOnlySubscriptionClient.AddRuleAsync(new RuleDescription
            {
                Filter = new SqlFilter(filter),
                Name = filter
            });
            await ReceiveMessagesAsync(SubscriptionName, log);
            await sqlFilterOnlySubscriptionClient.RemoveRuleAsync(filter);

            return filter != null
                ? (ActionResult)new OkObjectResult($"{filter}")
                : new BadRequestObjectResult("Please pass a filter on the query string or in the request body");
        }

        static async Task ReceiveMessagesAsync(string subscriptionName, ILogger log)
        {
            string subscriptionPath = EntityNameHelper.FormatSubscriptionPath(TopicName, subscriptionName);
            IMessageReceiver subscriptionReceiver = new MessageReceiver(ServiceBusConnectionString, subscriptionPath, ReceiveMode.ReceiveAndDelete);

            log.LogInformation($"{DateTime.Now} :: Receiving Messages From Subscription: {subscriptionName}");
            var receivedMessage = await subscriptionReceiver.ReceiveAsync(TimeSpan.FromSeconds(30));

            if (receivedMessage != null)
            {
                log.LogInformation($"Lable Property = {receivedMessage.Label}");
            }

            log.LogInformation($"{DateTime.Now} :: Messages From Subscription: {subscriptionName}");
        }
    }
}

要使用它，您需要将过滤器参数传递给您的函数。例如这样的：

http://localhost:7071/api/ReadSBOnHttpTrigger?filter=sys.Label=%27test%27或http://localhost:7071/api/ReadSBOnHttpTrigger?filter=sys.To=%27test%27

仅供参考，我将此代码用作源代码并进行了一些小的修改： https ://github.com/Azure/azure-service-bus/tree/master/samples/DotNet/GettingStarted/Microsoft.Azure.ServiceBus/TopicSubscriptionWithRuleOperationsSample