Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,14 @@ namespace CookieCrumble.Resources;

public class MongoResource : ContainerResource<MongoDbContainer>
{
private MongoClient? _client;
private readonly Lazy<MongoClient> _client;

public IMongoClient Client => _client ??= new MongoClient(ConnectionString);
public MongoResource()
{
_client = new(() => new MongoClient(ConnectionString));
}

public IMongoClient Client => _client.Value;

public string ConnectionString => Container.GetConnectionString();

Expand All @@ -29,4 +34,14 @@ protected override MongoDbContainer Build()
=> Configure(new MongoDbBuilder("mongo:6.0")).Build();

protected virtual MongoDbBuilder Configure(MongoDbBuilder builder) => builder;

protected override ValueTask DisposeAsyncCore()
{
if (_client.IsValueCreated)
{
_client.Value.Dispose();
}

return ValueTask.CompletedTask;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@
</PropertyGroup>

<ItemGroup>
<PackageReference Include="NATS.Client.Core" />
<PackageReference Include="NATS.Client.JetStream" />
<PackageReference Include="Testcontainers.Nats" />
</ItemGroup>

Expand Down
39 changes: 39 additions & 0 deletions src/CookieCrumble/src/CookieCrumble.Resources.Nats/NatsResource.cs
Original file line number Diff line number Diff line change
@@ -1,13 +1,52 @@
using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
using Testcontainers.Nats;

namespace CookieCrumble.Resources;

public class NatsResource : ContainerResource<NatsContainer>
{
private readonly Lazy<NatsConnection> _connection;

public NatsResource()
{
_connection = new(() => new NatsConnection(new NatsOpts { Url = NatsConnectionString }));
}

public string NatsConnectionString => Container.GetConnectionString();

/// <summary>
/// Creates a uniquely named JetStream stream over <paramref name="subjects"/>.
/// Disposing the returned stream deletes it.
/// </summary>
public async Task<NatsStream> CreateStreamAsync(
IReadOnlyList<string> subjects,
CancellationToken cancellationToken)
{
var name = $"S{Guid.NewGuid():N}";
var jetStream = new NatsJSContext(_connection.Value);
await jetStream.CreateStreamAsync(
new StreamConfig
{
Name = name,
Subjects = [.. subjects]
},
cancellationToken);

return new NatsStream(jetStream, name);
}

protected override NatsContainer Build()
=> Configure(new NatsBuilder("nats:2.10-alpine")).Build();

protected virtual NatsBuilder Configure(NatsBuilder builder) => builder;

protected override async ValueTask DisposeAsyncCore()
{
if (_connection.IsValueCreated)
{
await _connection.Value.DisposeAsync();
}
}
}
22 changes: 22 additions & 0 deletions src/CookieCrumble/src/CookieCrumble.Resources.Nats/NatsStream.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
using NATS.Client.JetStream;

namespace CookieCrumble.Resources;

/// <summary>
/// A JetStream stream created by <see cref="NatsResource.CreateStreamAsync"/>.
/// Disposing it deletes the stream.
/// </summary>
public sealed class NatsStream : IAsyncDisposable
{
private readonly NatsJSContext _jetStream;

internal NatsStream(NatsJSContext jetStream, string name)
{
_jetStream = jetStream;
Name = name;
}

public string Name { get; }

public async ValueTask DisposeAsync() => await _jetStream.DeleteStreamAsync(Name);
}
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,30 @@ namespace CookieCrumble.Resources;

public class RedisResource : ContainerResource<RedisContainer>
{
private readonly Lazy<ConnectionMultiplexer> _connection;

public RedisResource()
{
_connection = new(() => ConnectionMultiplexer.Connect(ConnectionString));
}

public string ConnectionString => Container.GetConnectionString();

public ConnectionMultiplexer GetConnection() => ConnectionMultiplexer.Connect(ConnectionString);
/// <summary>
/// Gets the connection shared by every caller of this resource. The resource owns it
/// and disposes it together with the container.
/// </summary>
public ConnectionMultiplexer GetConnection() => _connection.Value;

protected override RedisContainer Build() => Configure(new RedisBuilder("redis:7.0")).Build();

protected virtual RedisBuilder Configure(RedisBuilder builder) => builder;

protected override async ValueTask DisposeAsyncCore()
{
if (_connection.IsValueCreated)
{
await _connection.Value.DisposeAsync();
}
}
}
17 changes: 15 additions & 2 deletions src/CookieCrumble/src/CookieCrumble.Resources/ContainerResource.cs
Original file line number Diff line number Diff line change
Expand Up @@ -25,11 +25,24 @@ public async ValueTask InitializeAsync()

public async ValueTask DisposeAsync()
{
if (_container is not null)
try
{
await _container.DisposeAsync();
await DisposeAsyncCore();
}
finally
{
if (_container is not null)
{
await _container.DisposeAsync();
}
}
}

protected abstract TContainer Build();

/// <summary>
/// Releases the clients this resource created on top of the container.
/// Runs before the container itself is disposed.
/// </summary>
protected virtual ValueTask DisposeAsyncCore() => ValueTask.CompletedTask;
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
using CookieCrumble.Resources;

[assembly: AssemblyFixture(typeof(PostgreSqlResource))]
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,7 @@
namespace HotChocolate.Subscriptions.Postgres;

public class PostgresChannelTests
: IClassFixture<PostgreSqlResource>
, IAsyncLifetime
: IAsyncLifetime
{
private readonly PostgreSqlResource _resource;
private readonly string _dbName = $"DB_{Guid.NewGuid():N}";
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,7 @@
namespace HotChocolate.Subscriptions.Postgres;

public class PostgresChannelWriterTests
: IClassFixture<PostgreSqlResource>
, IAsyncLifetime
: IAsyncLifetime
{
private readonly PostgreSqlResource _resource;
private readonly string _dbName = $"DB_{Guid.NewGuid():N}";
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ namespace HotChocolate.Subscriptions.Postgres;

public class PostgresPubSubIntegrationTests
: SubscriptionIntegrationTestBase
, IClassFixture<PostgreSqlResource>
, IAsyncLifetime
{
private readonly PostgreSqlResource _resource;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

namespace HotChocolate.Subscriptions.Postgres;

public class ResilientNpgsqlConnectionTests : IClassFixture<PostgreSqlResource>
public class ResilientNpgsqlConnectionTests
{
private readonly PostgreSqlResource _resource;
private readonly SubscriptionTestDiagnostics _events;
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
using CookieCrumble.Resources;

[assembly: AssemblyFixture(typeof(RedisResource))]
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
using System.Security.Cryptography;
using CookieCrumble.Resources;
using HotChocolate.Execution;
using HotChocolate.Execution.Configuration;
Expand All @@ -6,8 +7,12 @@

namespace HotChocolate.Subscriptions.Redis;

public class RedisIntegrationTests : SubscriptionIntegrationTestBase, IClassFixture<RedisResource>
public class RedisIntegrationTests : SubscriptionIntegrationTestBase
{
// Redis channel names are the hex MD5 of the topic name.
private static readonly string s_onMessageChannel =
Convert.ToHexString(MD5.HashData("OnMessage"u8));

private readonly RedisResource _redisResource;

public RedisIntegrationTests(RedisResource redisResource, ITestOutputHelper output)
Expand Down Expand Up @@ -97,7 +102,10 @@ public static int WaitForChannelRemoval(

private async Task<RedisResult[]> GetActiveChannelsAsync()
{
return (RedisResult[])(await _redisResource.GetConnection().GetDatabase().ExecuteAsync("PUBSUB", "CHANNELS"))!;
var database = _redisResource.GetConnection().GetDatabase();
var channels = await database.ExecuteAsync("PUBSUB", "CHANNELS", s_onMessageChannel);

return (RedisResult[])channels!;
}

protected override void ConfigurePubSub(IRequestExecutorBuilder graphqlBuilder)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
namespace HotChocolate.Subscriptions.Redis;

public class RedisTopicPrefixIntegrationTests(RedisResource redisResource, ITestOutputHelper output)
: SubscriptionIntegrationTestBase(output), IClassFixture<RedisResource>
: SubscriptionIntegrationTestBase(output)
{
private const string TopicPrefix = "prefix:";

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
using Microsoft.Extensions.DependencyInjection;
using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
using OperationRequest = HotChocolate.Transport.OperationRequest;
using OperationResult = HotChocolate.Transport.OperationResult;

Expand All @@ -23,7 +22,8 @@ namespace HotChocolate.Fusion;
/// errors; the subscription must surface that as a per-event error result (subgraph error plus
/// standard non-null propagation) and stay alive, not silently close.
/// </summary>
public class EventStreamEnrichmentErrorOverHttpTests : FusionTestBase
[Collection(NatsCollectionFixture.DefinitionName)]
public class EventStreamEnrichmentErrorOverHttpTests(NatsResource nats) : FusionTestBase
{
private const string BrokerName = "nats";
private const string Topic = "onCreateReview";
Expand All @@ -33,12 +33,9 @@ public class EventStreamEnrichmentErrorOverHttpTests : FusionTestBase
public async Task Subscribe_Should_ReturnErrorAndStayAlive_When_EnrichmentLookupRejectsEventId()
{
// arrange
await using var nats = new NatsResource();
await nats.InitializeAsync();
var stream = "S" + Guid.NewGuid().ToString("N");
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(60));

await CreateStreamAsync(nats.NatsConnectionString, stream, Topic, cts.Token);
await using var stream = await nats.CreateStreamAsync([Topic], cts.Token);

using var events = CreateSourceSchema(
"EVENTS",
Expand All @@ -55,7 +52,7 @@ public async Task Subscribe_Should_ReturnErrorAndStayAlive_When_EnrichmentLookup
o =>
{
o.Url = nats.NatsConnectionString;
o.JetStream = new NatsJetStreamOptions { Stream = stream };
o.JetStream = new NatsJetStreamOptions { Stream = stream.Name };
}),
configureGatewayBuilder: b => b.ModifyRequestOptions(o => o.AllowOperationPlanRequests = false));

Expand Down Expand Up @@ -85,7 +82,11 @@ public async Task Subscribe_Should_ReturnErrorAndStayAlive_When_EnrichmentLookup
// The first event carries the raw database key (not a valid global id), so the node lookup
// that fetches `body` fails. The second event carries the encoded global id and resolves.
using var response = await client.PostAsync(request, new Uri(GatewayUrl), cts.Token);
await WaitForConsumerAsync(nats.NatsConnectionString, stream, expectedCount: 1, cts.Token);
await WaitForConsumerAsync(
nats.NatsConnectionString,
stream.Name,
expectedCount: 1,
cts.Token);
await PublishAsync(nats.NatsConnectionString, Topic, """{"review":{"id":1}}""", cts.Token);
await PublishAsync(
nats.NatsConnectionString,
Expand Down Expand Up @@ -149,13 +150,6 @@ await PublishAsync(
]);
}

private static async Task CreateStreamAsync(string url, string stream, string subject, CancellationToken ct)
{
await using var connection = new NatsConnection(new NatsOpts { Url = url });
var js = new NatsJSContext(connection);
await js.CreateStreamAsync(new StreamConfig { Name = stream, Subjects = [subject] }, ct);
}

private static async Task PublishAsync(string url, string subject, string body, CancellationToken ct)
{
await using var connection = new NatsConnection(new NatsOpts { Url = url });
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
using CookieCrumble.Resources;

namespace HotChocolate.Fusion;

[CollectionDefinition(DefinitionName)]
public class NatsCollectionFixture : ICollectionFixture<NatsResource>
{
internal const string DefinitionName = "Nats";
}
Loading
Loading