c# - 在 Azure 服务总线中发布到一个主题并订阅另一个主题的集成测试不可靠是否存在竞争条件？

Question

我正在尝试编写一个集成/验收测试来测试azure中的一些代码，问题ATM中的代码只是订阅一个主题并发布到另一个主题。

我已经编写了测试，但它并不总是通过，似乎可能存在竞争条件。我尝试过多种方式编写它，包括使用 OnMessage 和使用 Receive(我在此处显示的示例)。

当使用 OnMessage 时，测试似乎总是过早退出(大约 30 秒)，我想这可能意味着它无论如何都不适合此测试。

我对我的示例的具体查询，我假设一旦我创建了对目标主题的订阅，我就可以使用 Receive() 接收发送给它的任何消息，无论该消息到达的时间点是什么意思，如果消息在我调用 Receive() 之前到达目标主题，之后我仍然可以通过调用 Receive() 读取消息。有人可以解释一下吗？

    namespace somenamespace {
    [TestClass]
    public class SampleTopicTest
    {
        private static TopicClient topicClient;
        private static SubscriptionClient subClientKoEligible;
        private static SubscriptionClient subClientKoIneligible;

        private static OnMessageOptions options;
        public const string TEST_MESSAGE_SUB = "TestMessageSub";
        private static NamespaceManager namespaceManager;

        private static string topicFleKoEligible;
        private static string topicFleKoIneligible;

        private BrokeredMessage message;

        [ClassInitialize]
        public static void BeforeClass(TestContext testContext)
        {
            //client for publishing messages
            string connectionString = ConfigurationManager.AppSettings["ServiceBusConnectionString"];
            string topicDataReady = ConfigurationManager.AppSettings["DataReadyTopicName"];
            topicClient = TopicClient.CreateFromConnectionString(connectionString, topicDataReady);

            topicFleKoEligible = ConfigurationManager.AppSettings["KnockOutEligibleTopicName"];
            topicFleKoIneligible = ConfigurationManager.AppSettings["KnockOutIneligibleTopicName"];

            //create test subscription to receive messages
            namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);


            if (!namespaceManager.SubscriptionExists(topicFleKoEligible, TEST_MESSAGE_SUB))
            {
                namespaceManager.CreateSubscription(topicFleKoEligible, TEST_MESSAGE_SUB);
            }

            if (!namespaceManager.SubscriptionExists(topicFleKoIneligible, TEST_MESSAGE_SUB))
            {
                namespaceManager.CreateSubscription(topicFleKoIneligible, TEST_MESSAGE_SUB);
            }

            //subscriber client koeligible
            subClientKoEligible = SubscriptionClient.CreateFromConnectionString(connectionString, topicFleKoEligible, TEST_MESSAGE_SUB);

            subClientKoIneligible = SubscriptionClient.CreateFromConnectionString(connectionString, topicFleKoIneligible, TEST_MESSAGE_SUB);

            options = new OnMessageOptions()
            {
                AutoComplete = false,
                AutoRenewTimeout = TimeSpan.FromMinutes(1),

            };
        }

          [TestMethod]
        public void E2EPOCTopicTestLT50()
        {
            Random rnd = new Random();
            string customerId = rnd.Next(1, 49).ToString();

            FurtherLendingCustomer sentCustomer = new FurtherLendingCustomer { CustomerId = customerId };
            BrokeredMessage sentMessage = new BrokeredMessage(sentCustomer.ToJson());           
            sentMessage.CorrelationId = Guid.NewGuid().ToString();
            string messageId = sentMessage.MessageId;
            topicClient.Send(sentMessage);

            Boolean messageRead = false;

            //wait for message to arrive on the ko eligible queue
            while((message = subClientKoEligible.Receive(TimeSpan.FromMinutes(2))) != null){

                //read message
                string messageString = message.GetBody<String>();

                //Serialize
                FurtherLendingCustomer receivedCustomer =  JsonConvert.DeserializeObject<FurtherLendingCustomer>(messageString.Substring(messageString.IndexOf("{")));

                //assertion
                Assert.AreEqual(sentCustomer.CustomerId, receivedCustomer.CustomerId,"verify customer id");

                //pop message
                message.Complete();
                messageRead = true;

                //leave loop after processing one message
                break;
            }
            if (!messageRead)
                Assert.Fail("Didn't receive any message after 2 mins");

        }
    }
}

Answer 1

正如官方文档所述SubscriptionClient.Receive(TimeSpan) :

Parameters serverWaitTime TimeSpan

The time span the server waits for receiving a message before it times out.

A Null can be return by this API if operation exceeded the timeout specified, or the operations succeeded but there are no more messages to be received.

根据我的测试，如果消息发送到主题，然后在特定的 serverWaitTime 内传递到您的订阅，那么无论消息在调用 Receive 之前还是之后到达目标主题，您都可以收到消息.

When using OnMessage the test seemed to always exit prematurely (around 30 seconds), which I guess perhaps means its inappropriate for this test anyway.

[TestMethod]
public void ReceiveMessages()
{
    subClient.OnMessage(msg => {
        System.Diagnostics.Trace.TraceInformation($"{DateTime.Now}：{msg.GetBody<string>()}");
        msg.Complete();
    });
    Task.Delay(TimeSpan.FromMinutes(5)).Wait();
}

对于Subscription​Client.​On​Message ，我假设它基本上是一个调用 Receive 的循环。调用OnMessage后，需要等待一段时间，停止该方法才能退出。这里有一篇关于windows Azure Service Bus的事件驱动消息编程的博客，可以引用here .

此外，我发现您用于发送消息的 topicClient 和用于接收消息的 subClientKoEligible 并不是针对同一主题路径。