错误索引方法“IntegrationFunction”无法从类型“Microsoft.Azure.EventHubs.EventData”转换

Question

我很难理解为什么我的 Azure 函数收到以下错误。

错误：

Error indexing method 'IntegrationFunction' Can't convert from type 'Microsoft.Azure.EventHubs.EventData

该函数从 Azure 服务总线读取数据并进行转换，然后将其发布到 Azure 事件中心和另一个 Azure 服务总线。

功能：

[FunctionName("IntegrationFunction")]
[return: ServiceBus("%Transactions:SB:Transactions:ProcessorUpdateQueue%", Connection = "Transactions:SB:Transactions")]
public async Task<ServiceBusMessage> Run(
    [ServiceBusTrigger(
        "%Transactions:SB:Transactions:ReceiveTopic%",
        "%Transactions:SB:Transactions:AnalyticsTopicSubscription%",
        Connection = "Transactions:SB:Transactions")]
    string inputMsg,
    [EventHub("%Transactions:EVH:Transactions:Hub%", Connection = "Transactions:EVH:Transactions")] IAsyncCollector<EventData> outputEvent,
    ILogger log)
{
    if (string.IsNullOrWhiteSpace(inputMsg))
    {
        throw new ArgumentNullException(nameof(inputMsg));
    }

    log.LogInformation($"Service bus topic trigger function processing message: {inputMsg}");
    
    var retailTransaction = JsonConvert.DeserializeObject<RetailTransaction>(
        inputMsg,
        JsonSerialisationUtils.SerialiserSettings);
    
    if (retailTransaction == null)
    {
        throw new JsonException("Deserialized transaction was null");
    }

    try
    {
        // Transform
        var transaction = retailTransaction.ToDto();
        
        // Post to event hub
        await outputEvent.AddAsync(new EventData(ToByteArray(transaction)));

        log.LogInformation($"Transaction {transaction.TransactionNumber} processed.");

        // Post to service bus
        return retailTransaction.ToUpdateServiceBusMessage(processorId);
    }
    catch (Exception e)
    {
        log.LogError(e, "Error mapping transaction.");
    }

    return null;
}

Answer 1

对于扩展版本 5.x+，您需要使用

Azure.Messaging.EventHubs

。感谢 @Jesse Squire 的评论。

您也可以参考MS Doc，它也有同样的说法。

• 我有以下代码和给定的 .csproj 文件，它按预期工作。

using System;
using System.Text;
using System.Threading.Tasks;
using Azure.Messaging.EventHubs;
using Azure.Messaging.ServiceBus;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;

namespace _78237298
{
    public class IntegrationFunction
    {
        [FunctionName("IntegrationFunction")]
        [return: ServiceBus("%OtherServiceBus:QueueName%", Connection = "OtherServiceBus:ConnectionString")]
        public async Task<ServiceBusMessage> Run(
            [ServiceBusTrigger(
            "%Transactions:SB:Transactions:ReceiveTopic%", 
            "%Transactions:SB:Transactions:AnalyticsTopicSubscription%", 
            Connection = "Transactions:SB:Transactions")]
            string inputMsg,
            [EventHub("%Transactions:EVH:Transactions:Hub%", Connection = "Transactions:EVH:Transactions")] IAsyncCollector<EventData> outputEvent,
            ILogger log)
        {
            if (string.IsNullOrWhiteSpace(inputMsg))
            {
                throw new ArgumentNullException(nameof(inputMsg));
            }

            log.LogInformation($"Service bus topic trigger function processing message: {inputMsg}");

            var retailTransaction = JsonConvert.DeserializeObject<RetailTransaction>(
                inputMsg);

            if (retailTransaction == null)
            {
                throw new JsonException("Deserialized transaction was null");
            }

            try
            {
                var transaction = retailTransaction.ToDto();
                log.LogInformation($"Transaction {transaction.TransactionNumber} processed.");
                // Post to event hub
                await outputEvent.AddAsync(new EventData(ToByteArray(transaction)));

                log.LogInformation($"Transaction {transaction.TransactionNumber} processed.");

                // Create ServiceBusMessage instance
                var serviceBusMessage = new ServiceBusMessage(JsonConvert.SerializeObject(transaction));
                serviceBusMessage.MessageId = transaction.TransactionNumber.ToString();
                serviceBusMessage.Subject = "Transaction Updated"; 

                return serviceBusMessage;
            }
            catch (Exception e)
            {
                log.LogError(e, "Error mapping transaction.");
            }
            return null;
        }
        private static byte[] ToByteArray(object obj)
        {
            var json = JsonConvert.SerializeObject(obj);
            return Encoding.UTF8.GetBytes(json);
        }
    }

    public class RetailTransaction
    {
        public string TransactionNumber { get; set; }
        public decimal Amount { get; set; }

        public Transaction ToDto()
        {
            return new Transaction
            {
                TransactionNumber = this.TransactionNumber,
                Amount = this.Amount
            };
        }
    }
    public class Transaction
    {
        public string TransactionNumber { get; set; }
        public decimal Amount { get; set; }
    }
}

.csproj-

<Project Sdk="Microsoft.NET.Sdk">
  <PropertyGroup>
    <TargetFramework>net6.0</TargetFramework>
    <AzureFunctionsVersion>v4</AzureFunctionsVersion>
    <RootNamespace>_78237298</RootNamespace>
  </PropertyGroup>
  <ItemGroup>
    <PackageReference Include="Azure.Messaging.EventHubs" Version="5.11.1" />
    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions.EventHubs" Version="6.2.0" />
    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions.ServiceBus" Version="5.14.0" />
    <PackageReference Include="Microsoft.NET.Sdk.Functions" Version="4.3.0" />
  </ItemGroup>
  <ItemGroup>
    <None Update="host.json">
      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
    </None>
    <None Update="local.settings.json">
      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
      <CopyToPublishDirectory>Never</CopyToPublishDirectory>
    </None>
  </ItemGroup>
</Project>

输出-

活动中心-

第二趟服务巴士-