NetMQ: wait for a subscriber before the first publish (slow joiner)

.github/workflows/dotnet.yml

@@ -84,6 +84,7 @@ jobs:
         dotnet test --project Obvs.RabbitMQ.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
         dotnet test --project Obvs.NATS.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
         dotnet test --project Obvs.Kafka.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
+        dotnet test --project Obvs.NetMQ.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
     - name: Test
       # xUnit v3 runs on Microsoft.Testing.Platform (global.json). Exit code 8 means a project ran no tests,
       # as the transport test projects only contain Explicit tests.

Obvs.NetMQ.Tests/TestSlowJoiner.cs

@@ -0,0 +1,58 @@
+using System;
+using System.Collections.Concurrent;
+using System.Linq;
+using System.Threading.Tasks;
+using Obvs.Serialization;
+using Obvs.Serialization.Json;
+using Obvs.Types;
+using Xunit;
+
+namespace Obvs.NetMQ.Tests
+{
+    // ZeroMQ drops messages published before a subscriber has connected (the "slow joiner"), so the
+    // publisher's first message must wait for a subscriber. Ports are clear of the other NetMQ tests.
+    public class TestSlowJoiner
+    {
+        private const string Topic = "SlowJoinerTopic";
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task FirstMessageReachesASubscriberThatConnectsLate()
+        {
+            const string address = "tcp://localhost:5570";
+            var received = new ConcurrentQueue<IMessage>();
+            var publisher = new MessagePublisher<IMessage>(address, new JsonMessageSerializer(), Topic);
+            var source = new MessageSource<IMessage>(address, new IMessageDeserializer<IMessage>[] { new JsonMessageDeserializer<SlowJoinerMessage>() }, Topic);
+
+            // The subscriber joins well after the publisher starts, as on a slow machine
+            var subscribeLate = Task.Run(async () =>
+            {
+                await Task.Delay(TimeSpan.FromSeconds(1.5), TestContext.Current.CancellationToken);
+                return source.Messages.Subscribe(received.Enqueue);
+            }, TestContext.Current.CancellationToken);
+
+            await publisher.PublishAsync(new SlowJoinerMessage { Id = 1 });
+
+            using (await subscribeLate)
+            {
+                await WaitFor(() => !received.IsEmpty, TimeSpan.FromSeconds(5));
+            }
+            publisher.Dispose();
+
+            Assert.Contains(received, message => ((SlowJoinerMessage)message).Id == 1);
+        }
+
+        private static async Task WaitFor(Func<bool> condition, TimeSpan timeout)
+        {
+            var deadline = DateTime.UtcNow + timeout;
+            while (!condition() && DateTime.UtcNow < deadline)
+            {
+                await Task.Delay(50, TestContext.Current.CancellationToken);
+            }
+        }
+
+        public class SlowJoinerMessage : IMessage
+        {
+            public int Id { get; set; }
+        }
+    }
+}

Obvs.NetMQ/Extensions/SocketExtensions.cs

@@ -28,7 +28,7 @@ public static bool TryReceive(this SubscriberSocket socket, TimeSpan timeOut,
             return false;
         }
 
-        public static void SendToTopic(this PublisherSocket socket, string topic, string typeName, byte[] serializedMessage)
+        public static void SendToTopic(this NetMQSocket socket, string topic, string typeName, byte[] serializedMessage)
         {
             socket.TrySendMultipartMessage(
                 new NetMQMessage(new[]

Obvs.NetMQ/MessagePublisher.cs

@@ -1,6 +1,8 @@
 using System;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using NetMQ;
 using NetMQ.Sockets;
 using Obvs.NetMQ.Configuration;
 using Obvs.NetMQ.Extensions;
@@ -13,28 +15,68 @@ public class MessagePublisher<TMessage> : IMessagePublisher<TMessage> where TMes
         private readonly string _address;
         private readonly IMessageSerializer _serializer;
         private readonly string _topic;
-        private readonly Lazy<PublisherSocket> _socket;
+        private readonly Lazy<XPublisherSocket> _socket;
         private readonly SocketType _socketType;
 
+        // The first publish gives subscribers at least MinimumSubscriberWait to join, and if none has
+        // subscribed to the topic by then, keeps waiting for one up to SubscriberWaitTimeout
+        private static readonly TimeSpan MinimumSubscriberWait = TimeSpan.FromSeconds(1);
+        private static readonly TimeSpan SubscriberWaitTimeout = TimeSpan.FromSeconds(5);
+
         public MessagePublisher(string address, IMessageSerializer serializer, string topic, SocketType socketType = SocketType.Server)
         {
             _address = address;
             _serializer = serializer;
             _topic = topic;
             _socketType = socketType;
 
-            _socket = new Lazy<PublisherSocket>(() =>
+            _socket = new Lazy<XPublisherSocket>(() =>
             {
-	            var socket = new PublisherSocket();
+                var socket = new XPublisherSocket();
                 socket.Start(_address, _socketType);
-                Thread.Sleep(TimeSpan.FromSeconds(1)); // wait for subscribers
+                WaitForSubscriber(socket);
                 return socket;
             });
         }
 
+        // ZeroMQ drops messages published before a subscriber has connected (the "slow joiner"). An XPUB
+        // socket receives subscriptions, so we can tell when one for our topic has arrived.
+        private void WaitForSubscriber(XPublisherSocket socket)
+        {
+            var start = DateTime.UtcNow;
+            var subscribed = false;
+            for (var remaining = SubscriberWaitTimeout; !subscribed && remaining > TimeSpan.Zero; remaining = start + SubscriberWaitTimeout - DateTime.UtcNow)
+            {
+                byte[] frame;
+                subscribed = socket.TryReceiveFrameBytes(remaining, out frame) && IsSubscriptionTo(frame, _topic);
+            }
+
+            // Give other subscribers the same time to join as before
+            var settle = start + MinimumSubscriberWait - DateTime.UtcNow;
+            if (settle > TimeSpan.Zero)
+            {
+                Thread.Sleep(settle);
+            }
+        }
+
+        // A subscription frame is 1 followed by the subscribed prefix; 0 is an unsubscription
+        private static bool IsSubscriptionTo(byte[] frame, string topic)
+        {
+            return frame.Length > 0 && frame[0] == 1 &&
+                   topic.StartsWith(Encoding.UTF8.GetString(frame, 1, frame.Length - 1), StringComparison.Ordinal);
+        }
+
         private void Publish(TMessage message)
         {
-            _socket.Value.SendToTopic(_topic, message.GetType().Name, _serializer.Serialize(message));
+            var socket = _socket.Value;
+
+            // Discard subscription frames that arrived since, so they don't build up
+            byte[] frame;
+            while (socket.TryReceiveFrameBytes(TimeSpan.Zero, out frame))
+            {
+            }
+
+            socket.SendToTopic(_topic, message.GetType().Name, _serializer.Serialize(message));
         }
 
         public Task PublishAsync(TMessage message)