无法使用 .net 客户端 sdk 使用来自 Google PubSub 模拟器的邮件

Question

在启动Google PubSub模拟器并向其发送一些消息后，我启动了System Under Test，当我试图启动SubscriberClient时，我得到一个异常:

Grpc.Core.RpcException
HResult=0x80131500
Message=Status(StatusCode="NotFound", Detail="Subscription does not exist (resource=franchisingSalesReceipts-motorpromo-sub-dl)", DebugException="Grpc.Core.Internal.CoreErrorDetailException: {"created":"@1635433978.567000000","description":"Error received from peer ipv4:127.0.0.1:8085","file":"......srccorelibsurfacecall.cc","file_line":1067,"grpc_message":"Subscription does not exist (resource=franchisingSalesReceipts-motorpromo-sub-dl)","grpc_status":5}")
Source=Google.Cloud.PubSub.V1
StackTrace:
at Google.Cloud.PubSub.V1.SubscriberClientImpl.SingleChannel.HandleRpcFailure(Exception e)
at Google.Cloud.PubSub.V1.SubscriberClientImpl.SingleChannel.HandlePullMessageData(Task`1 moveNextTask)
at Google.Cloud.PubSub.V1.SubscriberClientImpl.SingleChannel.<>c__DisplayClass46_0.<HandlePullMoveNext>b__1()
at Google.Cloud.PubSub.V1.SubscriberClientImpl.SingleChannel.<StartAsync>d__38.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Runtime.CompilerServices.ConfiguredTaskAwaitable.ConfiguredTaskAwaiter.GetResult()
at Google.Cloud.PubSub.V1.Tasks.ForwardingAwaiter.GetResult()
at Google.Cloud.PubSub.V1.Tasks.Extensions.<>c__DisplayClass4_0.<<ConfigureAwaitHideErrors>g__Inner|0>d.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Runtime.CompilerServices.ConfiguredTaskAwaitable.ConfiguredTaskAwaiter.GetResult()
at PromotionEngine.OneToOne.Infrastructure.Services.GooglePubSubConsumerService.<RegisterEventHandler>d__6`2.MoveNext() in C:UsersjmouraoworkspaceMotorPromocaoPromotionEngine.OneToOnesrcPromotionEngine.OneToOne.InfrastructureServicesGooglePubSubConsumerService.cs:line 57
This exception was originally thrown at this call stack:
[External Code]
PromotionEngine.OneToOne.Infrastructure.Services.GooglePubSubConsumerService.RegisterEventHandler<TWorker, TMessage>(string) in GooglePubSubConsumerService.cs

重现步骤

使用以下命令启动模拟器后:

gcloud beta emulators pubsub start --project=integracao-apigee-dev

我用这样的东西连接出版商:

public GcpEmulatorPublisherService(
ILogger<GcpEmulatorPublisherService> logger,
IConfiguration config)
{
_logger = logger;
string endpoint = config.GetValue<string>("GoogleCloudPubSub:host");
string projectId = config.GetValue<string>("GoogleCloudPubSub:projectId");
string[] topics = config.GetSection("GoogleCloudPubSub:topics").Get<string[]>();
string currentTopicId = config.GetValue<string>("GoogleCloudPubSub:currentTopicId");
PublisherServiceApiClient publisherService = new PublisherServiceApiClientBuilder
{
Endpoint = endpoint,
ChannelCredentials = ChannelCredentials.Insecure
}.Build();
foreach (string topicId in topics)
{
TopicName topicName = new TopicName(projectId, topicId);
try
{
publisherService.CreateTopic(topicName);
}
catch (RpcException ex) when (ex.StatusCode == StatusCode.AlreadyExists)
{
_logger.LogInformation($"Topic already exists: ProjectId {projectId} TopicId {topicId}");
}
}
var settings = new PublisherClient.ClientCreationSettings(
credentials: ChannelCredentials.Insecure,
serviceEndpoint: endpoint
);
_publisherClient = PublisherClient.Create(new TopicName(projectId, currentTopicId), settings);
}

之后，我在test下启动我的系统使用不安全的凭据和本地主机ServiceEndpoint:

SubscriberClient.ClientCreationSettings clientCreationSettings = new(clientCount: 1, credentials: ChannelCredentials.Insecure, serviceEndpoint: "localhost:8085");
var subscriberClient = await SubscriberClient.CreateAsync(subscription, clientCreationSettings);

在此之前一切正常。

当我尝试像这样开始订阅时:

await subscriberClient.StartAsync(async (PubsubMessage message, CancellationToken cancellationToken) =>
{
try
{
TMessage messageBody = ConvertMessageBody<TMessage>(message);
TWorker worker = DependencyResolver.GetRequiredService<TWorker>();
await worker.ProcessAsync(messageBody);
return await Task.FromResult(SubscriberClient.Reply.Ack);
}
catch (Exception exception)
{
string exceptionMessage = string.Format(
"Exception reading message from subscription from supplier {0}. See inner exception for details. Message={1}",
topicName, exception.Message);
Exception trackedException = new(exceptionMessage, exception);
_telemetryWrapper.TrackException(trackedException);
_logger.LogError(trackedException, "Error processing message");
return await Task.FromResult(SubscriberClient.Reply.Nack);
}
}).ConfigureAwait(false);

我得到了上述异常。

Answer 1

我找到了导致问题的原因。

我没有创建我试图在测试中的系统(消费者)中使用的订阅。

我不知道您必须创建Topic和Subscription，以便消费者应用程序可以使用此订阅来接收消息。

因此，在我的Publisher测试应用程序中，我正在创建所需的订阅，就像在文档中一样:

// Subscribe to the topic.
SubscriberServiceApiClient subscriberService = await SubscriberServiceApiClient.CreateAsync();
SubscriptionName subscriptionName = new SubscriptionName(projectId, subscriptionId);
subscriberService.CreateSubscription(subscriptionName, topicName, pushConfig: null, ackDeadlineSeconds: 60);

我还发现，如果您尝试使用subscriptionId等于topicId来创建订阅，它将不起作用。