使用 MassTransit SQS 时如何配置主题名称？

Question

我使用 MassTransit SQS。我发现，当publishing a message时，出现了这个问题：

User is not authorized to perform: SNS:CreateTopic on resource

所以我需要创建主题的权限。

我通过请管理员扩展权限解决了这个问题。可以创建此作品和主题。但是，主题名称似乎是根据 class 名称自动生成的。我能否以编程方式修改此行为（因此通过提供自我描述的主题名称）。

这对于创建 FIFO 主题（需要以 .fifo 结尾的名称）很有用。

更新

我现在熟悉 formatting（感谢克里斯告诉我）和 SetEntityName。

但我的问题似乎是 fifo 特有的。

这个有效：

                cfg.Message<CustomerUpdate>(x =>
                {
                    x.SetEntityName("customerupdate");
                });

但这不起作用：

                cfg.Message<CustomerUpdate>(x =>
                {
                    x.SetEntityName("customerupdate.fifo");
                });

它导致异常并显示错误消息：

Invalid parameter: Topic Name

此类型：

Amazon.SimpleNotificationService.Model.InvalidParameterException

使用此堆栈跟踪：

Amazon.Runtime.Internal.HttpErrorResponseExceptionHandler.HandleExceptionStream(IRequestContext requestContext, IWebResponseData httpErrorResponse, HttpErrorResponseException exception, Stream responseStream)
Amazon.Runtime.Internal.HttpErrorResponseExceptionHandler.<HandleExceptionAsync>d__2.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
Amazon.Runtime.Internal.ExceptionHandler`1.<HandleAsync>d__6.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
Amazon.Runtime.Internal.ErrorHandler.<ProcessExceptionAsync>d__8.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
Amazon.Runtime.Internal.ErrorHandler.<InvokeAsync>d__5`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
Amazon.Runtime.Internal.CallbackHandler.<InvokeAsync>d__9`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
Amazon.Runtime.Internal.EndpointDiscoveryHandler.<InvokeAsync>d__2`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
Amazon.Runtime.Internal.EndpointDiscoveryHandler.<InvokeAsync>d__2`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
Amazon.Runtime.Internal.CredentialsRetriever.<InvokeAsync>d__7`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
Amazon.Runtime.Internal.RetryHandler.<InvokeAsync>d__10`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
Amazon.Runtime.Internal.RetryHandler.<InvokeAsync>d__10`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
Amazon.Runtime.Internal.CallbackHandler.<InvokeAsync>d__9`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
Amazon.Runtime.Internal.CallbackHandler.<InvokeAsync>d__9`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
Amazon.Runtime.Internal.ErrorCallbackHandler.<InvokeAsync>d__5`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
Amazon.Runtime.Internal.MetricsHandler.<InvokeAsync>d__1`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
MassTransit.AmazonSqsTransport.Contexts.TopicCache.<CreateMissingTopic>d__9.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
MassTransit.AmazonSqsTransport.Contexts.TopicCache.<>c__DisplayClass6_0.<<Get>b__0>d.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
GreenPipes.Caching.Internals.PendingValue`2.<CreateValue>d__6.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
MassTransit.AmazonSqsTransport.Pipeline.ConfigureTopologyFilter`1.<Declare>d__7.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult()
MassTransit.AmazonSqsTransport.Pipeline.ConfigureTopologyFilter`1.<ConfigureTopology>d__5.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable.ConfiguredTaskAwaiter.GetResult()
MassTransit.AmazonSqsTransport.Pipeline.ConfigureTopologyFilter`1.<>c__DisplayClass3_0.<<GreenPipes-IFilter<MassTransit-AmazonSqsTransport-ClientContext>-Send>b__0>d.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable.ConfiguredTaskAwaiter.GetResult()
GreenPipes.PipeExtensions.<OneTimeSetup>d__2`1.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable.ConfiguredTaskAwaiter.GetResult()
MassTransit.AmazonSqsTransport.Pipeline.ConfigureTopologyFilter`1.<GreenPipes-IFilter<MassTransit-AmazonSqsTransport-ClientContext>-Send>d__3.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable.ConfiguredTaskAwaiter.GetResult()
MassTransit.AmazonSqsTransport.Transport.TopicSendTransport.SendPipe`1.<Send>d__5.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.ConfiguredTaskAwaitable.ConfiguredTaskAwaiter.GetResult()
GreenPipes.Agents.PipeContextSupervisor`1.<GreenPipes-IPipeContextSource<TContext>-Send>d__7.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
GreenPipes.Agents.PipeContextSupervisor`1.<GreenPipes-IPipeContextSource<TContext>-Send>d__7.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
GreenPipes.Agents.PipeContextSupervisor`1.<GreenPipes-IPipeContextSource<TContext>-Send>d__7.MoveNext()
System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
System.Runtime.CompilerServices.TaskAwaiter.GetResult()<PublishAsync>d__2.MoveNext() in Publisher.cs: line: 18

这是我的代码

    public static void UseMassTransit(this IServiceCollection services, MassTransitConfiguration massTransitConfiguration)
    {
        services.AddMassTransit(x =>
        {
            x.AddConsumer<CustomerChangeConsumer>();
            x.UsingAmazonSqs((context, cfg) =>
            {
                cfg.Host(massTransitConfiguration.Host, h =>
                {
                    h.AccessKey(massTransitConfiguration.AccessKey);
                    h.SecretKey(massTransitConfiguration.SecretKey);

                    h.EnableScopedTopics();
                });

                cfg.ReceiveEndpoint("CustomerChangeConsumer",
                    configurator =>
                    {
                        configurator.ConfigureConsumer<CustomerChangeConsumer>(context);
                    });

                cfg.Message<CustomerUpdate>(x =>
                {
                    // This causes the exception when publishing:
                    x.SetEntityName("customerupdate.fifo");
                    // But this works. No issues when publishing
                    //x.SetEntityName("customerupdate");
                });
            });
        });
        
        services.AddMassTransitHostedService();
    }
}

Answer 1

您可以configure the message topology指定实体名称格式。

例如，要更改 OrderSubmitted 事件的主题名称：

Bus.Factory.CreateUsingAmazonSqs(cfg =>
{
    cfg.Message<OrderSubmitted>(x =>
    {
        x.SetEntityName("order-submitted.fifo");
    });

    cfg.Publish<OrderSubmitted>(x =>
    {
        x.TopicAttributes["FifoTopic"] = "true";
    });
});

要在发布时设置 MessageGroupId，请使用：

await bus.Publish<OrderSubmitted>(message, x => x.SetGroupId("..."));

I'm not 100% sure that survives Publish, vs. Send.

使用 MassTransit SQS 时如何配置主题名称？

How can I configure the topic name when using MassTransit SQS?

masstransit

amazon-sqs