Azure ServiceBus BrokeredMessage 正文为 null

Question

我有一个计时器触发函数，它将对象发送到服务总线主题，如下所示

        [FunctionName("ProcessTermedEmployee")]
        public static async Task Run(
                        [TimerTrigger("%TimerInterval%"
            #if DEBUG
            , RunOnStartup=true
             #endif
            )] TimerInfo myTimer,
            [ServiceBus("%TermedEmployeeTopicName%", Connection = "ServiceBusConnectionString")] IAsyncCollector<string> termedEmployeeCollector,
            ILogger log)
        {
            log.LogInformation("ProcessTermedEmployee Timer function invoked.");
            try
            {
               // var termedEmployees = new List<TermedEmployee>();
                using (SqlConnection conn = new SqlConnection(Environment.GetEnvironmentVariable("HRSQLConNString")))
                {
                    conn.Open();
                    SqlCommand cmd = new SqlCommand("hp_Cdi_Get_Termed_Employees_Recent", conn);
                    cmd.CommandType = CommandType.StoredProcedure;
                    cmd.Parameters.Add(new SqlParameter("@TermedAfterDate", DateTime.Now.AddDays(-10)));
                    using (SqlDataReader rdr = cmd.ExecuteReader())
                    {
                        try
                        {
                            while (rdr.Read())
                            {
                                TermedEmployee termedEmployee = ConvertTermedEmployee(rdr);
                              //  termedEmployees.Add(termedEmployee);
                                await 
               
               termedEmployeeCollector.AddAsync(JsonConvert.SerializeObject(termedEmployee));
                            }
                        }
                        catch (Exception ex)
                        {
                            ex.Data.Add("row", rdr);
                            log.LogCritical(ex, ex.Message);
                            throw ex;
                        }
                    }
                }
                log.LogInformation("ProcessTermedEmployee Timer function finished.");
            }
            catch (Exception ex)
            {
                log.LogCritical(ex, ex.Message);
                throw ex;
            }
        }

然后我有一个服务总线触发函数，该函数应该接收消息并将其添加到 CosmosDB

    public static class LogTermedEmployee
    {
        [FunctionName("LogTermedEmployee")]
        public static async Task Run([ServiceBusTrigger("%TermedEmployeeTopicName%", "%ServiceBusSubscriptionName%", Connection = "ServiceBusConnectionString")] BrokeredMessage message,
                [CosmosDB(
                databaseName: "%CosmosDbName%",
                collectionName: "%TermedEmployeesLogCollection%",
                ConnectionStringSetting = "CosmosConnection")]
                IAsyncCollector<TermedEmployeeLog> termedEmployeesLogCollection,
          ILogger log)
          {
            log.LogInformation($"C# ServiceBus topic trigger function LogTermedEmployee, 
     processed at {DateTime.Now}");
            try
            {
          
                if(message.GetBody<TermedEmployee>() != null)
                {
                    //StreamReader reader = new StreamReader(message.Body);
                  // string s = System.Text.Encoding.Default.GetString(message.Body);
                          var termedEmp =  JsonConvert.DeserializeObject<TermedEmployee> 
              (message.GetBody<string>());
                    await termedEmployeesLogCollection.AddAsync(new TermedEmployeeLog() { 
                   TermedEmployee = termedEmp, DateOfProcess = DateTime.Now, Id = 
                   Guid.NewGuid(), PartitionKey = "TermedEmployee", ModifiedOn = 
                      DateTime.Now, ModifiedBy = "TermedEployeeLogService", MessageId = "" 
               });// message.MessageId });

                }
            }
            catch (Exception ex)
            {

                log.LogCritical(ex, ex.Message);
                throw ex;
            }
        }

    <PackageReference Include="Azure.Messaging.ServiceBus" Version="7.6.0" />
<PackageReference Include="AzureFunctions.Extensions.DependencyInjection" Version="1.1.3" />
<PackageReference Include="CDI.Utilities.LogHelpers" Version="0.1.3" />
<PackageReference Include="Microsoft.AspNetCore.Mvc.Abstractions" Version="2.2.0" />
<PackageReference Include="Microsoft.Azure.Functions.Extensions" Version="1.1.0" />
<PackageReference Include="Microsoft.Azure.ServiceBus" Version="5.2.0" />
<PackageReference Include="Microsoft.Azure.WebJobs.Extensions" Version="4.0.1" />
<PackageReference Include="Microsoft.Azure.WebJobs.Extensions.CosmosDB" Version="3.0.10" />
<PackageReference Include="Microsoft.Azure.WebJobs.Extensions.Http" Version="3.0.12" />
<PackageReference Include="Microsoft.Azure.WebJobs.Extensions.ServiceBus" Version="5.2.0" />
<PackageReference Include="Microsoft.Azure.WebJobs.Script.ExtensionsMetadataGenerator" Version="4.0.1" />
<PackageReference Include="Microsoft.NET.Sdk.Functions" Version="4.0.1" />
<PackageReference Include="System.Data.SqlClient" Version="4.8.3" />
<PackageReference Include="WindowsAzure.ServiceBus" Version="6.2.2" />

问题

在 ServiceBus 触发器的签名中

如果我使用 BrokeredMessage 消息 message.GetBody() 为 null

如果我使用 Obsoleted Message 对象，则 Message.Body 为 null

但是如果我使用简单的字符串，我会得到正确的 Json。

请问有什么建议吗？

Answer 1

我建议在触发器中将 BrokeredMessage 更改为 ServiceBusReceivedMessage:

public static async Task Run(
    [ServiceBusTrigger(
        "%TermedEmployeeTopicName%", 
        "%ServiceBusSubscriptionName%", 
        Connection = "ServiceBusConnectionString")] 
    ServiceBusReceivedMessage message,
    [CosmosDB(
        databaseName: "%CosmosDbName%",
        collectionName: "%TermedEmployeesLogCollection%",
        ConnectionStringSetting = "CosmosConnection")]
        IAsyncCollector<TermedEmployeeLog> termedEmployeesLogCollection,
        ILogger log)

我相信这将解决绑定(bind)问题。

其他上下文

从 v5.0.0 开始，Microsoft.Azure.WebJobs.Extensions.ServiceBus 包开始在内部使用 Azure.Messaging.ServiceBus。新包中没有 BrokeredMessage 类型。传入消息的类型为 ServiceBusReceivedMessage，传出消息的类型为 ServiceBusMessage。

更多信息可以在Microsoft.Azure.WebJobs.Extensions.ServiceBus docs中找到.