使用 Azure 函数 [EventHubOutput] 返回具有属性的 EventData 对象

Question

我想在使用

IFunctionsWorkerMiddleware

的 Azure Function 触发器的输出上的

EventHubOutput

中设置相关 ID。我似乎找不到访问使用

EventHubOutput

时发回的事件数据的方法。

我尝试操纵

InvocationResult

但那是从触发器返回的任何对象（无法访问元数据）。 我还尝试从触发器返回

EventData

对象（与 EventHubTrigger 用作输入的类型相同），但这只是序列化为 EH 主题上的字符串

"Azure.Messaging.EventHubs.EventData"

，而不是实际的

EventData

对象

有什么想法如何在函数中间件的 EventHubOutput 绑定中设置标头/属性吗？

使用调用结果

[Function("MyFunction")]
[EventHubOutput("output", Connection = "output-connection")]
public async Task<string> Run([HttpTrigger(AuthorizationLevel.Anonymous, "post")] HttpRequest req)
{
    return "my payload";
}

public class CorrelationMiddleware : IFunctionsWorkerMiddleware
{
    public async Task Invoke(FunctionContext context, FunctionExecutionDelegate next)
    {
        await next(context);

        SetOutputCorrelation(context);
    }

    private void SetOutputCorrelation(FunctionContext context)
    {
        var correlationId = Guid.NewGuid().ToString();

        var result = context.GetInvocationResult();

        // result is just the string returned from trigger - no access to metadata
    }
}

返回事件数据

[Function("MyFunction")]
[EventHubOutput("output", Connection = "output-connection")]
public async Task<EventData> Run([HttpTrigger(AuthorizationLevel.Anonymous, "post")] HttpRequest req)
{
    var payload = "my payload";

    var eventData = new EventData(Encoding.UTF8.GetBytes(payload));

    // if return type is EventData, it doesnt get properly serialized on the EventHub topic, it's just a .ToString() so it ends up as a message with payload as string "Azure.Messaging.EventHubs.EventData"
    return eventData;
}

public class CorrelationMiddleware : IFunctionsWorkerMiddleware
{
    public async Task Invoke(FunctionContext context, FunctionExecutionDelegate next)
    {
        await next(context);

        SetOutputCorrelation(context);
    }

    private void SetOutputCorrelation(FunctionContext context)
    {
        var correlationId = Guid.NewGuid().ToString();

        var result = context.GetInvocationResult();

        if (result.Value is EventData eventData)
        {
            eventData.Properties["correlation-id"] = correlationId;
        }
    }
}

按照 Ikhtesam Afrin 的要求更新添加了 Program.cs

var builder = FunctionsApplication.CreateBuilder(args);

builder.ConfigureFunctionsWebApplication();

builder.UseMiddleware<CorrelationMiddleware>();

builder.Services
    .AddApplicationInsightsTelemetryWorkerService()
    .ConfigureFunctionsApplicationInsights();

builder.Build().Run();

更新

我认为这个问题的核心是

[EventHubOutput]

无法返回

EventData

对象。当返回

EventData

对象时，似乎只是通过执行

.ToString()

将其放在 EventHub 上，因为消息上的有效负载是

Azure.Messaging.EventHubs.EventData

。

[Function("Function1")]
[EventHubOutput("output", Connection = "output-connection")]
public EventData Run([HttpTrigger(AuthorizationLevel.Anonymous, "get")] HttpRequest req)
{
    var data = new EventData("test payload")
    {
        Properties =
        {
            { "CorrleationId", "123" }
        }
    };

    var str = data.ToString(); // -> "Azure.Messaging.EventHubs.EventData"

    return data;
}

我偶然发现这篇文章表明这应该以某种方式可能EventHubTrigger EventData[] 绑定不起作用

Answer 1

或者，我使用下面的代码使用输出绑定将有效负载和相关 ID 发送到事件中心。

using Azure.Messaging.EventHubs;
using Microsoft.AspNetCore.Http;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;
using System.Text;

namespace _79326949
{
    public class Function1
    {
        private readonly ILogger<Function1> _logger;

        public Function1(ILogger<Function1> logger)
        {
            _logger = logger;
        }

        [Function("MyFunction")]
        [EventHubOutput("output", Connection = "output-connection")]
        public async Task<EventDataResponse> Run([HttpTrigger(AuthorizationLevel.Anonymous, "post")] HttpRequest req)
        {
            var payload = "my payload";
            var eventData = new EventData(Encoding.UTF8.GetBytes(payload))
            {
                ContentType = "application/json", 
                CorrelationId = string.Empty 
            };

            return new EventDataResponse
            {
                EventBody = eventData.Data.ToString(),
                CorrelationId = eventData.CorrelationId
            };
        }
    }

    public class EventDataResponse
    {
        public string? EventBody { get; set; }
        public string? CorrelationId { get; set; }
    }
}

中间件代码：-

using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Middleware;

namespace _79326949
{
    public class CorrelationMiddleware : IFunctionsWorkerMiddleware
    {
        public async Task Invoke(FunctionContext context, FunctionExecutionDelegate next)
        {
            await next(context);

            SetOutputCorrelation(context);
        }

        private void SetOutputCorrelation(FunctionContext context)
        {
            var correlationId = Guid.NewGuid().ToString();
            var result = context.GetInvocationResult(); 

            if (result.Value is EventDataResponse eventData)
            {
                eventData.CorrelationId = correlationId;
            }
        }
    }
}

我收到的回复如下。