首页 文章

如何从azure功能向azure服务总线添加消息?

提问于
浏览
0

如果我已经有一个json格式的消息,如何向服务总线添加消息 . 我能够使用azure函数输出绑定添加消息,但在servicebusexplorer或queueexplorer中没有看到任何消息属性 .

我需要重新提交大约1K的消息,消息上有错误,所以我将它们导出到文件,在记事本中修复它,现在我创建了一个读取文件并将其放入队列的azure函数 . 但是当我查看消息时,servicebusexploerer中没有消息属性显示出来 .

run.csx

#r "Newtonsoft.Json"

using System.Net;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Primitives;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using System;
using System.Threading.Tasks;
using System.Configuration;

const string QueueName = "commands";
static string FileName = "messages.json";

public static async Task<string> Run(HttpRequest req, ILogger log,
                 ExecutionContext context, ICollector<string> outputSbQueue)
{
    log.LogInformation("Starting processing messages.");

    var filePath = System.IO.Path.Combine(context.FunctionDirectory, FileName);

    log.LogInformation("Path: " + filePath);

    var text = File.ReadAllText(filePath);

    log.LogInformation("Message: " + text);

    JArray messages = JArray.Parse(text);

    log.LogInformation("Number of message: " + messages.Count);

    await SendMessagesAsync(messages,log,outputSbQueue);
    // return req.CreateResponse(HttpStatusCode.OK,
    //                             "Updated",
    //                             "text/plain");
    return "test";
}

static async Task SendMessagesAsync(JArray messages, ILogger log, 
ICollector<string> outputSbQueue )
{
    log.LogInformation("About to iterate messages");

    foreach (var message in messages)
    {
        log.LogInformation("Sending Message");
        outputSbQueue.Add(message.ToString());
        log.LogInformation("Sent message: " + message);
    }
}

messages.json

[
  {
    "Body": {
      "PaymentPlanId": "2141110b-07da-46b7-a166-ffc7f9f6c5af",
      "InstallmentId": "3bd27b0d-3372-456c-856c-74e09de1413a",
      "Date": "2018-12-05T00:00:00",
      "Amount": 66.89,
      "Attempt": 0,
      "PaymentCorrelationId": "2ae7511e-706f-4d7f-b44b-9690d0fcbf38",
      "CommandId": "a2d5ae26-6289-4cca-bce0-7a1905b64378"
    },
    "ContentType": "text/plain",
    "CorrelationId": null,
    "DeadLetterSource": "commands",
    "DeliveryCount": 1,
    "EnqueuedSequenceNumber": 14684,
    "EnqueuedTimeUtc": "2018-12-06T13:22:37.131Z",
    "ExpiresAtUtc": "9999-12-31T23:59:59.9999999",
    "ForcePersistence": false,
    "IsBodyConsumed": false,
    "Label": "PayDueInstallmentCommand",
    "LockedUntilUtc": null,
    "LockToken": null,
    "MessageId": "a2d5ae26-6289-4cca-bce0-7a1905b64378",
    "PartitionKey": null,
    "Properties": {
      "BodyClrType": "SR.Domain.Commands.PayDueInstallmentCommand, SR.Domain, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null",
      "ParentId": "|Dz4Pxv65XMA=.3975a8a2_32.",
      "RootId": "Dz4Pxv65XMA=",
      "Diagnostic-Id": "|Dz4Pxv65XMA=.3975a8a2_32.1.",
      "DeadLetterReason": "NoCommandInMessage",
      "DeadLetterErrorDescription": "There was no command in the message.",
      "Test":"1"
    },
    "ReplyTo": null,
    "ReplyToSessionId": null,
    "ScheduledEnqueueTimeUtc": "2018-12-06T13:22:36.877Z",
    "SequenceNumber": 14684,
    "SessionId": null,
    "Size": 938,
    "State": 0,
    "TimeToLive": "10675199.02:48:05.4775807",
    "To": null,
    "ViaPartitionKey": null
  }
 ]

function.json

{
  "bindings": [
    {
      "authLevel": "function",
      "name": "req",
      "type": "httpTrigger",
      "direction": "in",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "outputSbQueue",
      "type": "serviceBus",
      "queueName": "deadletter",
      "connection": "ServiceBusConnectionString",
      "direction": "out"
    }
  ],
  "disabled": false
}

1 回答

  • 0

    装配问题很容易修复,就像@Roman提到的那样 . 由于您已创建v2函数(如果尚未修改新的Function应用程序的运行时版本,则为默认值),请在下面添加命令 .

    #r "..\\bin\\Microsoft.Azure.ServiceBus.dll" 
    using Microsoft.Azure.ServiceBus;
    

    另一个问题是JSON模型的结构 . 它实际上基于 Microsoft.ServiceBus.Messaging 中的BrokeredMessage而不是 Microsoft.Azure.ServiceBus 中的Message . 您可能必须决定使用哪一个,并在必要时重构Json . 请注意,某些属性是由Azure Service Bus Service设置的,我们无法在新创建的消息中修改这些属性 .

    举个例子 Message . 根据 Message Class重构JSON,包括可配置的所有属性 .

    [
      {
        "Body": {
          "PaymentPlanId": "2141110b-07da-46b7-a166-ffc7f9f6c5af",
          ...
        },
        "ContentType": "text/plain",
        "Label": "MyLable",
        "MessageId": "a2d5ae26-6289-4cca-bce0-7a1905b64378",
        "ScheduledEnqueueTimeUtc": "2018-12-06T13:22:36.877Z",
        "TimeToLive": "10675199.02:48:05.4775807",
        "CorrelationId": null,
        "PartitionKey": null,
        "ReplyTo": null,
        "ReplyToSessionId": null,
        "SessionId": null,
        "To": null,
        "ViaPartitionKey": null
        "UserProperties": {
            "CustomProperty":"test",
            ...
        }
      }
     ]
    

    我们不能像 JsonConvert.DeserializeObject 那样直接使用反序列化,因为消息体需要 byte[] .

    foreach (var message in messages)
            {
                // Get Body first
                var body = System.Text.Encoding.UTF8.GetBytes(message["Body"].ToString());
                // Empty Body section to deserialize properties
                message["Body"] = "";
                var SBMessage = JsonConvert.DeserializeObject<Message>(message.ToString());
                SBMessage.Body = body;
                log.LogInformation("Sending Message");
                outputSbQueue.Add(SBMessage);
                log.LogInformation("Sent message: " + SBMessage.MessageId);
            }
    

相关问题