为什么 NetMQ 不能在 NUnit 环境中工作

Question

我用我的 NetMQ 实现创建了一个 Gist，因为我觉得在这里粘贴有点多：https://gist.github.com/gthvidsten/e626d7e6c51012b1ba152d22e034d93d

如果我在 .Net Core 控制台应用程序中执行以下操作，一切正常，我会收到 MessageReceived 事件：

static void Main(string[] args)
{
    _transportWithHost = new NetMqTransport(
        "tcp://localhost:9990",
        "tcp://localhost:9991",
        true);
    _transportWithHost.Start();

    Console.WriteLine("Press [Enter] to publish");
    Console.ReadLine();

    _transportWithHost.MessageReceived += (sender, e) =>
    {
        ; // Breakpoints here are hit
    };
    _transportWithHost.Publish(new byte[] { 1, 2, 3, 4 });

    Console.WriteLine("Press [Enter] to exit");
    Console.ReadLine();
}

但是，如果我尝试在 NUnit 测试环境中做同样的事情，MessageReceived 事件将永远不会触发！

class NetMqTransportTests
{
    private NetMqTransport _transportWithHost;

    [OneTimeSetUp]
    public void Setup()
    {
        _transportWithHost = new NetMqTransport(
            "tcp://localhost:9990",
            "tcp://localhost:9991",
            true);
        _transportWithHost.Start();
    }

    [Test]
    public void PublishTest()
    {
        ManualResetEvent mre = new ManualResetEvent(false);

        _transportWithHost.MessageReceived += (sender, e) =>
        {
            mre.Set();
            // Breakpoints here are never hit as MessageReceived is never called
        };

        _transportWithHost.Publish(new byte[] { 1, 2, 3, 4 });

        bool eventFired = mre.WaitOne(new TimeSpan(0, 0, 5));
        Assert.True(eventFired);
    }
}

为什么几乎相同的代码在控制台应用程序中有效，但在 NUnit 环境中无效？

Answer 1

我能够重现它并找到此线程 https://github.com/zeromq/netmq/issues/482，这表明它在启动发布者和订阅者接收消息的时间之间存在时间问题。

using NetMQ;
using NetMQ.Sockets;
using NUnit.Framework;
using System;
using System.Threading;
using System.Threading.Tasks;

namespace Tests
{
    class NetMqTransportTests
    {
        [Test]
        public void TestMulticastNetMQ()
        {
            bool wasCalled = false;

            var coreEventbus = "tcp://localhost:12345";

            Task.Run(() =>
            {
                using (var subSocket = new SubscriberSocket())
                {
                    subSocket.Connect(coreEventbus);
                    subSocket.Subscribe("account");

                    while (true)
                    {
                        string messageTopicReceived = subSocket.ReceiveFrameString();
                        string messageReceived = subSocket.ReceiveFrameString();

                        Assert.IsTrue(messageReceived == "testing");
                        wasCalled = true;
                        break;
                    }
                }

            });

            Thread.Sleep(TimeSpan.FromSeconds(1));

            using (var pubSocket = new PublisherSocket())
            {
                pubSocket.Bind(coreEventbus);

                Thread.Sleep(500);
                pubSocket.SendMoreFrame("account").SendFrame("testing");
            }

            Thread.Sleep(TimeSpan.FromSeconds(5));

            Assert.IsTrue(wasCalled);
        }
    }
}

更新：

以下是 NetMQ 库附带的单元测试：https://github.com/zeromq/netmq/blob/master/src/NetMQ.Tests/XPubSubTests.cs

看看他们如何将 NetMqTransport 实例化分解为同时使用 XPublisherSocket 和 XPublisherSocket...

还请注意，根据问题 482，他们会延迟 500 毫秒以让订阅者在收到消息之前连接，就像他们在问题中所说的那样：

[Fact]
public void TopicPubSub()
{
    using (var pub = new XPublisherSocket())
    using (var sub = new XPublisherSocket())
    {
        var port = pub.BindRandomPort("tcp://127.0.0.1");
        sub.Connect("tcp://127.0.0.1:" + port);
        sub.SendFrame(new byte[] { 1, (byte)'A' });

        // let the subscriber connect to the publisher before sending a message
        Thread.Sleep(500);

        var msg = pub.ReceiveFrameBytes();