Run broker-backed transport tests in CI

.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

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 =>

Obvs.NATS.Tests/TestSourceAndPublisher.cs

@@ -19,7 +19,7 @@ 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
+                return cf.CreateConnection(TestBroker.Url);
             }, LazyThreadSafetyMode.ExecutionAndPublication);
 
             const string subjectPrefix = "Obvs.NATS.Tests";