Why doesn't NetMQ work in an NUnit environment

Question

I've created a Gist with my NetMQ implementation as I feel it a bit much to paste here: https://gist.github.com/gthvidsten/e626d7e6c51012b1ba152d22e034d93d

If I do the following in a .Net Core console app, everything works fine and I receive the MessageReceived event:

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();
}

However, if I try to do the same in an NUnit test environment, the MessageReceived event is never fired!

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);
    }
}

Why does the virtually identical code work in a console app, but not in an NUnit environment?

Answer 1

I was able to reproduce it and found this thread https://github.com/zeromq/netmq/issues/482 which indicates its a timing issue between starting the Publisher and time for the Subscriber to receive the message.

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);
        }
    }
}

Update:

Here are the Unit Tests that come with the NetMQ library: https://github.com/zeromq/netmq/blob/master/src/NetMQ.Tests/XPubSubTests.cs

See how they break up instantiating NetMqTransport into using both XPublisherSocket and XPublisherSocket...

Also notice as per the issue 482 they do a 500ms delay to let the subscriber connect before receiving the message, just like they were talking about in the issue:

[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();