Azure Service Bus主题的EventGridTrigger

Question

我基于EventGrid Trigger创建了一个Azure功能。每当新消息到达服务总线主题时，该触发器将触发。下面是生成的函数模板

using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

public static void Run(JObject eventGridEvent, TraceWriter log)
{
    log.Info(eventGridEvent.ToString(Formatting.Indented));
}

我对Azure功能的要求是处理数据并将其存储在ADLS中。现在我如何从JObject类型解析/反序列化数据。我需要在将此函数持久化到Data Lake存储之前对其进行规范化。我需要覆盖这个功能吗？

请提供一些细节/参考以满足此要求

Answer 1

Service Bus（Premium）为两种情况发送事件：

1. ActiveMessagesWithNoListenersAvailable
2. DeadletterMessagesAvailable

当存在与特定实体关联的消息且不存在活动侦听器时，将发出第一个事件。实体将在有效载荷中指示，以及访问它的其他所需信息（例如命名空间或要从中接收的订阅主题）。架构在documentation中定义。

第二个事件模式与第一个类似，并且针对死信字母队列发出。

现在我如何从JObject类型解析/反序列化数据。我需要在将此函数持久化到Data Lake存储之前对其进行规范化。我需要覆盖这个功能吗？

eventGridEvent JSON本身不会为您提供Azure Service Bus消息。您需要先了解原始邮件是如何序列化的，即发件人方使用的是什么。反序列化将需要进入函数，然后是编写对象Data Lake的代码。

Answer 2

除了Sean的回答，Azure Service Bus与AEG的集成还可以为ASB实体构建一些监视功能。请注意，此集成与存储blob帐户不同，每次创建/删除blob时都会发布事件。换句话说，ASB不会为到达ASB实体的每条消息发布事件，事件像实体监视器一样发布。

这种实体看门狗使用以下逻辑：

1. 当实体中没有消息时，不会发布任何事件。
2. 当第一条消息到达实体并且实体上没有360+秒的活动侦听器时，会立即发布事件
3. 当侦听器仍处于非活动状态且实体中至少有一条消息时，事件每120秒发布一次
4. 事件在360秒监听器空闲（非活动）时间之后发布，并且实体中仍然至少有一条消息。例如，如果我们在实体中有5条消息，并且订户将使用REST Api仅提取一条消息，则下一个事件将在360秒后发布。换句话说，监视器实体允许将监听器保持在空闲时间360秒。

基于上述“监视程序实体”行为，此功能看起来更适合于缓慢的流量消息传递，例如ASB实体上的唤醒和监听侦听器。

注意，可以使用订阅级别的短重试时间策略来避免侦听器的360秒空​​闲时间，因此可以在5分钟重试时间内再次调用订户3次。

出于测试目的，以下是用于ASB事件订阅者的EventGridTrigger函数的代码片段。

#r "..\\bin\\Microsoft.Azure.ServiceBus.dll"
#r "Newtonsoft.Json"

using System;
using System.Threading.Tasks;
using System.Text;
using System.Linq;
using System.Net;
using System.Net.Http;
using System.Web;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Microsoft.Azure.ServiceBus.Primitives;



// sasToken cache
static SasTokenHelper helper = new SasTokenHelper(Environment.GetEnvironmentVariable("AzureServiceBusConnectionString"));

public static async Task Run(JObject eventGridEvent, ILogger log)
{
    log.LogInformation(eventGridEvent.ToString());

    // from the eventgrid payload
    var requestUri = $"{eventGridEvent["data"]?["requestUri"]?.Value<string>()}?api-version=2015-01";

    using (var client = new HttpClient())
    {

        client.DefaultRequestHeaders.Add("Authorization", helper.GetSasToken());

        do
        {
            // read & delete the message 
            var response = await client.DeleteAsync(requestUri);

            // check for message
            if (response.StatusCode != HttpStatusCode.OK)
            {
                log.LogWarning($">>> No message <<<");
                break;
            }

            // message body
            string jsontext = await response.Content.ReadAsStringAsync();

            // show the message
            log.LogInformation($"\nHeaders:\n\t{string.Join("\n\t", response.Headers.Select(i => $"{i.Key}={i.Value.First()}"))}\nBody:\n\t{jsontext}");
        } while (true);

    }

    await Task.CompletedTask;
}




// helpers
class SasTokenHelper
{
    DateTime expiringSaS;
    uint sasTTLInMinutes = 10;
    string sasToken = string.Empty;
    (string hostname, string keyname, string key) config;

    public SasTokenHelper(string connectionString)
    {
        config = GetPartsFromConnectionString(connectionString);
        GetSasToken();
    }

    public string GetSasToken()
    {
        lock (sasToken)
        {
            if (expiringSaS < DateTime.UtcNow.AddMinutes(-1))
            {
                this.sasToken = GetSASToken(config.hostname, config.key, config.keyname, sasTTLInMinutes);
                expiringSaS = DateTime.UtcNow.AddMinutes(sasTTLInMinutes);
            }
            return sasToken;
        }
    }

    internal (string hostname, string keyname, string key) GetPartsFromConnectionString(string connectionString)
    {
        var parts = connectionString.Split(new[] { ';' }, StringSplitOptions.RemoveEmptyEntries).Select(s => s.Split(new[] { '=' }, 2)).ToDictionary(x => x[0].Trim(), x => x[1].Trim());
        return (parts["Endpoint"] ?? "", parts["SharedAccessKeyName"] ?? "", parts["SharedAccessKey"] ?? "");
    }

    internal string GetSASToken(string resourceUri, string key, string keyName = null, uint minutes = 10)
    {
        var tp = SharedAccessSignatureTokenProvider.CreateSharedAccessSignatureTokenProvider(keyName, key, TimeSpan.FromMinutes(minutes));
        return tp.GetTokenAsync(resourceUri, TimeSpan.FromSeconds(60)).Result.TokenValue;
    }
}