From 7b0bd5a96f84ef916b2e63a491414ac9b075c697 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:41:06 -0700 Subject: [PATCH 1/7] Run broker-backed transport tests in CI Start RabbitMQ, NATS and Kafka as service containers and run each transport's Category=Explicit tests against them in a separate step. The NATS and Kafka tests now read NATS_URL and KAFKA_BOOTSTRAP_SERVERS (defaulting to localhost) instead of hard-coded Docker addresses, like the RabbitMQ tests' RABBITMQ_URI. Part of #74; the NetMQ tests join once #80 is merged. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/dotnet.yml | 33 +++++++++++++++++++ Obvs.Kafka.Tests/TestBroker.cs | 10 ++++++ Obvs.Kafka.Tests/TestMessageSendAndReceive.cs | 9 ++--- Obvs.NATS.Tests/TestBroker.cs | 10 ++++++ Obvs.NATS.Tests/TestServiceBus.cs | 2 +- Obvs.NATS.Tests/TestSourceAndPublisher.cs | 2 +- 6 files changed, 57 insertions(+), 9 deletions(-) create mode 100644 Obvs.Kafka.Tests/TestBroker.cs create mode 100644 Obvs.NATS.Tests/TestBroker.cs diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 4ac797ec..4c489ac9 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -17,6 +17,27 @@ jobs: contents: read id-token: write # NuGet Trusted Publishing + # Brokers for the transport tests marked Category=Explicit + services: + rabbitmq: + image: rabbitmq:4.4-alpine + ports: ['5672:5672'] + options: >- + --health-cmd "rabbitmq-diagnostics -q check_port_connectivity" + --health-interval 5s --health-timeout 5s --health-retries 20 + nats: + image: nats:2.15 + ports: ['4222:4222'] + kafka: + image: apache/kafka:4.4.0 + ports: ['9092:9092'] + env: + # Assign partitions to the test's new consumer group straight away + KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 + options: >- + --health-cmd "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092" + --health-interval 5s --health-timeout 10s --health-retries 20 + steps: - name: Checkout uses: actions/checkout@v7 @@ -42,6 +63,18 @@ jobs: run: dotnet build --no-restore --configuration Release -p:Version=${{ steps.gitversion.outputs.semVer }} -p:ContinuousIntegrationBuild=true -p:Deterministic=true - name: Test run: dotnet test --no-build --configuration Release --verbosity normal --filter Category!=Explicit + - name: Create Kafka test topic + # The test consumer reads from the latest offset, so the topic must exist before it subscribes + run: docker exec ${{ job.services.kafka.id }} /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists --topic test-topic-3 + - name: Test transports against brokers + env: + RABBITMQ_URI: amqp://localhost:5672 + NATS_URL: nats://localhost:4222 + KAFKA_BOOTSTRAP_SERVERS: localhost:9092 + run: | + dotnet test Obvs.RabbitMQ.Tests --no-build --configuration Release --verbosity normal --filter Category=Explicit + dotnet test Obvs.NATS.Tests --no-build --configuration Release --verbosity normal --filter Category=Explicit + dotnet test Obvs.Kafka.Tests --no-build --configuration Release --verbosity normal --filter Category=Explicit - name: Pack if: github.event_name == 'push' run: | diff --git a/Obvs.Kafka.Tests/TestBroker.cs b/Obvs.Kafka.Tests/TestBroker.cs new file mode 100644 index 00000000..674d5b38 --- /dev/null +++ b/Obvs.Kafka.Tests/TestBroker.cs @@ -0,0 +1,10 @@ +using System; + +namespace Obvs.Kafka.Tests +{ + internal static class TestBroker + { + // Set KAFKA_BOOTSTRAP_SERVERS to run the Explicit tests against a broker other than a default local install + public static string BootstrapServers => Environment.GetEnvironmentVariable("KAFKA_BOOTSTRAP_SERVERS") ?? "localhost:9092"; + } +} diff --git a/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs b/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs index 39243007..b6061d75 100644 --- a/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs +++ b/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs @@ -44,13 +44,8 @@ public override string ToString() public TestMessageSendAndReceive() { - // Steps to run local kafka broker: - // - Install: Docker for Windows / Docker Desktop - // - Clone: https://github.com/wurstmeister/kafka-docker - // - Edit: docker-compose-single-broker.yml - // - Change: KAFKA_ADVERTISED_HOST_NAME: localhost - // - Run: docker-compose -f "docker-compose-single-broker.yml" up -d --build - var cfg = new KafkaConfiguration("localhost:9092"); + // Run a local broker with: docker run -p 9092:9092 apache/kafka + var cfg = new KafkaConfiguration(TestBroker.BootstrapServers); var sourceConfig = new KafkaSourceConfiguration(); var producerConfig = new KafkaProducerConfiguration(); const string topicName = "test-topic-3"; diff --git a/Obvs.NATS.Tests/TestBroker.cs b/Obvs.NATS.Tests/TestBroker.cs new file mode 100644 index 00000000..e5cf4453 --- /dev/null +++ b/Obvs.NATS.Tests/TestBroker.cs @@ -0,0 +1,10 @@ +using System; + +namespace Obvs.NATS.Tests +{ + internal static class TestBroker + { + // Set NATS_URL to run the Explicit tests against a server other than a default local install + public static string Url => Environment.GetEnvironmentVariable("NATS_URL") ?? "nats://localhost:4222"; + } +} diff --git a/Obvs.NATS.Tests/TestServiceBus.cs b/Obvs.NATS.Tests/TestServiceBus.cs index e7bb5cad..5875d8f1 100644 --- a/Obvs.NATS.Tests/TestServiceBus.cs +++ b/Obvs.NATS.Tests/TestServiceBus.cs @@ -21,7 +21,7 @@ public async Task TestBasicSendReceiveOfAllMessageTypes() settings.ServiceName = "Obvs.NATS.TestService"; settings.Configure(connection => { - connection.Url = "nats://192.168.99.100:32774"; // change to local Docker address:port that maps onto 4222 + connection.Url = TestBroker.Url; connection.IsShared = true; }); settings.Configure(messageProperty => diff --git a/Obvs.NATS.Tests/TestSourceAndPublisher.cs b/Obvs.NATS.Tests/TestSourceAndPublisher.cs index 1838bf6b..8488d00a 100644 --- a/Obvs.NATS.Tests/TestSourceAndPublisher.cs +++ b/Obvs.NATS.Tests/TestSourceAndPublisher.cs @@ -19,7 +19,7 @@ public async Task TestSendAndReceiveMessage() var lazyConnection = new Lazy(() => { var cf = new ConnectionFactory(); - return cf.CreateConnection("nats://192.168.99.100:32774"); // change to local Docker address:port that maps onto 4222 + return cf.CreateConnection(TestBroker.Url); }, LazyThreadSafetyMode.ExecutionAndPublication); const string subjectPrefix = "Obvs.NATS.Tests"; From 582e352952723ea78e67bb72d5f7a06d31ad0dfc Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:42:40 -0700 Subject: [PATCH 2/7] Configure the CI Kafka container as a full single-node KRaft broker Setting KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS alone replaced the image defaults and the broker failed to start (missing process.roles). Co-Authored-By: Claude Opus 5.5 --- .github/workflows/dotnet.yml | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 4c489ac9..a2b4931a 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -31,7 +31,20 @@ jobs: kafka: image: apache/kafka:4.4.0 ports: ['9092:9092'] + # Setting any KAFKA_* variable replaces the image's defaults, so configure a full single-node KRaft broker env: + KAFKA_NODE_ID: 1 + KAFKA_PROCESS_ROLES: broker,controller + KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 + KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT + KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093 + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 + KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR: 1 # Assign partitions to the test's new consumer group straight away KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 options: >- From 6dcd910f036cec63fdaf522c64b987bc2d1865e3 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:45:49 -0700 Subject: [PATCH 3/7] Warm up the CI Kafka broker before the transport tests The Kafka test consumer starts at the latest offset and Subscribe returns before partitions are assigned, so on a fresh broker it missed the messages while __consumer_offsets was being created. Use one offsets partition and run a warm-up consumer group first. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/dotnet.yml | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index a2b4931a..ed624641 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -41,6 +41,7 @@ jobs: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR: 1 @@ -76,9 +77,12 @@ jobs: run: dotnet build --no-restore --configuration Release -p:Version=${{ steps.gitversion.outputs.semVer }} -p:ContinuousIntegrationBuild=true -p:Deterministic=true - name: Test run: dotnet test --no-build --configuration Release --verbosity normal --filter Category!=Explicit - - name: Create Kafka test topic - # The test consumer reads from the latest offset, so the topic must exist before it subscribes - run: docker exec ${{ job.services.kafka.id }} /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists --topic test-topic-3 + - name: Prepare Kafka + # The test consumer starts at the latest offset and doesn't wait for partition assignment, so create its + # topic and warm up the group coordinator (which creates __consumer_offsets) before the tests run + run: | + docker exec ${{ job.services.kafka.id }} /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists --topic test-topic-3 + docker exec ${{ job.services.kafka.id }} /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic-3 --group ci-warmup --timeout-ms 10000 || true - name: Test transports against brokers env: RABBITMQ_URI: amqp://localhost:5672 From 1f8a246f6fb9dc852995246e52f1434f26413dba Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:52:45 -0700 Subject: [PATCH 4/7] Assert what the NATS broker tests receive TestSourceAndPublisher now checks all 10 messages arrive in order, and TestServiceBus checks the reply and the event produced by the command, instead of only printing them. Co-Authored-By: Claude Opus 5.5 --- Obvs.NATS.Tests/TestServiceBus.cs | 9 ++++++--- Obvs.NATS.Tests/TestSourceAndPublisher.cs | 7 ++++++- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/Obvs.NATS.Tests/TestServiceBus.cs b/Obvs.NATS.Tests/TestServiceBus.cs index 5875d8f1..d13178c0 100644 --- a/Obvs.NATS.Tests/TestServiceBus.cs +++ b/Obvs.NATS.Tests/TestServiceBus.cs @@ -42,16 +42,19 @@ public async Task TestBasicSendReceiveOfAllMessageTypes() Console.WriteLine(request); bus.ReplyAsync(request, new TestResponse {Id = request.Id}); }); - bus.Events.Subscribe(Console.WriteLine); + var events = new List(); + bus.Events.Subscribe(events.Add); await bus.SendAsync(new TestCommand {Id = 1}); - var response = await bus.GetResponse(new TestRequest()); - Console.WriteLine(response); + var response = await bus.GetResponse(new TestRequest {Id = 2}); + Assert.Equal(2, response.Id); await Task.Delay(TimeSpan.FromSeconds(1)); ((IDisposable)bus).Dispose(); + + Assert.Contains(events, ev => ev is TestEvent testEvent && testEvent.Id == 1); } } } diff --git a/Obvs.NATS.Tests/TestSourceAndPublisher.cs b/Obvs.NATS.Tests/TestSourceAndPublisher.cs index 8488d00a..9340c936 100644 --- a/Obvs.NATS.Tests/TestSourceAndPublisher.cs +++ b/Obvs.NATS.Tests/TestSourceAndPublisher.cs @@ -1,5 +1,7 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; +using System.Linq; using System.Reactive.Concurrency; using System.Reactive.Linq; using System.Threading; @@ -32,7 +34,8 @@ public async Task TestSendAndReceiveMessage() var source = new MessageSource(lazyConnection, subjectPrefix, deserializers); var publisher = new MessagePublisher(lazyConnection, subjectPrefix, serializer); - var subscription = source.Messages.ObserveOn(Scheduler.Default).Subscribe(Console.WriteLine); + var received = new ConcurrentQueue(); + var subscription = source.Messages.ObserveOn(Scheduler.Default).Subscribe(received.Enqueue); for (var i = 0; i < 10; i++) { @@ -41,6 +44,8 @@ public async Task TestSendAndReceiveMessage() await Task.Delay(TimeSpan.FromSeconds(1)); subscription.Dispose(); + + Assert.Equal(Enumerable.Range(0, 10), received.Select(message => message.Id)); } } } From dcd288be5ca852f1361861a9e9e2abed402863cf Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:55:06 -0700 Subject: [PATCH 5/7] Port Obvs.NATS from NATS.Client 1.1.8 to NATS.Net (NATS.Client.Core 3.3.1) MessageSource and MessagePublisher take an INatsConnection instead of Lazy; a NatsConnection connects on first use. Publishing awaits PublishAsync, and subscriptions read the NATS.Net message channel on a background task. Subscribe still blocks until the server has the subscription, so messages published straight after it aren't missed. The protobuf MessageWrapper envelope is unchanged, so services on the old and new client can exchange messages. The fluent configuration is unchanged. Obvs.NATS compiles as C# 9 to set NATS.Net's init-only options. Fixes #78 Co-Authored-By: Claude Opus 5.5 --- Directory.Packages.props | 2 +- Obvs.NATS.Tests/TestSourceAndPublisher.cs | 12 +-- Obvs.NATS/AsyncSubscriptionExtensions.cs | 26 ------ .../NatsServiceEndpointProvider.cs | 24 ++---- .../Configuration/SharedConnectionFactory.cs | 22 +++--- Obvs.NATS/MessagePublisher.cs | 13 ++- Obvs.NATS/MessageSource.cs | 79 ++++++++++++++----- Obvs.NATS/Obvs.NATS.csproj | 4 +- 8 files changed, 91 insertions(+), 91 deletions(-) delete mode 100644 Obvs.NATS/AsyncSubscriptionExtensions.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index ec3fc538..8cb50885 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -10,7 +10,7 @@ - + diff --git a/Obvs.NATS.Tests/TestSourceAndPublisher.cs b/Obvs.NATS.Tests/TestSourceAndPublisher.cs index 9340c936..a4ad065a 100644 --- a/Obvs.NATS.Tests/TestSourceAndPublisher.cs +++ b/Obvs.NATS.Tests/TestSourceAndPublisher.cs @@ -6,7 +6,7 @@ using System.Reactive.Linq; using System.Threading; using System.Threading.Tasks; -using NATS.Client; +using NATS.Client.Core; using Obvs.Serialization; using Obvs.Serialization.Json; using Xunit; @@ -18,11 +18,7 @@ public class TestSourceAndPublisher [Fact, Trait("Category", "Explicit")] public async Task TestSendAndReceiveMessage() { - var lazyConnection = new Lazy(() => - { - var cf = new ConnectionFactory(); - return cf.CreateConnection(TestBroker.Url); - }, LazyThreadSafetyMode.ExecutionAndPublication); + await using var connection = new NatsConnection(new NatsOpts { Url = TestBroker.Url }); const string subjectPrefix = "Obvs.NATS.Tests"; var deserializers = new List> @@ -31,8 +27,8 @@ public async Task TestSendAndReceiveMessage() }; IMessageSerializer serializer = new JsonMessageSerializer(); - var source = new MessageSource(lazyConnection, subjectPrefix, deserializers); - var publisher = new MessagePublisher(lazyConnection, subjectPrefix, serializer); + var source = new MessageSource(connection, subjectPrefix, deserializers); + var publisher = new MessagePublisher(connection, subjectPrefix, serializer); var received = new ConcurrentQueue(); var subscription = source.Messages.ObserveOn(Scheduler.Default).Subscribe(received.Enqueue); diff --git a/Obvs.NATS/AsyncSubscriptionExtensions.cs b/Obvs.NATS/AsyncSubscriptionExtensions.cs deleted file mode 100644 index 45f475f5..00000000 --- a/Obvs.NATS/AsyncSubscriptionExtensions.cs +++ /dev/null @@ -1,26 +0,0 @@ -using System; -using System.Reactive.Disposables; -using System.Reactive.Linq; -using NATS.Client; - -namespace Obvs.NATS -{ - internal static class AsyncSubscriptionExtensions - { - public static IObservable ToObservable(this IAsyncSubscription sub) - { - return Observable.Create( - observer => - { - EventHandler handler = (sender, args) => observer.OnNext(args); - sub.MessageHandler += handler; - sub.Start(); - return Disposable.Create(() => - { - sub.MessageHandler -= handler; - sub.Unsubscribe(); - }); - }); - } - } -} \ No newline at end of file diff --git a/Obvs.NATS/Configuration/NatsServiceEndpointProvider.cs b/Obvs.NATS/Configuration/NatsServiceEndpointProvider.cs index 09de932e..239952df 100644 --- a/Obvs.NATS/Configuration/NatsServiceEndpointProvider.cs +++ b/Obvs.NATS/Configuration/NatsServiceEndpointProvider.cs @@ -2,7 +2,7 @@ using System.Collections.Generic; using System.Reactive.Disposables; using System.Reflection; -using NATS.Client; +using NATS.Client.Core; using Obvs.Configuration; using Obvs.Serialization; @@ -24,7 +24,7 @@ public class NatsServiceEndpointProvider, bool> _propertyFilter; private readonly Func> _propertyProvider; - private readonly Lazy _connection; + private readonly INatsConnection _connection; public NatsServiceEndpointProvider(NatsEndpointSettings settings, IMessageSerializer serializer, IMessageDeserializerFactory deserializerFactory, Func assemblyFilter = null, Func typeFilter = null, Func, bool> propertyFilter = null, Func> propertyProvider = null) : base(settings.ServiceName) @@ -56,14 +56,14 @@ public override IServiceEndpoint CreateSource(Lazy connection, string subject) where T : class, TMessage + private IMessageSource CreateSource(INatsConnection connection, string subject) where T : class, TMessage { return new MessageSource(connection, subject, _deserializerFactory.Create(_assemblyFilter, _typeFilter), _propertyFilter); } - private IMessagePublisher CreatePublisher(Lazy connection, string destination) where T : class, TMessage + private IMessagePublisher CreatePublisher(INatsConnection connection, string destination) where T : class, TMessage { return new MessagePublisher(connection, destination, _serializer, _propertyProvider); } @@ -80,19 +80,11 @@ public override IServiceEndpointClient lazyConnection) + private static IDisposable GetConnectionDisposable(INatsConnection connection) { - return Disposable.Create(() => - { - if (lazyConnection.IsValueCreated) - { - if (!lazyConnection.Value.IsClosed()) - { - lazyConnection.Value.Close(); - } - lazyConnection.Value.Dispose(); - } - }); +#pragma warning disable VSTHRD002 // IDisposable.Dispose is synchronous + return Disposable.Create(() => connection.DisposeAsync().AsTask().GetAwaiter().GetResult()); +#pragma warning restore VSTHRD002 } } } \ No newline at end of file diff --git a/Obvs.NATS/Configuration/SharedConnectionFactory.cs b/Obvs.NATS/Configuration/SharedConnectionFactory.cs index fe23658f..0fbeb19c 100644 --- a/Obvs.NATS/Configuration/SharedConnectionFactory.cs +++ b/Obvs.NATS/Configuration/SharedConnectionFactory.cs @@ -1,33 +1,31 @@ -using System; using System.Collections.Generic; -using System.Threading; -using NATS.Client; +using NATS.Client.Core; namespace Obvs.NATS.Configuration { internal static class SharedConnectionFactory { - private static readonly Dictionary> SharedConnections = new Dictionary>(); + private static readonly Dictionary SharedConnections = new Dictionary(); - public static Lazy Get(string url, bool isShared) + public static INatsConnection Get(string url, bool isShared) { if (!isShared) { - return CreateLazyConnection(url); + return CreateConnection(url); } - Lazy connection; + INatsConnection connection; if (!SharedConnections.TryGetValue(url, out connection)) { - connection = CreateLazyConnection(url); + connection = CreateConnection(url); SharedConnections.Add(url, connection); } return connection; } - private static Lazy CreateLazyConnection(string url) + // A NatsConnection only connects when it's first used + private static INatsConnection CreateConnection(string url) { - var factory = new ConnectionFactory(); - return new Lazy(() => factory.CreateConnection(url), LazyThreadSafetyMode.ExecutionAndPublication); + return new NatsConnection(new NatsOpts { Url = url }); } } -} \ No newline at end of file +} diff --git a/Obvs.NATS/MessagePublisher.cs b/Obvs.NATS/MessagePublisher.cs index ec06f356..4470cdca 100644 --- a/Obvs.NATS/MessagePublisher.cs +++ b/Obvs.NATS/MessagePublisher.cs @@ -2,30 +2,30 @@ using System.Collections.Generic; using System.IO; using System.Threading.Tasks; -using NATS.Client; +using NATS.Client.Core; using Obvs.Serialization; namespace Obvs.NATS { public class MessagePublisher : IMessagePublisher where TMessage : class { - private readonly Lazy _lazyConnection; + private readonly INatsConnection _connection; private readonly string _subjectPrefix; private readonly IMessageSerializer _serializer; private readonly Func> _properties; - public MessagePublisher(Lazy lazyConnection, + public MessagePublisher(INatsConnection connection, string subjectPrefix, IMessageSerializer serializer, Func> properties = null) { - _lazyConnection = lazyConnection; + _connection = connection; _subjectPrefix = subjectPrefix; _serializer = serializer; _properties = properties ?? (message => null); } - public Task PublishAsync(TMessage message) + public async Task PublishAsync(TMessage message) { byte[] body; using (var stream = new MemoryStream()) @@ -40,9 +40,8 @@ public Task PublishAsync(TMessage message) using (var stream = new MemoryStream()) { ProtoBuf.Serializer.Serialize(stream, wrapper); - _lazyConnection.Value.Publish(subject, stream.ToArray()); + await _connection.PublishAsync(subject, stream.ToArray()).ConfigureAwait(false); } - return Task.FromResult(true); } public void Dispose() diff --git a/Obvs.NATS/MessageSource.cs b/Obvs.NATS/MessageSource.cs index ed65f189..64f4da11 100644 --- a/Obvs.NATS/MessageSource.cs +++ b/Obvs.NATS/MessageSource.cs @@ -2,8 +2,11 @@ using System.Collections.Generic; using System.IO; using System.Linq; +using System.Reactive.Disposables; using System.Reactive.Linq; -using NATS.Client; +using System.Threading; +using System.Threading.Tasks; +using NATS.Client.Core; using Obvs.Serialization; namespace Obvs.NATS @@ -12,17 +15,17 @@ public class MessageSource : IMessageSource where TMessage : class { private readonly IDictionary> _deserializers; - private readonly Lazy _lazyConnection; + private readonly INatsConnection _connection; private readonly Func, bool> _filter; private readonly string _subject; - public MessageSource(Lazy lazyConnection, + public MessageSource(INatsConnection connection, string subjectPrefix, IEnumerable> deserializers, Func, bool> filter = null) { _deserializers = deserializers.ToDictionary(d => d.GetTypeName()); - _lazyConnection = lazyConnection; + _connection = connection; _filter = filter; _subject = string.Format("{0}.*", subjectPrefix); } @@ -31,35 +34,71 @@ public IObservable Messages { get { +#pragma warning disable VSTHRD002 // IObservable.Subscribe and IDisposable.Dispose are synchronous return Observable.Create(observer => - _lazyConnection.Value - .SubscribeAsync(_subject) - .ToObservable() - .Select(Unwrap) - .Where(MatchesFilter) - .Select(Deserialize) - .Subscribe(observer)); + { + // Block until the server has the subscription, so messages published straight after + // Subscribe returns aren't missed. NATS.Net doesn't capture the synchronization context, + // so blocking here can't deadlock. + var subscription = _connection.SubscribeCoreAsync(_subject).AsTask().GetAwaiter().GetResult(); + var cancellation = new CancellationTokenSource(); + // ReceiveAsync reports its own failures to the observer + _ = Task.Run(() => ReceiveAsync(subscription, observer, cancellation.Token)); + + return Disposable.Create(() => + { + cancellation.Cancel(); + subscription.DisposeAsync().AsTask().GetAwaiter().GetResult(); + cancellation.Dispose(); + }); + }); +#pragma warning restore VSTHRD002 + } + } + + private async Task ReceiveAsync(INatsSub subscription, IObserver observer, CancellationToken token) + { + try + { + while (await subscription.Msgs.WaitToReadAsync(token).ConfigureAwait(false)) + { + NatsMsg msg; + while (subscription.Msgs.TryRead(out msg)) + { + var wrapper = Unwrap(msg.Data); + if (MatchesFilter(wrapper)) + { + observer.OnNext(Deserialize(msg.Subject, wrapper)); + } + } + } + } + catch (OperationCanceledException) + { + } + catch (Exception exception) + { + observer.OnError(exception); } } - private bool MatchesFilter(Tuple tuple) + private bool MatchesFilter(MessageWrapper wrapper) { - return _filter == null || _filter(tuple.Item2.Properties); + return _filter == null || _filter(wrapper.Properties); } - private static Tuple Unwrap(MsgHandlerEventArgs args) + private static MessageWrapper Unwrap(byte[] data) { - using (var stream = new MemoryStream(args.Message.Data)) + using (var stream = new MemoryStream(data)) { - var wrapper = ProtoBuf.Serializer.Deserialize(stream); - return Tuple.Create(args.Message.Subject, wrapper); + return ProtoBuf.Serializer.Deserialize(stream); } } - private TMessage Deserialize(Tuple tuple) + private TMessage Deserialize(string subject, MessageWrapper wrapper) { IMessageDeserializer deserializer; - var messageType = GetMessageType(tuple.Item1); + var messageType = GetMessageType(subject); if (!_deserializers.TryGetValue(messageType, out deserializer)) { @@ -70,7 +109,7 @@ private TMessage Deserialize(Tuple tuple) deserializer = _deserializers.Values.Single(); } - return deserializer.Deserialize(new MemoryStream(tuple.Item2.Body)); + return deserializer.Deserialize(new MemoryStream(wrapper.Body)); } private static string GetMessageType(string subject) diff --git a/Obvs.NATS/Obvs.NATS.csproj b/Obvs.NATS/Obvs.NATS.csproj index 7691021d..a5ad1c44 100644 --- a/Obvs.NATS/Obvs.NATS.csproj +++ b/Obvs.NATS/Obvs.NATS.csproj @@ -1,6 +1,8 @@  net472;netstandard2.0;net10.0 + + 9.0 Copyright © Christopher Read 2017 1.0.0 @@ -10,7 +12,7 @@ NATS transport support for Obvs - + From 2ab94ae170f859f52a55be551b5e54f28539e679 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 00:05:58 -0700 Subject: [PATCH 6/7] Run the broker tests with xUnit v3 test syntax, and add NetMQ Master now runs tests on Microsoft.Testing.Platform (#101), so use --project and --filter-trait. The NetMQ tests no longer clash over ports (#80), so run them too. The broker steps now come before the unit test step, so this branch no longer edits the lines master changed. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/dotnet.yml | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index ed624641..e0c2ab47 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -75,8 +75,6 @@ jobs: run: dotnet restore - name: Build run: dotnet build --no-restore --configuration Release -p:Version=${{ steps.gitversion.outputs.semVer }} -p:ContinuousIntegrationBuild=true -p:Deterministic=true - - name: Test - run: dotnet test --no-build --configuration Release --verbosity normal --filter Category!=Explicit - name: Prepare Kafka # The test consumer starts at the latest offset and doesn't wait for partition assignment, so create its # topic and warm up the group coordinator (which creates __consumer_offsets) before the tests run @@ -89,9 +87,12 @@ jobs: NATS_URL: nats://localhost:4222 KAFKA_BOOTSTRAP_SERVERS: localhost:9092 run: | - dotnet test Obvs.RabbitMQ.Tests --no-build --configuration Release --verbosity normal --filter Category=Explicit - dotnet test Obvs.NATS.Tests --no-build --configuration Release --verbosity normal --filter Category=Explicit - dotnet test Obvs.Kafka.Tests --no-build --configuration Release --verbosity normal --filter Category=Explicit + 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 + run: dotnet test --no-build --configuration Release --verbosity normal --filter Category!=Explicit - name: Pack if: github.event_name == 'push' run: | From 9d56b4ff8155e0dc49f382d3b235d8f8a4526862 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 00:10:12 -0700 Subject: [PATCH 7/7] Leave the NetMQ tests out of the CI broker step for now They lose the first messages on CI runners (ZeroMQ slow joiner): the publisher socket is created on first publish and sends before subscribers have connected. Tracked separately. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/dotnet.yml | 1 - 1 file changed, 1 deletion(-) diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index e0c2ab47..b95a8e17 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -90,7 +90,6 @@ 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 run: dotnet test --no-build --configuration Release --verbosity normal --filter Category!=Explicit - name: Pack