diff --git a/Aspire.sln b/Aspire.sln
index 12f3719255e..e1532ebfef7 100644
--- a/Aspire.sln
+++ b/Aspire.sln
@@ -550,6 +550,7 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Aspire.Hosting.Keycloak.Tes
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Aspire.Keycloak.Authentication.Tests", "tests\Aspire.Keycloak.Authentication.Tests\Aspire.Keycloak.Authentication.Tests.csproj", "{48FF09E9-7D33-4A3F-9FF2-4C43A219C7B7}"
EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Aspire.Hosting.Nats.Tests", "tests\Aspire.Hosting.Nats.Tests\Aspire.Hosting.Nats.Tests.csproj", "{F492357C-682E-4CBB-A374-1A124B3976A3}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Aspire.Hosting.Azure.Tests", "tests\Aspire.Hosting.Azure.Tests\Aspire.Hosting.Azure.Tests.csproj", "{8691F993-7B19-496E-B8E1-EF1199ACF2E1}"
EndProject
Global
@@ -1458,6 +1459,10 @@ Global
{48FF09E9-7D33-4A3F-9FF2-4C43A219C7B7}.Debug|Any CPU.Build.0 = Debug|Any CPU
{48FF09E9-7D33-4A3F-9FF2-4C43A219C7B7}.Release|Any CPU.ActiveCfg = Release|Any CPU
{48FF09E9-7D33-4A3F-9FF2-4C43A219C7B7}.Release|Any CPU.Build.0 = Release|Any CPU
+ {F492357C-682E-4CBB-A374-1A124B3976A3}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {F492357C-682E-4CBB-A374-1A124B3976A3}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {F492357C-682E-4CBB-A374-1A124B3976A3}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {F492357C-682E-4CBB-A374-1A124B3976A3}.Release|Any CPU.Build.0 = Release|Any CPU
{8691F993-7B19-496E-B8E1-EF1199ACF2E1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{8691F993-7B19-496E-B8E1-EF1199ACF2E1}.Debug|Any CPU.Build.0 = Debug|Any CPU
{8691F993-7B19-496E-B8E1-EF1199ACF2E1}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -1731,6 +1736,7 @@ Global
{C556D61C-7E11-43EC-9098-C8D170FEA905} = {EBC55A17-B0D6-4E0A-9DC2-7D264E96F631}
{5867BAF2-FEF0-4661-BFDE-9ADCDC2921CD} = {830A89EC-4029-4753-B25A-068BAE37DEC7}
{48FF09E9-7D33-4A3F-9FF2-4C43A219C7B7} = {C424395C-1235-41A4-BF55-07880A04368C}
+ {F492357C-682E-4CBB-A374-1A124B3976A3} = {830A89EC-4029-4753-B25A-068BAE37DEC7}
{8691F993-7B19-496E-B8E1-EF1199ACF2E1} = {830A89EC-4029-4753-B25A-068BAE37DEC7}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
diff --git a/tests/Aspire.Hosting.Tests/Nats/AddNatsTests.cs b/tests/Aspire.Hosting.Nats.Tests/AddNatsTests.cs
similarity index 98%
rename from tests/Aspire.Hosting.Tests/Nats/AddNatsTests.cs
rename to tests/Aspire.Hosting.Nats.Tests/AddNatsTests.cs
index aa0bbaca4b0..3a9d71f0ba8 100644
--- a/tests/Aspire.Hosting.Tests/Nats/AddNatsTests.cs
+++ b/tests/Aspire.Hosting.Nats.Tests/AddNatsTests.cs
@@ -1,13 +1,13 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
-using Aspire.Hosting.Nats;
using Aspire.Hosting.Utils;
using System.Net.Sockets;
using Microsoft.Extensions.DependencyInjection;
using Xunit;
+using Aspire.Hosting.ApplicationModel;
-namespace Aspire.Hosting.Tests.Nats;
+namespace Aspire.Hosting.Nats.Tests;
public class AddNatsTests
{
diff --git a/tests/Aspire.Hosting.Nats.Tests/AppEvent.cs b/tests/Aspire.Hosting.Nats.Tests/AppEvent.cs
new file mode 100644
index 00000000000..e022f64c0c8
--- /dev/null
+++ b/tests/Aspire.Hosting.Nats.Tests/AppEvent.cs
@@ -0,0 +1,14 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Text.Json.Serialization;
+
+namespace Aspire.Hosting.Nats.Tests;
+
+public record AppEvent(string Subject, string Name, string Description, decimal Priority);
+
+[JsonSerializable(typeof(AppEvent))]
+[JsonSourceGenerationOptions(PropertyNameCaseInsensitive = true)]
+public partial class AppJsonContext : JsonSerializerContext
+{
+}
diff --git a/tests/Aspire.Hosting.Nats.Tests/Aspire.Hosting.Nats.Tests.csproj b/tests/Aspire.Hosting.Nats.Tests/Aspire.Hosting.Nats.Tests.csproj
new file mode 100644
index 00000000000..40f379f5543
--- /dev/null
+++ b/tests/Aspire.Hosting.Nats.Tests/Aspire.Hosting.Nats.Tests.csproj
@@ -0,0 +1,19 @@
+
+
+
+ $(NetCurrent)
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/tests/Aspire.Hosting.Nats.Tests/NatsFunctionalTests.cs b/tests/Aspire.Hosting.Nats.Tests/NatsFunctionalTests.cs
new file mode 100644
index 00000000000..3204715a209
--- /dev/null
+++ b/tests/Aspire.Hosting.Nats.Tests/NatsFunctionalTests.cs
@@ -0,0 +1,247 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using Aspire.Components.Common.Tests;
+using Aspire.Hosting.Utils;
+using Xunit;
+using Xunit.Abstractions;
+using Microsoft.Extensions.Logging;
+using Polly;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.DependencyInjection;
+using NATS.Client.Core;
+using NATS.Client.JetStream;
+using NATS.Client.JetStream.Models;
+namespace Aspire.Hosting.Nats.Tests;
+
+public class NatsFunctionalTests(ITestOutputHelper testOutputHelper)
+{
+ private const string StreamName = "test-stream";
+ private const string SubjectName = "test-subject";
+
+ [Fact]
+ [RequiresDocker]
+ public async Task VerifyNatsResource()
+ {
+ var cts = new CancellationTokenSource(TimeSpan.FromMinutes(5));
+ var pipeline = new ResiliencePipelineBuilder()
+ .AddRetry(new() { MaxRetryAttempts = 10, Delay = TimeSpan.FromSeconds(1), ShouldHandle = new PredicateBuilder().Handle() })
+ .Build();
+
+ var builder = CreateDistributedApplicationBuilder();
+
+ var nats = builder.AddNats("nats")
+ .WithJetStream();
+
+ using var app = builder.Build();
+
+ await app.StartAsync();
+
+ var hb = Host.CreateApplicationBuilder();
+
+ hb.Configuration[$"ConnectionStrings:{nats.Resource.Name}"] = await nats.Resource.ConnectionStringExpression.GetValueAsync(default);
+
+ hb.AddNatsClient("nats", configureOptions: opts =>
+ {
+ var jsonRegistry = new NatsJsonContextSerializerRegistry(AppJsonContext.Default);
+ return opts with { SerializerRegistry = jsonRegistry };
+ });
+
+ hb.AddNatsJetStream();
+
+ using var host = hb.Build();
+
+ await host.StartAsync();
+
+ await pipeline.ExecuteAsync(async token =>
+ {
+ var jetStream = host.Services.GetRequiredService();
+
+ await CreateTestData(jetStream, token);
+ await ConsumeTestData(jetStream, token);
+ }, cts.Token);
+ }
+
+ [Theory]
+ [InlineData(true)]
+ [InlineData(false)]
+ [RequiresDocker]
+ public async Task WithDataShouldPersistStateBetweenUsages(bool useVolume)
+ {
+ var cts = new CancellationTokenSource(TimeSpan.FromMinutes(3));
+ var pipeline = new ResiliencePipelineBuilder()
+ .AddRetry(new() { MaxRetryAttempts = 10, Delay = TimeSpan.FromSeconds(1), ShouldHandle = new PredicateBuilder().Handle() })
+ .Build();
+ string? volumeName = null;
+ string? bindMountPath = null;
+
+ try
+ {
+ var builder1 = CreateDistributedApplicationBuilder();
+ var nats1 = builder1.AddNats("nats")
+ .WithJetStream();
+
+ if (useVolume)
+ {
+ // Use a deterministic volume name to prevent them from exhausting the machines if deletion fails
+ volumeName = VolumeNameGenerator.CreateVolumeName(nats1, nameof(WithDataShouldPersistStateBetweenUsages));
+
+ // if the volume already exists (because of a crashing previous run), try to delete it
+ DockerUtils.AttemptDeleteDockerVolume(volumeName);
+ nats1.WithDataVolume(volumeName);
+ }
+ else
+ {
+ bindMountPath = Directory.CreateTempSubdirectory().FullName;
+ nats1.WithDataBindMount(bindMountPath);
+ }
+
+ using (var app = builder1.Build())
+ {
+ await app.StartAsync();
+ try
+ {
+ var hb = Host.CreateApplicationBuilder();
+
+ hb.Configuration[$"ConnectionStrings:{nats1.Resource.Name}"] = await nats1.Resource.ConnectionStringExpression.GetValueAsync(default);
+
+ hb.AddNatsClient("nats", configureOptions: opts =>
+ {
+ var jsonRegistry = new NatsJsonContextSerializerRegistry(AppJsonContext.Default);
+ return opts with { SerializerRegistry = jsonRegistry };
+ });
+
+ hb.AddNatsJetStream();
+
+ using (var host = hb.Build())
+ {
+ await host.StartAsync();
+
+ await pipeline.ExecuteAsync(async token =>
+ {
+ var jetStream = host.Services.GetRequiredService();
+ await CreateTestData(jetStream, token);
+ await ConsumeTestData(jetStream, token);
+
+ }, cts.Token);
+ }
+ }
+ finally
+ {
+ // Stops the container, or the Volume/mount would still be in use
+ await app.StopAsync();
+ }
+ }
+
+ var builder2 = CreateDistributedApplicationBuilder();
+ var nats2 = builder2.AddNats("nats")
+ .WithJetStream();
+
+ if (useVolume)
+ {
+ nats2.WithDataVolume(volumeName);
+ }
+ else
+ {
+ nats2.WithDataBindMount(bindMountPath!);
+ }
+
+ using (var app = builder2.Build())
+ {
+ await app.StartAsync();
+ try
+ {
+ var hb = Host.CreateApplicationBuilder();
+
+ hb.Configuration[$"ConnectionStrings:{nats2.Resource.Name}"] = await nats2.Resource.ConnectionStringExpression.GetValueAsync(default);
+ hb.AddNatsClient("nats", configureOptions: opts =>
+ {
+ var jsonRegistry = new NatsJsonContextSerializerRegistry(AppJsonContext.Default);
+ return opts with { SerializerRegistry = jsonRegistry };
+ });
+
+ hb.AddNatsJetStream();
+
+ using (var host = hb.Build())
+ {
+ await host.StartAsync();
+
+ await pipeline.ExecuteAsync(async token =>
+ {
+ var jetStream = host.Services.GetRequiredService();
+ await ConsumeTestData(jetStream, token);
+ });
+ }
+ }
+ finally
+ {
+ // Stops the container, or the Volume/mount would still be in use
+ await app.StopAsync();
+ }
+ }
+ }
+ finally
+ {
+ if (volumeName is not null)
+ {
+ DockerUtils.AttemptDeleteDockerVolume(volumeName);
+ }
+
+ if (bindMountPath is not null)
+ {
+ try
+ {
+ Directory.Delete(bindMountPath, recursive: true);
+ }
+ catch
+ {
+ // Don't fail test if we can't clean the temporary folder
+ }
+ }
+ }
+ }
+
+ private static async Task ConsumeTestData(INatsJSContext jetStream, CancellationToken token)
+ {
+ var stream = await jetStream.GetStreamAsync(StreamName, cancellationToken: token);
+ var consumer = await stream.CreateOrderedConsumerAsync(cancellationToken: token);
+
+ var events = new List();
+ await foreach (var msg in consumer.ConsumeAsync(cancellationToken: token))
+ {
+ events.Add(msg.Data!);
+ await msg.AckAsync(cancellationToken: token);
+ if (msg.Metadata?.NumPending == 0)
+ {
+ break;
+ }
+ }
+
+ for (var i = 0; i < 10; i++)
+ {
+ var @event = events[i];
+ Assert.Equal($"test-event-{i}", @event.Name);
+ Assert.Equal($"test-event-description-{i}", @event.Description);
+ }
+ }
+
+ private static async Task CreateTestData(INatsJSContext jetStream, CancellationToken token)
+ {
+ var stream = await jetStream.CreateStreamAsync(new StreamConfig(StreamName, [SubjectName]), cancellationToken: token);
+ Assert.Equal(StreamName, stream.Info.Config.Name);
+
+ for (var i = 0; i < 10; i++)
+ {
+ var appEvent = new AppEvent(SubjectName, $"test-event-{i}", $"test-event-description-{i}", i);
+ var ack = await jetStream.PublishAsync(appEvent.Subject, appEvent, cancellationToken: token);
+ ack.EnsureSuccess();
+ }
+ }
+
+ private TestDistributedApplicationBuilder CreateDistributedApplicationBuilder()
+ {
+ var builder = TestDistributedApplicationBuilder.CreateWithTestContainerRegistry();
+ builder.Services.AddXunitLogging(testOutputHelper);
+ return builder;
+ }
+}
diff --git a/tests/Aspire.Hosting.Tests/Aspire.Hosting.Tests.csproj b/tests/Aspire.Hosting.Tests/Aspire.Hosting.Tests.csproj
index aae8ef5214d..80fee8b0b3f 100644
--- a/tests/Aspire.Hosting.Tests/Aspire.Hosting.Tests.csproj
+++ b/tests/Aspire.Hosting.Tests/Aspire.Hosting.Tests.csproj
@@ -13,7 +13,6 @@
-
@@ -25,7 +24,6 @@
-