如果我已经有一个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 回答
装配问题很容易修复,就像@Roman提到的那样 . 由于您已创建v2函数(如果尚未修改新的Function应用程序的运行时版本,则为默认值),请在下面添加命令 .
另一个问题是JSON模型的结构 . 它实际上基于
Microsoft.ServiceBus.Messaging
中的BrokeredMessage而不是Microsoft.Azure.ServiceBus
中的Message . 您可能必须决定使用哪一个,并在必要时重构Json . 请注意,某些属性是由Azure Service Bus Service设置的,我们无法在新创建的消息中修改这些属性 .举个例子
Message
. 根据Message
Class重构JSON,包括可配置的所有属性 .我们不能像
JsonConvert.DeserializeObject
那样直接使用反序列化,因为消息体需要byte[]
.