NATS Aspire container

playground/nats/Nats.ApiService/Nats.ApiService.csproj

@@ -1,3 +1,3 @@
 <Project Sdk="Microsoft.NET.Sdk.Web">
 
   <PropertyGroup>
@@ -7,6 +7,10 @@
     <InvariantGlobalization>true</InvariantGlobalization>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Swashbuckle.AspNetCore" />
+  </ItemGroup>
+
   <ItemGroup>
     <ProjectReference Include="..\..\..\src\Components\Aspire.Nats\Aspire.Nats.csproj" />
     <ProjectReference Include="..\..\Playground.ServiceDefaults\Playground.ServiceDefaults.csproj" />

playground/nats/Nats.ApiService/Nats.ApiService.http

@@ -0,0 +1,47 @@
+@HostAddress = http://localhost:5156
+
+POST {{HostAddress}}/stream/
+Content-Type: application/json
+
+{
+  "name": "EVENTS",
+  "description": "Events Description",
+  "subjects": ["events.>"],
+  "max_msgs": 10
+}
+
+###
+
+GET {{HostAddress}}/stream/EVENTS
+Accept: application/json
+
+###
+
+POST {{HostAddress}}/publish/
+Content-Type: application/json
+
+{
+  "subject": "events.mouse.click",
+  "name": "Mouse Event",
+  "description": "Mouse Event Description",
+  "priority": 0.1
+}
+
+###
+
+POST {{HostAddress}}/publish/
+Content-Type: application/json
+
+{
+  "subject": "events.key.a",
+  "name": "Key Event",
+  "description": "Key Event Description",
+  "priority": 0.5
+}
+
+###
+
+GET {{HostAddress}}/consume/EVENTS
+Accept: application/json
+
+###

playground/nats/Nats.ApiService/Program.cs

@@ -1,17 +1,82 @@
+using System.Text.Json.Serialization;
 using NATS.Client.Core;
+using NATS.Client.JetStream;
+using NATS.Client.JetStream.Models;
 
 var builder = WebApplication.CreateBuilder(args);
 
+// Add services to the container.
+// Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
+builder.Services.AddEndpointsApiExplorer();
+builder.Services.AddSwaggerGen();
+
 builder.AddServiceDefaults();
 
-builder.AddNats("nats");
+builder.AddNats("nats", opts =>
+{
+    var jsonRegistry = new NatsJsonContextSerializerRegistry(AppJsonContext.Default);
+    return opts with { SerializerRegistry = jsonRegistry };
+});
+
+builder.Services.AddSingleton<INatsJSContext>(static provider =>
+{
+    return new NatsJSContextFactory().CreateContext(provider.GetService<INatsConnection>()!);
+});
 
 var app = builder.Build();
 
-app.MapGet("/", async (INatsConnection nats) => $"""
-                                                 Hello NATS!
-                                                 rtt: {await nats.PingAsync()}
-                                                 {nats.ServerInfo}
-                                                 """);
+// Configure the HTTP request pipeline.
+if (app.Environment.IsDevelopment())
+{
+    app.UseSwagger();
+    app.UseSwaggerUI();
+}
+
+app.MapPost("/stream", async (StreamConfig config, INatsJSContext jetStream) =>
+{
+    var stream = await jetStream.CreateStreamAsync(config);
+    var name = stream.Info.Config.Name;
+    return Results.Created($"/stream/{name}", name);
+});
+
+app.MapGet("/stream/{name}", async (string name, INatsJSContext jetStream) =>
+{
+    var stream = await jetStream.GetStreamAsync(name);
+    return Results.Ok(stream.Info);
+});
+
+app.MapPost("/publish/", async (AppEvent @event, INatsJSContext jetStream) =>
+{
+    var ack = await jetStream.PublishAsync(@event.Subject, @event);
+    ack.EnsureSuccess();
+    return Results.Created();
+});
+
+app.MapGet("/consume/{name}", async (string name, INatsJSContext jetStream) =>
+{
+    var stream = await jetStream.GetStreamAsync(name);
+    var consumer = await stream.CreateOrderedConsumerAsync();
+
+    var events = new List<AppEvent>();
+    await foreach(var msg in consumer.ConsumeAsync<AppEvent>())
+    {
+        events.Add(msg.Data!);
+
+        if (msg.Metadata?.NumPending == 0)
+        {
+            break;
+        }
+    }
+
+    return Results.Ok(events);
+});
 
 app.Run();
+
+public record AppEvent(string Subject, string Name, string Description, decimal Priority);
+
+[JsonSerializable(typeof(AppEvent))]
+[JsonSourceGenerationOptions(PropertyNameCaseInsensitive = true)]
+public partial class AppJsonContext : JsonSerializerContext
+{
+}

playground/nats/Nats.AppHosting/Program.cs

@@ -2,7 +2,7 @@
 
 var builder = DistributedApplication.CreateBuilder(args);
 
-var nats = builder.AddNats("nats");
+var nats = builder.AddNats("nats", enableJetStream: true);
 
 builder.AddProject<Projects.Nats_ApiService>("apiservice")
     .WithReference(nats);

src/Components/Aspire.Nats/NatsBuilderExtensions.cs

@@ -1,20 +1,29 @@
-// Licensed to the .NET Foundation under one or more agreements.
+// Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
 using Microsoft.Extensions.Configuration;
+using NATS.Client.Core;
 using NATS.Client.Hosting;
 
 namespace Microsoft.Extensions.Hosting;
 
 public static class NatsBuilderExtensions
 {
-    public static void AddNats(this IHostApplicationBuilder builder, string connectionName)
+    public static void AddNats(this IHostApplicationBuilder builder, string connectionName, Func<NatsOpts, NatsOpts>? configureOptions = null)
     {
         ArgumentNullException.ThrowIfNull(builder);
 
         if (builder.Configuration.GetConnectionString(connectionName) is string connectionString)
         {
-            builder.Services.AddNats(configureOpts: opts => opts with { Url = connectionString });
+            builder.Services.AddNats(configureOpts: opts =>
+            {
+                if (configureOptions != null)
+                {
+                    opts = configureOptions(opts);
+                }
+
+                return opts with { Url = connectionString };
+            });
         }
         else
         {