如何用NServiceBus的SqsTransport订阅已有AWS Textract事件SQS队列？

一、配置NServiceBus使用现有SQS队列 #

要让NServiceBus复用已存在的SQS队列且不创建新队列，需在端点配置中做以下设置：

1. 禁用自动队列创建
   阻止NServiceBus尝试创建输入队列、错误队列或超时队列：

var endpointConfig = new EndpointConfiguration("YourEndpointIdentifier");
var sqsTransport = endpointConfig.UseTransport<SqsTransport>();

// 禁用所有队列的自动创建逻辑
sqsTransport.DisableCreateQueue();
sqsTransport.DisableCreateErrorQueue();
sqsTransport.DisableCreateTimeoutQueues();

2. 指定现有队列作为输入队列
   将输入队列名称设置为你已有的SQS队列名称：

sqsTransport.InputQueueName("YourExistingQueueName");

3. 配置AWS区域与凭证
   确保NServiceBus能访问目标队列所在的AWS区域，同时配置正确的IAM凭证（优先使用IAM角色或环境变量，必要时可显式指定）：

sqsTransport.Region("us-east-1"); // 替换为你的队列所在区域
// 若需显式指定凭证（可选）
sqsTransport.Credentials(new Amazon.Runtime.BasicAWSCredentials("YourAccessKey", "YourSecretKey"));

4. 设置为仅接收端点（可选）
   若该端点仅用于接收消息无需发送，可简化配置避免不必要的发送相关初始化：

// 禁用即时重试等非必要发送相关功能
endpointConfig.Recovery(c => c.DisableImmediateRetries());

二、处理原生AWS事件 #

SNS转发到SQS的消息为AWS原生格式（外层是SNS通知结构，内部嵌套Textract事件JSON），NServiceBus默认无法直接解析，需自定义处理逻辑：

1. 定义事件模型
   创建对应Textract完成事件及SNS外层通知的实体类：

// Textract完成事件结构
public class TextractJobCompletedEvent
{
    public string JobId { get; set; }
    public string Status { get; set; }
    public string DocumentLocation { get; set; }
    // 按需添加其他Textract事件字段
}

// SNS通知外层结构
public class SnsNotification
{
    public string Message { get; set; }
    public string MessageId { get; set; }
    public string TopicArn { get; set; }
}

2. 使用Raw消息处理器解析原生消息
   通过AddRawMessageHandler获取原生SQS消息，解析后转换为NServiceBus可处理的事件：

endpointConfig.AddRawMessageHandler(async context =>
{
    // 获取原生SQS消息对象
    var sqsMessage = context.Extensions.Get<Amazon.SQS.Model.Message>();
    // 解析外层SNS通知
    var snsNotification = JsonConvert.DeserializeObject<SnsNotification>(sqsMessage.Body);
    // 解析内部Textract事件
    var textractEvent = JsonConvert.DeserializeObject<TextractJobCompletedEvent>(snsNotification.Message);

    // 将解析后的事件发送到本地处理程序
    await context.SendLocal(textractEvent);
    // 标记消息已处理
    return RawMessageHandlerResult.Handled;
});

3. 添加事件处理程序
   创建常规的NServiceBus处理程序来处理解析后的Textract事件：

public class TextractJobCompletedHandler : IHandleMessages<TextractJobCompletedEvent>
{
    public async Task Handle(TextractJobCompletedEvent message, IMessageHandlerContext context)
    {
        // 在这里实现Textract完成事件的业务逻辑
        Console.WriteLine($"Textract任务 {message.JobId} 完成，状态：{message.Status}");
        await Task.CompletedTask;
    }
}

关键注意事项 #

• IAM权限：确保NServiceBus使用的IAM身份拥有目标SQS队列的ReceiveMessage、DeleteMessage和ChangeMessageVisibility权限。
• 消息封装：若需保留原生消息属性，可配置sqsTransport.DoNotWrapMessages()，但此设置会让NServiceBus跳过自身的消息封装逻辑，需自行处理消息的可靠性相关逻辑。

内容的提问来源于stack exchange，提问作者Sebastian Weber