From 056c619c8b134a0123104e311f9aa2c45161934e Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:51:23 -0700 Subject: [PATCH] NetMQ: wait for a subscriber before the first publish (slow joiner) The publisher slept a fixed second after binding before its first send, so subscribers slower than that (as on CI runners) missed the first messages. Use an XPUB socket, which receives subscriptions: still give subscribers at least the same second to join, and if none has subscribed to the topic by then, keep waiting for one, up to 5 seconds. Run the NetMQ tests in the CI broker step again. Fixes #102 Co-Authored-By: Claude Opus 5.5 --- .github/workflows/dotnet.yml | 1 + Obvs.NetMQ.Tests/TestSlowJoiner.cs | 58 +++++++++++++++++++++++ Obvs.NetMQ/Extensions/SocketExtensions.cs | 2 +- Obvs.NetMQ/MessagePublisher.cs | 52 ++++++++++++++++++-- 4 files changed, 107 insertions(+), 6 deletions(-) create mode 100644 Obvs.NetMQ.Tests/TestSlowJoiner.cs diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 9b8bce02..1bdea7be 100644 --- a/.github/workflows/dotnet.yml +++ b/.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. diff --git a/Obvs.NetMQ.Tests/TestSlowJoiner.cs b/Obvs.NetMQ.Tests/TestSlowJoiner.cs new file mode 100644 index 00000000..32538d7f --- /dev/null +++ b/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(); + var publisher = new MessagePublisher(address, new JsonMessageSerializer(), Topic); + var source = new MessageSource(address, new IMessageDeserializer[] { new JsonMessageDeserializer() }, 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 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; } + } + } +} diff --git a/Obvs.NetMQ/Extensions/SocketExtensions.cs b/Obvs.NetMQ/Extensions/SocketExtensions.cs index ffad22c4..d584abe9 100644 --- a/Obvs.NetMQ/Extensions/SocketExtensions.cs +++ b/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[] diff --git a/Obvs.NetMQ/MessagePublisher.cs b/Obvs.NetMQ/MessagePublisher.cs index 26e50e7f..f3d5668e 100644 --- a/Obvs.NetMQ/MessagePublisher.cs +++ b/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,9 +15,14 @@ public class MessagePublisher : IMessagePublisher where TMes private readonly string _address; private readonly IMessageSerializer _serializer; private readonly string _topic; - private readonly Lazy _socket; + private readonly Lazy _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; @@ -23,18 +30,53 @@ public MessagePublisher(string address, IMessageSerializer serializer, string to _topic = topic; _socketType = socketType; - _socket = new Lazy(() => + _socket = new Lazy(() => { - 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)