feat: add NATS / JetStream messaging gateway

.agent_instructions/generated_tests.md

@@ -1,6 +1,6 @@
 # Generated Tests
 
-Brighter uses a test generation tool to produce consistent test suites across all provider implementations for both **outbox** (MSSQL, PostgreSQL, MySQL, SQLite, DynamoDB, MongoDB, GCP Spanner/Firestore) and **messaging gateway** (RabbitMQ, Kafka, AWS SNS/SQS, Redis, RocketMQ, GCP Pub/Sub, PostgreSQL, MSSQL) tests. See [ADR 0035](../docs/adr/0035-generated-test.md) for the design rationale.
+Brighter uses a test generation tool to produce consistent test suites across all provider implementations for both **outbox** (MSSQL, PostgreSQL, MySQL, SQLite, DynamoDB, MongoDB, GCP Spanner/Firestore) and **messaging gateway** (RabbitMQ, Kafka, AWS SNS/SQS, NATS, Redis, RocketMQ, GCP Pub/Sub, PostgreSQL, MSSQL) tests. See [ADR 0035](../docs/adr/0035-generated-test.md) for the design rationale.
 
 ## Key Principle
 
@@ -60,6 +60,7 @@ tests/Paramore.Brighter.*.Tests/
 | `Paramore.Brighter.AWS.Tests` | | ✅ | SnsStandard, SnsFifo, SqsStandard, SqsFifo |
 | `Paramore.Brighter.AWS.V4.Tests` | | ✅ | SnsStandard, SnsFifo, SqsStandard, SqsFifo |
 | `Paramore.Brighter.Kafka.Tests` | | ✅ | Standard, PartitionKey |
+| `Paramore.Brighter.NATS.Tests` | | ✅ | single |
 | `Paramore.Brighter.Redis.Tests` | | ✅ | single |
 | `Paramore.Brighter.RocketMQ.Tests` | | ✅ | single |
 

.github/workflows/ci.yml

@@ -217,6 +217,31 @@ jobs:
       - name: MQTT Transport Tests
         run: dotnet test ./tests/Paramore.Brighter.MQTT.Tests/Paramore.Brighter.MQTT.Tests.csproj --filter "Category=MQTT&Fragile!=CI" --configuration Release --logger "console;verbosity=normal" --logger GitHubActions --blame -v n
 
+  nats-ci:
+    runs-on: ubuntu-latest
+    timeout-minutes: 8
+    needs: [build]
+
+    services:
+      nats:
+        image: nats:latest
+        ports:
+        - 4222:4222
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v7
+      - name: Setup dotnet
+        uses: actions/setup-dotnet@v5
+        with:
+          dotnet-version: |
+            9.0.x
+            10.0.x
+      - name: Install dependencies
+        run: dotnet restore
+      - name: NATS Transport Tests
+        run: dotnet test ./tests/Paramore.Brighter.NATS.Tests/Paramore.Brighter.NATS.Tests.csproj --filter "Category=NATS&Fragile!=CI" --configuration Release --logger "console;verbosity=normal" --logger GitHubActions --blame -v n
+
   rabbitmq-async-ci:
     runs-on: ubuntu-latest
     timeout-minutes: 8

Brighter.slnx

@@ -223,6 +223,7 @@
     <Project Path="tests/Paramore.Brighter.MQTT.Tests/Paramore.Brighter.MQTT.Tests.csproj" />
     <Project Path="tests/Paramore.Brighter.MSSQL.Tests/Paramore.Brighter.MSSQL.Tests.csproj" />
     <Project Path="tests/Paramore.Brighter.MySQL.Tests/Paramore.Brighter.MySQL.Tests.csproj" />
+    <Project Path="tests/Paramore.Brighter.NATS.Tests/Paramore.Brighter.NATS.Tests.csproj" />
     <Project Path="tests/Paramore.Brighter.PostgresSQL.Tests/Paramore.Brighter.PostgresSQL.Tests.csproj" />
     <Project Path="tests/Paramore.Brighter.Quartz.Tests/Paramore.Brighter.Quartz.Tests.csproj" />
     <Project Path="tests/Paramore.Brighter.Redis.Tests/Paramore.Brighter.Redis.Tests.csproj" />
@@ -290,6 +291,7 @@
   <Project Path="src/Paramore.Brighter.MessagingGateway.GcpPubSub/Paramore.Brighter.MessagingGateway.GcpPubSub.csproj" />
   <Project Path="src/Paramore.Brighter.MessagingGateway.Kafka/Paramore.Brighter.MessagingGateway.Kafka.csproj" />
   <Project Path="src/Paramore.Brighter.MessagingGateway.MQTT/Paramore.Brighter.MessagingGateway.MQTT.csproj" />
+  <Project Path="src/Paramore.Brighter.MessagingGateway.NATS/Paramore.Brighter.MessagingGateway.NATS.csproj" />
   <Project Path="src/Paramore.Brighter.MessagingGateway.MsSql/Paramore.Brighter.MessagingGateway.MsSql.csproj" />
   <Project Path="src/Paramore.Brighter.MessagingGateway.Postgres/Paramore.Brighter.MessagingGateway.Postgres.csproj" />
   <Project Path="src/Paramore.Brighter.MessagingGateway.Redis/Paramore.Brighter.MessagingGateway.Redis.csproj" />

Directory.Packages.props

@@ -104,6 +104,7 @@
     <PackageVersion Include="MongoDB.Driver" Version="3.8.1" />
     <PackageVersion Include="MQTTnet" Version="4.3.7.1207" />
     <PackageVersion Include="MySqlConnector" Version="2.5.0" />
+    <PackageVersion Include="NATS.Net" Version="3.0.0" />
     <PackageVersion Include="Newtonsoft.Json" Version="13.0.4" />
     <PackageVersion Include="Neuroglia.AsyncApi.Core" Version="3.0.6" />
     <PackageVersion Include="Neuroglia.AsyncApi.IO" Version="3.0.6" />

docker-compose-nats.yaml

@@ -0,0 +1,16 @@
+version: '3.8'
+
+services:
+  nats:
+    image: nats:latest
+    hostname: nats-server
+    # -js enables JetStream; -sd sets the stream storage directory; -m enables HTTP monitoring
+    command: ["-js", "-sd", "/data", "-m", "8222"]
+    ports:
+      - "4222:4222"   # client connections (NATS_URL=nats://localhost:4222)
+      - "8222:8222"   # HTTP monitoring
+    volumes:
+      - nats_data:/data
+
+volumes:
+  nats_data:

specs/README.md

@@ -10,6 +10,7 @@ Spec numbers are not strictly sequential. Gaps exist to accommodate parallel wor
 |-------|------------|
 | 0001-0009 | General / early specs |
 | 0010-0019 | Universal DLQ - transport implementations |
+| 0030-0039 | Transport implementations |
 
 ## Specifications
 

src/Paramore.Brighter.MessagingGateway.NATS/Extensions/MessageHeaderExtensions.cs

@@ -0,0 +1,95 @@
+// The MIT License (MIT)
+// Copyright © 2014 Ian Cooper <ian_hammond_cooper@yahoo.co.uk>
+// 
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+//  of this software and associated documentation files (the "Software"), to deal
+//  in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+// 
+//  The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+// 
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+using NATS.Client.Core;
+
+namespace Paramore.Brighter.MessagingGateway.NATS.Extensions;
+
+internal static class MessageHeaderExtensions
+{
+    public static NatsHeaders ToNatsHeaders(this MessageHeader messageHeader)
+    {
+        var headers = new NatsHeaders();
+        headers[HeadersName.Id] = messageHeader.MessageId.Value;
+        headers[HeadersName.Baggage] = messageHeader.Baggage.ToString();
+        headers[HeadersName.ContentType] = messageHeader.ContentType.ToString();
+        headers[HeadersName.CorrelationId] = messageHeader.CorrelationId.Value;
+        headers[HeadersName.MessageType] = messageHeader.MessageType.ToString();
+        headers[HeadersName.SpecVersion] = messageHeader.SpecVersion;
+        headers[HeadersName.Source] = messageHeader.Source.ToString();
+        headers[HeadersName.Time] = messageHeader.TimeStamp.ToString("O");
+
+        if (messageHeader.DataSchema != null)
+        {
+            headers[HeadersName.DataSchema] = messageHeader.DataSchema.ToString();
+        }
+
+        if (messageHeader.DataRef != null)
+        {
+            headers[HeadersName.DataRef] = messageHeader.DataRef;
+        }
+
+        if (messageHeader.JobId != null)
+        {
+            headers[HeadersName.JobId] = messageHeader.JobId.ToString();
+        }
+
+        if (!RoutingKey.IsNullOrEmpty(messageHeader.ReplyTo))
+        {
+            headers[HeadersName.ReplyTo] = messageHeader.ReplyTo.Value;
+        }
+
+        if (messageHeader.Subject != null)
+        {
+            headers[HeadersName.Subject] = messageHeader.Subject;
+        }
+
+        if (messageHeader.TraceParent != null)
+        {
+            headers[HeadersName.TraceParent] = messageHeader.TraceParent.Value;
+        }
+
+        if (messageHeader.TraceState != null)
+        {
+            headers[HeadersName.TraceState] = messageHeader.TraceState.Value;
+        }
+
+        if (!string.IsNullOrEmpty(messageHeader.Type?.Value))
+        {
+            headers[HeadersName.Type] = messageHeader.Type.Value;
+        }
+
+        if (messageHeader.WorkflowId != null)
+        {
+            headers[HeadersName.WorkflowId] = messageHeader.WorkflowId.ToString();
+        }
+
+        foreach (var keyPair in messageHeader.Baggage)
+        {
+            if(keyPair.Value != null)
+            {
+                headers[keyPair.Key] = keyPair.Value;
+            }
+        }
+
+        return headers;
+    }
+}