Port Obvs.NATS from NATS.Client to NATS.Net

.github/workflows/dotnet.yml

@@ -17,6 +17,41 @@ 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']
+        # 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_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
+          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: >-
+          --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
@@ -40,6 +75,21 @@ 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: 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
+        NATS_URL: nats://localhost:4222
+        KAFKA_BOOTSTRAP_SERVERS: localhost:9092
+      run: |
+        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"
     - name: Test
       run: dotnet test --no-build --configuration Release --verbosity normal --filter Category!=Explicit 
     - name: Pack

Directory.Packages.props

@@ -10,7 +10,7 @@
         <PackageVersion Include="MessagePack" Version="3.1.12"/>
         <PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="10.0.13"/>
         <PackageVersion Include="MsgPack.Cli" Version="1.0.1"/>
-        <PackageVersion Include="NATS.Client" Version="1.1.8"/>
+        <PackageVersion Include="NATS.Client.Core" Version="3.3.1"/>
         <PackageVersion Include="NEST" Version="7.17.5"/>
         <PackageVersion Include="NetJSON" Version="1.4.7"/>
         <PackageVersion Include="NetMQ" Version="4.0.4.5"/>

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";
+    }
+}

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";

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";
+    }
+}

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 =>
@@ -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<IEvent>();
+            bus.Events.Subscribe(events.Add);
 
             await bus.SendAsync(new TestCommand {Id = 1});
 
-            var response = await bus.GetResponse<TestResponse>(new TestRequest());
-            Console.WriteLine(response);
+            var response = await bus.GetResponse<TestResponse>(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);
         }
     }
 }

Obvs.NATS.Tests/TestSourceAndPublisher.cs

@@ -1,10 +1,12 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Linq;
 using System.Reactive.Concurrency;
 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;
@@ -16,11 +18,7 @@ public class TestSourceAndPublisher
         [Fact, Trait("Category", "Explicit")]
         public async Task TestSendAndReceiveMessage()
         {
-            var lazyConnection = new Lazy<IConnection>(() =>
-            {
-                var cf = new ConnectionFactory();
-                return cf.CreateConnection("nats://192.168.99.100:32774"); // change to local Docker address:port that maps onto 4222
-            }, LazyThreadSafetyMode.ExecutionAndPublication);
+            await using var connection = new NatsConnection(new NatsOpts { Url = TestBroker.Url });
 
             const string subjectPrefix = "Obvs.NATS.Tests";
             var deserializers = new List<IMessageDeserializer<TestMessage>>
@@ -29,10 +27,11 @@ public async Task TestSendAndReceiveMessage()
             };
             IMessageSerializer serializer = new JsonMessageSerializer();
 
-            var source = new MessageSource<TestMessage>(lazyConnection, subjectPrefix, deserializers);
-            var publisher = new MessagePublisher<TestMessage>(lazyConnection, subjectPrefix, serializer);
+            var source = new MessageSource<TestMessage>(connection, subjectPrefix, deserializers);
+            var publisher = new MessagePublisher<TestMessage>(connection, subjectPrefix, serializer);
 
-            var subscription = source.Messages.ObserveOn(Scheduler.Default).Subscribe(Console.WriteLine);
+            var received = new ConcurrentQueue<TestMessage>();
+            var subscription = source.Messages.ObserveOn(Scheduler.Default).Subscribe(received.Enqueue);
 
             for (var i = 0; i < 10; i++)
             {
@@ -41,6 +40,8 @@ public async Task TestSendAndReceiveMessage()
 
             await Task.Delay(TimeSpan.FromSeconds(1));
             subscription.Dispose();
+
+            Assert.Equal(Enumerable.Range(0, 10), received.Select(message => message.Id));
         }
     }
 }

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<TServiceMessage, TMessage, TCommand, TE
 
         private readonly Func<IDictionary<string, string>, bool> _propertyFilter;
         private readonly Func<TMessage, Dictionary<string, string>> _propertyProvider;
-        private readonly Lazy<IConnection> _connection;
+        private readonly INatsConnection _connection;
 
         public NatsServiceEndpointProvider(NatsEndpointSettings<TMessage> settings, IMessageSerializer serializer, IMessageDeserializerFactory deserializerFactory, Func<Assembly, bool> assemblyFilter = null, Func<Type, bool> typeFilter = null, Func<IDictionary<string, string>, bool> propertyFilter = null, Func<TMessage, Dictionary<string, string>> propertyProvider = null)
             : base(settings.ServiceName)
@@ -56,14 +56,14 @@ public override IServiceEndpoint<TMessage, TCommand, TEvent, TRequest, TResponse
                 GetConnectionDisposable(_connection));
         }
 
-        private IMessageSource<T> CreateSource<T>(Lazy<IConnection> connection, string subject) where T : class, TMessage
+        private IMessageSource<T> CreateSource<T>(INatsConnection connection, string subject) where T : class, TMessage
         {
             return new MessageSource<T>(connection, subject,
                 _deserializerFactory.Create<T, TServiceMessage>(_assemblyFilter, _typeFilter), 
                 _propertyFilter);
         }
 
-        private IMessagePublisher<T> CreatePublisher<T>(Lazy<IConnection> connection, string destination) where T : class, TMessage
+        private IMessagePublisher<T> CreatePublisher<T>(INatsConnection connection, string destination) where T : class, TMessage
         {
            return new MessagePublisher<T>(connection, destination, _serializer, _propertyProvider);
         }
@@ -80,19 +80,11 @@ public override IServiceEndpointClient<TMessage, TCommand, TEvent, TRequest, TRe
                 GetConnectionDisposable(_connection));
         }
 
-        private static IDisposable GetConnectionDisposable(Lazy<IConnection> 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
         }
     }
 }

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<string, Lazy<IConnection>> SharedConnections = new Dictionary<string, Lazy<IConnection>>();
+        private static readonly Dictionary<string, INatsConnection> SharedConnections = new Dictionary<string, INatsConnection>();
 
-        public static Lazy<IConnection> Get(string url, bool isShared)
+        public static INatsConnection Get(string url, bool isShared)
         {
             if (!isShared)
             {
-                return CreateLazyConnection(url);
+                return CreateConnection(url);
             }
-            Lazy<IConnection> connection;
+            INatsConnection connection;
             if (!SharedConnections.TryGetValue(url, out connection))
             {
-                connection = CreateLazyConnection(url);
+                connection = CreateConnection(url);
                 SharedConnections.Add(url, connection);
             }
             return connection;
         }
 
-        private static Lazy<IConnection> 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<IConnection>(() => factory.CreateConnection(url), LazyThreadSafetyMode.ExecutionAndPublication);
+            return new NatsConnection(new NatsOpts { Url = url });
         }
     }
-}
+}

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<TMessage> : IMessagePublisher<TMessage> where TMessage : class
     {
-        private readonly Lazy<IConnection> _lazyConnection;
+        private readonly INatsConnection _connection;
         private readonly string _subjectPrefix;
         private readonly IMessageSerializer _serializer;
         private readonly Func<TMessage, Dictionary<string, string>> _properties;
 
-        public MessagePublisher(Lazy<IConnection> lazyConnection,
+        public MessagePublisher(INatsConnection connection,
             string subjectPrefix,
             IMessageSerializer serializer,
             Func<TMessage, Dictionary<string, string>> 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()