From 1f11a2503cc386ce4b79a434c705dcf03d98affd Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Fri, 31 Jul 2026 16:59:36 +0200 Subject: [PATCH 1/5] Code cleanup --- .../Bookings/CommandApiWithCustomResult.cs | 2 +- .../ConvertEventToMessage.cs | 2 +- .../Benchmarks/ChannelBatchingBenchmarks.cs | 5 +- .../EventUsageAnalyzer.cs | 10 +- .../ConsumeContextConverterGenerator.cs | 2 +- .../AggregateService/CommandService.cs | 2 +- .../Diagnostics/CommandServiceMetrics.cs | 2 +- .../Persistence/WriterExtensions.cs | 4 +- .../EventuousDiagnostics.cs | 2 +- src/Core/src/Eventuous.Domain/Aggregate.cs | 2 +- .../Diagnostics/PersistenceMetrics.cs | 2 +- .../Diagnostics/Tracing/BaseTracer.cs | 6 +- .../Diagnostics/Tracing/TracedEventReader.cs | 4 +- .../Diagnostics/Tracing/TracedEventWriter.cs | 8 +- .../EventStore/StoreFunctions.cs | 185 +++++++++--------- .../src/Eventuous.Producers/BaseProducer.cs | 2 +- src/Core/src/Eventuous.Shared/Tools/Ensure.cs | 4 - .../Eventuous.Shared/Tools/TaskExtensions.cs | 10 +- .../src/Eventuous.Shared/Tools/TaskRunner.cs | 4 - .../Eventuous.Shared/TypeMap/TypeMapper.cs | 2 +- .../Channels/ChannelWorkerBase.cs | 6 +- .../Diagnostics/SubscriptionMetrics.cs | 2 +- .../Filters/PartitioningFilter.cs | 2 +- .../Filters/TracingFilter.cs | 2 +- .../Registrations/SubscriptionBuilder.cs | 2 +- .../SubscriptionHostedService.cs | 3 - .../ServiceTestBase.Amendments.cs | 2 +- .../Fixtures/Helpers.cs | 2 +- .../Fixtures/StoreFixtureBase.cs | 4 - .../Store/Read.cs | 16 +- .../Analyzed.cs | 6 - .../Analyzer_Ev001_Tests.cs | 6 +- .../Fixtures/TestEventHandler.cs | 2 +- .../SubscribeToStream.cs | 2 +- .../SubscriptionDropBase.cs | 2 +- .../SubscriptionMeasureBase.cs | 1 - ...heckpointCommitHandlerBackpressureTests.cs | 40 ++-- ...CheckpointCommitHandlerConcurrencyTests.cs | 2 +- .../ResubscribeOnHandlerFailureTests.cs | 2 +- .../Eventuous.Tests/Fixtures/IdGenerator.cs | 2 +- .../LoggingEventListener.cs | 2 +- .../Fakes/TestExporter.cs | 4 +- .../src/ElasticPlayground/Generator.cs | 2 +- .../Producers/ElasticProducer.cs | 2 +- .../Store/ElasticEventStore.cs | 22 +-- .../Eventuous.Spyglass/SpyglassRegistry.cs | 44 ++--- .../CompilationHelper.cs | 2 +- .../Http/CommandMappingRegistry.cs | 4 +- .../AnalyzerTestShared.cs | 2 +- .../DiscoveredCommandsTests.cs | 4 +- .../KurrentDBEventStore.cs | 12 +- .../Subscriptions/AllStreamSubscription.cs | 10 +- .../Subscriptions/CheckpointReachedTests.cs | 2 +- .../PersistentSubscriptionFailureTests.cs | 2 +- .../ResolvedLinkCheckpointTests.cs | 2 +- .../Projections/PostgresProjector.cs | 4 +- .../Store/TieredStoreTests.cs | 1 + .../Subscriptions/GapIgnoreTest.cs | 4 +- .../Subscriptions/TombstonesCreationTest.cs | 6 +- src/Redis/src/Eventuous.Redis/RedisStore.cs | 2 +- .../RedisAllStreamSubscription.cs | 2 +- .../Subscriptions/RedisStreamSubscription.cs | 6 +- .../Eventuous.Tests.Redis/Store/Helpers.cs | 2 +- .../Subscriptions/SubscribeToAll.cs | 2 +- .../Subscriptions/SubscribeToStream.cs | 2 +- .../SignalRSubscriptionClient.cs | 2 +- .../TypedStreamSubscriptionTests.cs | 18 +- .../Store/StoreFixture.cs | 2 +- .../Subscriptions/SubscribeTests.cs | 6 +- .../Eventuous.Testing/InMemoryEventStore.cs | 1 + .../TestEventListener.cs | 2 +- test/Eventuous.TestHelpers/TestHelper.cs | 21 +- 72 files changed, 261 insertions(+), 304 deletions(-) diff --git a/samples/kurrentdb/Bookings/HttpApi/Bookings/CommandApiWithCustomResult.cs b/samples/kurrentdb/Bookings/HttpApi/Bookings/CommandApiWithCustomResult.cs index 0728946d1..b52ea0140 100644 --- a/samples/kurrentdb/Bookings/HttpApi/Bookings/CommandApiWithCustomResult.cs +++ b/samples/kurrentdb/Bookings/HttpApi/Bookings/CommandApiWithCustomResult.cs @@ -51,7 +51,7 @@ static BadRequestObjectResult MapValidationExceptionAsValidationProblemDetails(R var groupFailures = exception.Errors.GroupBy(v => v.PropertyName); foreach (var groupFailure in groupFailures) { - problemDetails.Errors.Add(groupFailure.Key, groupFailure.Select(s => s.ErrorMessage).ToArray()); + problemDetails.Errors.Add(groupFailure.Key, [.. groupFailure.Select(s => s.ErrorMessage)]); } return new(problemDetails); diff --git a/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/ConvertEventToMessage.cs b/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/ConvertEventToMessage.cs index f30a0376c..59c525360 100644 --- a/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/ConvertEventToMessage.cs +++ b/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/ConvertEventToMessage.cs @@ -75,7 +75,7 @@ public async Task CorrelationId() { [Test] public async Task ScheduledEnqueueTime() => - await Assert.That(_message.ScheduledEnqueueTime).IsEqualTo(new DateTimeOffset(2026, 3, 23, 16, 31, 0, TimeSpan.Zero)); + await Assert.That(_message.ScheduledEnqueueTime).IsEqualTo(new(2026, 3, 23, 16, 31, 0, TimeSpan.Zero)); [Test] [Arguments("MessageId")] diff --git a/src/Benchmarks/Benchmarks/ChannelBatchingBenchmarks.cs b/src/Benchmarks/Benchmarks/ChannelBatchingBenchmarks.cs index 182ffde04..3ffe79da7 100644 --- a/src/Benchmarks/Benchmarks/ChannelBatchingBenchmarks.cs +++ b/src/Benchmarks/Benchmarks/ChannelBatchingBenchmarks.cs @@ -1,5 +1,6 @@ using BenchmarkDotNet.Attributes; using System.Buffers; +using System.Runtime.InteropServices; namespace Benchmarks; @@ -26,12 +27,12 @@ public void Setup() { [Benchmark(Baseline = true, Description = "Current: List.ToArray()")] public int[] CurrentApproach_ToArray() { - return _buffer.ToArray(); + return [.. _buffer]; } [Benchmark(Description = "Alternative 1: CollectionsMarshal.AsSpan()")] public ReadOnlySpan Alternative1_CollectionsMarshalAsSpan() { - return System.Runtime.InteropServices.CollectionsMarshal.AsSpan(_buffer); + return CollectionsMarshal.AsSpan(_buffer); } [Benchmark(Description = "Alternative 2: ArrayPool rent/copy")] diff --git a/src/Core/gen/Eventuous.Shared.Generators/EventUsageAnalyzer.cs b/src/Core/gen/Eventuous.Shared.Generators/EventUsageAnalyzer.cs index 1f64a184e..069f17d4a 100644 --- a/src/Core/gen/Eventuous.Shared.Generators/EventUsageAnalyzer.cs +++ b/src/Core/gen/Eventuous.Shared.Generators/EventUsageAnalyzer.cs @@ -141,15 +141,7 @@ static void AnalyzeInvocation(OperationAnalysisContext ctx, KnownTypeSymbols kno return; } // Case 1c: State.On(...) handler registrations - case { Name: "On", TypeArguments.Length: 1 } when IsState(method.ContainingType, knownTypes): { - var eventType = method.TypeArguments[0]; - - if (IsConcreteEvent(eventType) && !HasEventTypeAttribute(eventType, knownTypes) && !IsExplicitlyRegistered(eventType, ctx, knownTypes)) { - ctx.ReportDiagnostic(Diagnostic.Create(MissingEventTypeAttribute, inv.Syntax.GetLocation(), eventType.ToDisplayString())); - } - - return; - } + case { Name: "On", TypeArguments.Length: 1 } when IsState(method.ContainingType, knownTypes): // Case 1d: EventHandler.On(...) handler registrations case { Name: "On", TypeArguments.Length: 1 } when IsEventHandler(method.ContainingType, knownTypes): { var eventType = method.TypeArguments[0]; diff --git a/src/Core/gen/Eventuous.Subscriptions.Generators/ConsumeContextConverterGenerator.cs b/src/Core/gen/Eventuous.Subscriptions.Generators/ConsumeContextConverterGenerator.cs index 88416786a..ff7f37c03 100644 --- a/src/Core/gen/Eventuous.Subscriptions.Generators/ConsumeContextConverterGenerator.cs +++ b/src/Core/gen/Eventuous.Subscriptions.Generators/ConsumeContextConverterGenerator.cs @@ -287,7 +287,7 @@ void ProcessNamespace(INamespaceSymbol ns, bool isReferenced) { } static bool IsPublicType(INamedTypeSymbol type) { - for (var t = (ITypeSymbol)type; t != null; t = t.ContainingType) { + for (ITypeSymbol? t = type; t != null; t = t.ContainingType) { if (t.DeclaredAccessibility != Accessibility.Public) return false; } return true; diff --git a/src/Core/src/Eventuous.Application/AggregateService/CommandService.cs b/src/Core/src/Eventuous.Application/AggregateService/CommandService.cs index a48c79e1d..bee29094b 100644 --- a/src/Core/src/Eventuous.Application/AggregateService/CommandService.cs +++ b/src/Core/src/Eventuous.Application/AggregateService/CommandService.cs @@ -89,7 +89,7 @@ public async Task> Handle(TCommand command, Cancellatio // Zero in the global position would mean nothing, so the receiver needs to check the Changes.Length if (result.Changes.Count == 0) return Result.FromSuccess(result.State, [], 0); - var proposed = new ProposedAppend(stream, new(result.OriginalVersion), result.Changes.Select(x => new ProposedEvent(x, new())).ToArray()); + var proposed = new ProposedAppend(stream, new(result.OriginalVersion), [.. result.Changes.Select(x => new ProposedEvent(x, new()))]); var final = registeredHandler.AmendAppend?.Invoke(proposed, command) ?? proposed; var writer = registeredHandler.ResolveWriter(command); var storeResult = await writer.Store(final, Amend, cancellationToken).NoContext(); diff --git a/src/Core/src/Eventuous.Application/Diagnostics/CommandServiceMetrics.cs b/src/Core/src/Eventuous.Application/Diagnostics/CommandServiceMetrics.cs index a9c32d32d..d9f7a5925 100644 --- a/src/Core/src/Eventuous.Application/Diagnostics/CommandServiceMetrics.cs +++ b/src/Core/src/Eventuous.Application/Diagnostics/CommandServiceMetrics.cs @@ -48,7 +48,7 @@ public void Dispose() { _meter.Dispose(); } - public void SetCustomTags(TagList customTags) => _customTags = customTags.ToArray(); + public void SetCustomTags(TagList customTags) => _customTags = [.. customTags]; } record CommandServiceMetricsContext(string ServiceName, string CommandName); diff --git a/src/Core/src/Eventuous.Application/Persistence/WriterExtensions.cs b/src/Core/src/Eventuous.Application/Persistence/WriterExtensions.cs index b7a80e87f..0ca92690b 100644 --- a/src/Core/src/Eventuous.Application/Persistence/WriterExtensions.cs +++ b/src/Core/src/Eventuous.Application/Persistence/WriterExtensions.cs @@ -14,7 +14,7 @@ public async Task Store(ProposedAppend append, AmendEvent? a return await writer.AppendEvents( append.StreamName, append.ExpectedVersion, - append.Events.Select(ToStreamEvent).ToArray(), + [.. append.Events.Select(ToStreamEvent)], cancellationToken ) .NoContext(); @@ -44,7 +44,7 @@ CancellationToken cancellationToken return new NewStreamAppend( a.StreamName, a.ExpectedVersion, - a.Events.Select(evt => ToStreamEvent(evt, amendEvent)).ToArray() + [.. a.Events.Select(evt => ToStreamEvent(evt, amendEvent))] ); } ) diff --git a/src/Core/src/Eventuous.Diagnostics/EventuousDiagnostics.cs b/src/Core/src/Eventuous.Diagnostics/EventuousDiagnostics.cs index 33b8f1f89..add74b01f 100644 --- a/src/Core/src/Eventuous.Diagnostics/EventuousDiagnostics.cs +++ b/src/Core/src/Eventuous.Diagnostics/EventuousDiagnostics.cs @@ -20,7 +20,7 @@ public static class EventuousDiagnostics { public static void AddDefaultTag(string key, object? value) { var tags = new List>(Tags) { new(key, value) }; - Tags = tags.ToArray(); + Tags = [.. tags]; } [MethodImpl(MethodImplOptions.AggressiveInlining)] diff --git a/src/Core/src/Eventuous.Domain/Aggregate.cs b/src/Core/src/Eventuous.Domain/Aggregate.cs index bb53ce91c..10c0fd519 100644 --- a/src/Core/src/Eventuous.Domain/Aggregate.cs +++ b/src/Core/src/Eventuous.Domain/Aggregate.cs @@ -79,7 +79,7 @@ protected void EnsureExists(Func? getException = null) { } public void Load(long version, IEnumerable events) { - Original = events.Where(x => x != null).ToArray()!; + Original = [.. events.Where(x => x != null)!]; OriginalVersion = version; State = Original.Aggregate(State, Fold); diff --git a/src/Core/src/Eventuous.Persistence/Diagnostics/PersistenceMetrics.cs b/src/Core/src/Eventuous.Persistence/Diagnostics/PersistenceMetrics.cs index 3fa093586..8c3ad4992 100644 --- a/src/Core/src/Eventuous.Persistence/Diagnostics/PersistenceMetrics.cs +++ b/src/Core/src/Eventuous.Persistence/Diagnostics/PersistenceMetrics.cs @@ -42,7 +42,7 @@ public void Dispose() { _meter.Dispose(); } - public void SetCustomTags(TagList customTags) => _customTags = customTags.ToArray(); + public void SetCustomTags(TagList customTags) => _customTags = [.. customTags]; } record PersistenceMetricsContext(string Component, string Operation); diff --git a/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/BaseTracer.cs b/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/BaseTracer.cs index 53f80874e..d1a320a6a 100644 --- a/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/BaseTracer.cs +++ b/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/BaseTracer.cs @@ -49,9 +49,9 @@ protected async Task Trace(StreamName stream, string operation, Func task) } protected async IAsyncEnumerable TraceEnumerable( - StreamName stream, - string operation, - IAsyncEnumerable source, + StreamName stream, + string operation, + IAsyncEnumerable source, [EnumeratorCancellation] CancellationToken cancellationToken = default ) { using var activity = StartActivity(stream, operation); diff --git a/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/TracedEventReader.cs b/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/TracedEventReader.cs index f9b8048ff..35ac8dea6 100644 --- a/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/TracedEventReader.cs +++ b/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/TracedEventReader.cs @@ -15,10 +15,10 @@ public class TracedEventReader(IEventReader reader) : BaseTracer, IEventReader { IEventReader Inner { get; } = reader; public IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken) - => TraceEnumerable(stream, Operations.ReadEvents, Inner.ReadEvents(stream, start, count, cancellationToken)); + => TraceEnumerable(stream, Operations.ReadEvents, Inner.ReadEvents(stream, start, count, cancellationToken), cancellationToken); public IAsyncEnumerable ReadEventsBackwards(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken) - => TraceEnumerable(stream, Operations.ReadEvents, Inner.ReadEventsBackwards(stream, start, count, cancellationToken)); + => TraceEnumerable(stream, Operations.ReadEvents, Inner.ReadEventsBackwards(stream, start, count, cancellationToken), cancellationToken); // ReSharper disable once ConvertToAutoProperty protected override string ComponentName => _componentName; diff --git a/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/TracedEventWriter.cs b/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/TracedEventWriter.cs index 1c95d69a6..b5b387449 100644 --- a/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/TracedEventWriter.cs +++ b/src/Core/src/Eventuous.Persistence/Diagnostics/Tracing/TracedEventWriter.cs @@ -48,13 +48,7 @@ public async Task AppendEvents(IReadOnlyCollection new NewStreamAppend( - a.StreamName, - a.ExpectedVersion, - a.Events.Select(x => x with { Metadata = x.Metadata.AddActivityTags(activity) }).ToArray() - ) - ) - .ToArray(); + var tracedAppends = appends.Select(a => a with { Events = [.. a.Events.Select(x => x with { Metadata = x.Metadata.AddActivityTags(activity) })] }).ToArray(); try { var results = await writer.AppendEvents(tracedAppends, cancellationToken).NoContext(); diff --git a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs index 58c828d55..44fc54e36 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs @@ -32,7 +32,7 @@ public async Task Store( var result = await eventWriter.AppendEvents( streamName, expectedStreamVersion, - changes.Select(ToStreamEvent).ToArray(), + [.. changes.Select(ToStreamEvent)], cancellationToken ) .NoContext(); @@ -64,7 +64,7 @@ public async Task Store( return new NewStreamAppend( s.StreamName, s.ExpectedVersion, - s.Changes.Select(evt => ToStreamEvent(evt, amendEvent)).ToArray() + [.. s.Changes.Select(evt => ToStreamEvent(evt, amendEvent))] ); } ) @@ -86,106 +86,103 @@ static NewStreamEvent ToStreamEvent(object evt, AmendEvent? amendEvent) { } } - /// - /// Read a fixed number of events from an existing stream to an array. - /// Returns an empty array when the stream is not found and is false. - /// /// Event reader or event store - /// Stream name - /// Where to start reading events - /// How many events to read - /// Throw an exception if the stream is not found - /// Cancellation token - /// An array with events retrieved from the stream - public static async Task ReadEvents( - this IEventReader eventReader, - StreamName stream, - StreamReadPosition start, - int count, - bool failIfNotFound, - CancellationToken cancellationToken - ) { - try { - var result = new List(); - - await foreach (var evt in eventReader.ReadEvents(stream, start, count, cancellationToken).ConfigureAwait(false)) { - result.Add(evt); - } + extension(IEventReader eventReader) { + /// + /// Read a fixed number of events from an existing stream to an array. + /// Returns an empty array when the stream is not found and is false. + /// + /// Stream name + /// Where to start reading events + /// How many events to read + /// Throw an exception if the stream is not found + /// Cancellation token + /// An array with events retrieved from the stream + public async Task ReadEvents( + StreamName stream, + StreamReadPosition start, + int count, + bool failIfNotFound, + CancellationToken cancellationToken + ) { + try { + var result = new List(); - return result.ToArray(); - } catch (StreamNotFound) when (!failIfNotFound) { - return []; - } - } + await foreach (var evt in eventReader.ReadEvents(stream, start, count, cancellationToken).ConfigureAwait(false)) { + result.Add(evt); + } - /// - /// Read a number of events from a given stream, backwards (from the stream end), to an array. - /// Returns an empty array when the stream is not found and is false. - /// - /// Event reader or event store - /// Stream name - /// Where to start reading events - /// How many events to read - /// Throw an exception if the stream is not found - /// Cancellation token - /// An array with events retrieved from the stream - public static async Task ReadEventsBackwards( - this IEventReader eventReader, - StreamName stream, - StreamReadPosition start, - int count, - bool failIfNotFound, - CancellationToken cancellationToken - ) { - try { - var result = new List(); - - await foreach (var evt in eventReader.ReadEventsBackwards(stream, start, count, cancellationToken).ConfigureAwait(false)) { - result.Add(evt); + return [.. result]; + } catch (StreamNotFound) when (!failIfNotFound) { + return []; } - - return result.ToArray(); - } catch (StreamNotFound) when (!failIfNotFound) { - return []; } - } - /// - /// Reads a stream from the event store to a collection of - /// - /// Event reader or event store - /// Name of the stream to read from - /// Stream version to start reading from - /// Set to true if the function needs to throw when the stream isn't found. Default is false, and if there's no - /// stream with the given name found in the store, the function will return an empty collection. - /// Cancellation token - /// Collection of events wrapped in - public static async Task ReadStream( - this IEventReader eventReader, - StreamName streamName, - StreamReadPosition start, - bool failIfNotFound = true, - CancellationToken cancellationToken = default - ) { - const int pageSize = 500; - - var streamEvents = new List(); - - var position = start; - - try { - while (true) { - var events = await eventReader.ReadEvents(streamName, position, pageSize, failIfNotFound, cancellationToken).NoContext(); - streamEvents.AddRange(events); - - if (events.Length < pageSize) break; - - position = new(position.Value + events.Length); + /// + /// Read a number of events from a given stream, backwards (from the stream end), to an array. + /// Returns an empty array when the stream is not found and is false. + /// + /// Stream name + /// Where to start reading events + /// How many events to read + /// Throw an exception if the stream is not found + /// Cancellation token + /// An array with events retrieved from the stream + public async Task ReadEventsBackwards( + StreamName stream, + StreamReadPosition start, + int count, + bool failIfNotFound, + CancellationToken cancellationToken + ) { + try { + var result = new List(); + + await foreach (var evt in eventReader.ReadEventsBackwards(stream, start, count, cancellationToken).ConfigureAwait(false)) { + result.Add(evt); + } + + return [.. result]; + } catch (StreamNotFound) when (!failIfNotFound) { + return []; } - } catch (StreamNotFound) when (!failIfNotFound) { - return []; } - return streamEvents.ToArray(); + /// + /// Reads a stream from the event store to a collection of + /// + /// Name of the stream to read from + /// Stream version to start reading from + /// Set to true if the function needs to throw when the stream isn't found. Default is false, and if there's no + /// stream with the given name found in the store, the function will return an empty collection. + /// Cancellation token + /// Collection of events wrapped in + public async Task ReadStream( + StreamName streamName, + StreamReadPosition start, + bool failIfNotFound = true, + CancellationToken cancellationToken = default + ) { + const int pageSize = 500; + + var streamEvents = new List(); + + var position = start; + + try { + while (true) { + var events = await eventReader.ReadEvents(streamName, position, pageSize, failIfNotFound, cancellationToken).NoContext(); + streamEvents.AddRange(events); + + if (events.Length < pageSize) break; + + position = new(position.Value + events.Length); + } + } catch (StreamNotFound) when (!failIfNotFound) { + return []; + } + + return [.. streamEvents]; + } } } diff --git a/src/Core/src/Eventuous.Producers/BaseProducer.cs b/src/Core/src/Eventuous.Producers/BaseProducer.cs index 46e51fdf0..ca04cef26 100644 --- a/src/Core/src/Eventuous.Producers/BaseProducer.cs +++ b/src/Core/src/Eventuous.Producers/BaseProducer.cs @@ -20,7 +20,7 @@ public abstract class BaseProducer : IProducer /// Tracing options for the producer protected BaseProducer(ProducerTracingOptions? tracingOptions = null) { var options = tracingOptions ?? new ProducerTracingOptions(); - DefaultTags = options.AllTags.Concat(EventuousDiagnostics.Tags).ToArray(); + DefaultTags = [.. options.AllTags, .. EventuousDiagnostics.Tags]; } /// diff --git a/src/Core/src/Eventuous.Shared/Tools/Ensure.cs b/src/Core/src/Eventuous.Shared/Tools/Ensure.cs index 1a0187f15..b79d59572 100644 --- a/src/Core/src/Eventuous.Shared/Tools/Ensure.cs +++ b/src/Core/src/Eventuous.Shared/Tools/Ensure.cs @@ -40,13 +40,9 @@ public static T NotNull(T? value, [CallerArgumentExpression("value")] string? [DebuggerHidden] [MethodImpl(MethodImplOptions.AggressiveInlining)] public static string NotEmptyString(string? value, [CallerArgumentExpression("value")] string? name = default) { -#if NET8_0_OR_GREATER ArgumentException.ThrowIfNullOrWhiteSpace(value, name); return value; -#else - return value is null ? throw new ArgumentNullException(name) : !string.IsNullOrWhiteSpace(value) ? value : throw new ArgumentException(name); -#endif } /// diff --git a/src/Core/src/Eventuous.Shared/Tools/TaskExtensions.cs b/src/Core/src/Eventuous.Shared/Tools/TaskExtensions.cs index f6218bb78..67d5d3870 100644 --- a/src/Core/src/Eventuous.Shared/Tools/TaskExtensions.cs +++ b/src/Core/src/Eventuous.Shared/Tools/TaskExtensions.cs @@ -30,7 +30,7 @@ public static ConfiguredCancelableAsyncEnumerable NoContext(this IAsyncEnu [MethodImpl(MethodImplOptions.AggressiveInlining)] public static ValueTask WhenAll(this IEnumerable tasks) { - return tasks is ValueTask[] array ? AwaitArray(array) : AwaitArray(tasks.ToArray()); + return tasks is ValueTask[] array ? AwaitArray(array) : AwaitArray([.. tasks]); // ReSharper disable once SuggestBaseTypeForParameter async ValueTask AwaitArray(ValueTask[] t) { @@ -54,14 +54,6 @@ public static async Task> WhenAll(this IEnumerable(); var registration = cancellationToken.Register((s => (((TaskCompletionSource)s!)).SetCanceled(cancellationToken)), state); diff --git a/src/Core/src/Eventuous.Shared/TypeMap/TypeMapper.cs b/src/Core/src/Eventuous.Shared/TypeMap/TypeMapper.cs index b8d614a8b..865e11514 100644 --- a/src/Core/src/Eventuous.Shared/TypeMap/TypeMapper.cs +++ b/src/Core/src/Eventuous.Shared/TypeMap/TypeMapper.cs @@ -110,7 +110,7 @@ public void RegisterKnownEventTypes(params Assembly[] assembliesWithEvents) { Assembly[] GetDefaultAssemblies() { var firstLevel = AppDomain.CurrentDomain.GetAssemblies().Where(x => !x.IsDynamic && NamePredicate(x.GetName())).ToArray(); - return firstLevel.SelectMany(Get).Concat(firstLevel).Distinct().ToArray(); + return [.. firstLevel.SelectMany(Get).Concat(firstLevel).Distinct()]; [RequiresUnreferencedCode("Calls System.Reflection.Assembly.GetReferencedAssemblies()")] IEnumerable Get(Assembly assembly) { diff --git a/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs b/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs index 6f0f60109..57a9e7d69 100644 --- a/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs +++ b/src/Core/src/Eventuous.Subscriptions/Channels/ChannelWorkerBase.cs @@ -21,17 +21,13 @@ public ValueTask Write(T element, CancellationToken cancellationToken) protected ChannelWorkerBase(Channel channel, Func processor, int concurrencyLevel, bool throwOnFull = false) { _channel = channel; _throwOnFull = throwOnFull; - _readerTasks = Enumerable.Range(0, concurrencyLevel).Select(_ => Task.Run(() => processor(_cts.Token))).ToArray(); + _readerTasks = [.. Enumerable.Range(0, concurrencyLevel).Select(_ => Task.Run(() => processor(_cts.Token)))]; } public async ValueTask DisposeAsync() { _stopping = true; await _channel.Stop(_cts, _readerTasks, OnDispose).NoContext(); -#if NET8_0_OR_GREATER await _cts.CancelAsync().NoContext(); -#else - _cts.Cancel(); -#endif await Task.WhenAll(_readerTasks).NoThrow(); _cts.Dispose(); GC.SuppressFinalize(this); diff --git a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionMetrics.cs b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionMetrics.cs index fcfd9103f..8deebfa10 100644 --- a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionMetrics.cs +++ b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionMetrics.cs @@ -147,7 +147,7 @@ static IEnumerable> TryObserving(string metric, ObserveMetric< KeyValuePair[] _customTags = EventuousDiagnostics.Tags; - public void SetCustomTags(TagList customTags) => _customTags = _customTags.Concat(customTags).ToArray(); + public void SetCustomTags(TagList customTags) => _customTags = [.. _customTags, .. customTags]; public void Dispose() { _listener.Dispose(); diff --git a/src/Core/src/Eventuous.Subscriptions/Filters/PartitioningFilter.cs b/src/Core/src/Eventuous.Subscriptions/Filters/PartitioningFilter.cs index 0e0599029..75c30adcc 100644 --- a/src/Core/src/Eventuous.Subscriptions/Filters/PartitioningFilter.cs +++ b/src/Core/src/Eventuous.Subscriptions/Filters/PartitioningFilter.cs @@ -20,7 +20,7 @@ public PartitioningFilter(int partitionCount, GetPartitionKey? partitioner = nul _partitionCount = partitionCount; _partitioner = partitioner ?? (ctx => ctx.Stream); - _filters = Enumerable.Range(0, _partitionCount).Select(_ => new AsyncHandlingFilter(1)).ToArray(); + _filters = [.. Enumerable.Range(0, _partitionCount).Select(_ => new AsyncHandlingFilter(1))]; } protected override ValueTask Send(AsyncConsumeContext context, LinkedListNode? next) { diff --git a/src/Core/src/Eventuous.Subscriptions/Filters/TracingFilter.cs b/src/Core/src/Eventuous.Subscriptions/Filters/TracingFilter.cs index d7c38fb5c..32efda5ce 100644 --- a/src/Core/src/Eventuous.Subscriptions/Filters/TracingFilter.cs +++ b/src/Core/src/Eventuous.Subscriptions/Filters/TracingFilter.cs @@ -16,7 +16,7 @@ public class TracingFilter : ConsumeFilter { public TracingFilter(string consumerName) { var tags = new KeyValuePair[] { new(TelemetryTags.Eventuous.Consumer, consumerName) }; - _defaultTags = tags.Concat(EventuousDiagnostics.Tags).ToArray(); + _defaultTags = [.. tags, .. EventuousDiagnostics.Tags]; } protected override async ValueTask Send(IMessageConsumeContext context, LinkedListNode? next) { diff --git a/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionBuilder.cs b/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionBuilder.cs index 37b0c0d57..e3379b96b 100644 --- a/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionBuilder.cs +++ b/src/Core/src/Eventuous.Subscriptions/Registrations/SubscriptionBuilder.cs @@ -24,7 +24,7 @@ public abstract class SubscriptionBuilder(IServiceCollection services, string su protected ConsumePipe Pipe { get; } = new(); protected ResolveConsumer ResolveConsumer { get; set; } = null!; - protected IEventHandler[] ResolveHandlers(IServiceProvider sp) => _handlers.Select(x => x(sp)).ToArray(); + protected IEventHandler[] ResolveHandlers(IServiceProvider sp) => [.. _handlers.Select(x => x(sp))]; /// /// Adds an event handler to the subscription diff --git a/src/Core/src/Eventuous.Subscriptions/SubscriptionHostedService.cs b/src/Core/src/Eventuous.Subscriptions/SubscriptionHostedService.cs index 3a374a7cd..b3127f5f6 100644 --- a/src/Core/src/Eventuous.Subscriptions/SubscriptionHostedService.cs +++ b/src/Core/src/Eventuous.Subscriptions/SubscriptionHostedService.cs @@ -6,7 +6,6 @@ namespace Eventuous.Subscriptions; -using System.Diagnostics.CodeAnalysis; using Diagnostics; // ReSharper disable once ClassWithVirtualMembersNeverInherited.Global @@ -20,8 +19,6 @@ public class SubscriptionHostedService( ILogger? Log { get; } = loggerFactory?.CreateLogger(); - [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Enough warnings from the subscription")] - [UnconditionalSuppressMessage("AOT", "IL2026", Justification = "Enough warnings from the subscription")] public virtual async Task StartAsync(CancellationToken cancellationToken) { Log?.LogDebug("Starting subscription {SubscriptionId}", subscription.SubscriptionId); diff --git a/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.Amendments.cs b/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.Amendments.cs index f52a0a33e..13c6af7b2 100644 --- a/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.Amendments.cs +++ b/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.Amendments.cs @@ -9,7 +9,7 @@ public async Task Should_amend_event_from_command(CancellationToken cancellation var service = CreateService(amendEvent: AmendEvent); var cmd = CreateCommand(); - var result = await service.Handle(cmd, cancellationToken); + var dummy = await service.Handle(cmd, cancellationToken); var stream = await Store.ReadStream(StreamName.For(cmd.BookingId), StreamReadPosition.Start, cancellationToken: cancellationToken); await Assert.That(stream[0].Metadata["userId"]).IsEqualTo(cmd.ImportedBy); diff --git a/src/Core/test/Eventuous.Tests.Persistence.Base/Fixtures/Helpers.cs b/src/Core/test/Eventuous.Tests.Persistence.Base/Fixtures/Helpers.cs index 60b7f97bf..a2e13965f 100644 --- a/src/Core/test/Eventuous.Tests.Persistence.Base/Fixtures/Helpers.cs +++ b/src/Core/test/Eventuous.Tests.Persistence.Base/Fixtures/Helpers.cs @@ -14,7 +14,7 @@ public static class Helpers { public Task AppendEvents(StreamName stream, object[] evt, ExpectedStreamVersion version) { var streamEvents = evt.Select(x => new NewStreamEvent(Guid.NewGuid(), x, new())); - return fixture.EventStore.AppendEvents(stream, version, streamEvents.ToArray(), default); + return fixture.EventStore.AppendEvents(stream, version, [.. streamEvents], default); } public Task AppendEvent(StreamName stream, object evt, ExpectedStreamVersion version, Metadata? metadata = null) { diff --git a/src/Core/test/Eventuous.Tests.Persistence.Base/Fixtures/StoreFixtureBase.cs b/src/Core/test/Eventuous.Tests.Persistence.Base/Fixtures/StoreFixtureBase.cs index 9c5287091..27cc6833b 100644 --- a/src/Core/test/Eventuous.Tests.Persistence.Base/Fixtures/StoreFixtureBase.cs +++ b/src/Core/test/Eventuous.Tests.Persistence.Base/Fixtures/StoreFixtureBase.cs @@ -81,10 +81,6 @@ protected virtual void GetDependencies(IServiceProvider provider) { } protected static string GetSchemaName() => NormaliseRegex().Replace(new Faker().Internet.UserName(), "").ToLower(); -#if NET8_0_OR_GREATER [GeneratedRegex(@"[\.\-\s]")] private static partial Regex NormaliseRegex(); -#else - static Regex NormaliseRegex() => new(@"[\.\-\s]"); -#endif } diff --git a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs index edcc230c9..25adbbfba 100644 --- a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs +++ b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs @@ -29,7 +29,7 @@ public async Task ShouldReadOne(CancellationToken cancellationToken) { [Test] [Category("Store")] public async Task ShouldReadMany(CancellationToken cancellationToken) { - object[] events = _fixture.CreateEvents(20).ToArray(); + object[] events = [.. _fixture.CreateEvents(20)]; var streamName = Helpers.GetStreamName(); await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); @@ -42,7 +42,7 @@ public async Task ShouldReadMany(CancellationToken cancellationToken) { [Test] [Category("Store")] public async Task ShouldReadTail(CancellationToken cancellationToken) { - object[] events = _fixture.CreateEvents(20).ToArray(); + object[] events = [.. _fixture.CreateEvents(20)]; var streamName = Helpers.GetStreamName(); await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); @@ -55,7 +55,7 @@ public async Task ShouldReadTail(CancellationToken cancellationToken) { [Test] [Category("Store")] public async Task ShouldReadHead(CancellationToken cancellationToken) { - object[] events = _fixture.CreateEvents(20).ToArray(); + object[] events = [.. _fixture.CreateEvents(20)]; var streamName = Helpers.GetStreamName(); await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); @@ -87,7 +87,7 @@ await Assert.That(result[0].Metadata.ToDictionary(m => m.Key, m => ((JsonElement [Test] [Category("Store")] public async Task ShouldThrowWhenReadingForwardsFromNegativePosition(CancellationToken cancellationToken) { - object[] events = _fixture.CreateEvents(10).ToArray(); + object[] events = [.. _fixture.CreateEvents(10)]; var streamName = Helpers.GetStreamName(); await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); @@ -102,7 +102,7 @@ public async Task ShouldThrowWhenReadingForwardsFromNegativePosition(Cancellatio [Test] [Category("Store")] public async Task ShouldReadBackwardsFromEnd(CancellationToken cancellationToken) { - object[] events = _fixture.CreateEvents(10).ToArray(); + object[] events = [.. _fixture.CreateEvents(10)]; var streamName = Helpers.GetStreamName(); await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); @@ -118,7 +118,7 @@ public async Task ShouldReadBackwardsFromEnd(CancellationToken cancellationToken [Test] [Category("Store")] public async Task ShouldReadBackwardsFromMiddle(CancellationToken cancellationToken) { - object[] events = _fixture.CreateEvents(20).ToArray(); + object[] events = [.. _fixture.CreateEvents(20)]; var streamName = Helpers.GetStreamName(); await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); @@ -136,7 +136,7 @@ public async Task ShouldReadBackwardsFromMiddle(CancellationToken cancellationTo [Test] [Category("Store")] public async Task ShouldReturnWhenReadingBackwards(CancellationToken cancellationToken) { - object[] events = _fixture.CreateEvents(10).ToArray(); + object[] events = [.. _fixture.CreateEvents(10)]; var streamName = Helpers.GetStreamName(); await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); @@ -149,7 +149,7 @@ public async Task ShouldReturnWhenReadingBackwards(CancellationToken cancellatio [Test] [Category("Store")] public async Task ShouldThrowWhenReadingBackwardsFromNegativePosition(CancellationToken cancellationToken) { - object[] events = _fixture.CreateEvents(10).ToArray(); + object[] events = [.. _fixture.CreateEvents(10)]; var streamName = Helpers.GetStreamName(); await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); diff --git a/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs b/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs index 23335a705..ae8fbd857 100644 --- a/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs +++ b/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs @@ -17,12 +17,6 @@ public TestState() { public void Process() => Apply(new Events.RoomBooked("1", DateTime.Now, DateTime.Now.AddDays(1), 100)); } -file class TestEventHandler : Eventuous.Subscriptions.EventHandler { - public TestEventHandler() { - On(ctx => new System.Threading.Tasks.ValueTask()); - } -} - file static class Events { [PublicAPI] public record RoomBooked(string RoomId, DateTime CheckIn, DateTime CheckOut, decimal Price); diff --git a/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzer_Ev001_Tests.cs b/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzer_Ev001_Tests.cs index fea6ec2fb..7ada3bdc1 100644 --- a/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzer_Ev001_Tests.cs +++ b/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzer_Ev001_Tests.cs @@ -37,7 +37,7 @@ static async Task GetAnalyzerDiagnosticsAsync(Compilation compilat var withAnalyzers = compilation.WithAnalyzers([analyzer]); var diagnostics = await withAnalyzers.GetAnalyzerDiagnosticsAsync().ConfigureAwait(false); // Filter out anything not from our analyzer id just in case - return diagnostics.Where(d => d.Id == EventUsageAnalyzer.DiagnosticId).ToArray(); + return [.. diagnostics.Where(d => d.Id == EventUsageAnalyzer.DiagnosticId)]; } static string LoadAnalyzedSource([CallerFilePath] string? caller = null) { @@ -60,9 +60,7 @@ static CSharpCompilation CreateCompilation(string source) { // Add runtime assemblies to resolve core types (DateTime, ValueTask, etc.) var runtimeDir = Path.GetDirectoryName(typeof(object).Assembly.Location)!; - foreach (var dll in Directory.GetFiles(runtimeDir, "System.*.dll")) { - refs.Add(MetadataReference.CreateFromFile(dll)); - } + refs.AddRange(Directory.GetFiles(runtimeDir, "System.*.dll").Select(dll => MetadataReference.CreateFromFile(dll)).Cast()); TryAddRef(refs, "netstandard"); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/TestEventHandler.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/TestEventHandler.cs index 21092da92..38202ac5b 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/TestEventHandler.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions.Base/Fixtures/TestEventHandler.cs @@ -41,7 +41,7 @@ public Hypothesis AssertCollection(TimeSpan deadline, List colle /// Messages handled so far. Backed by a concurrent queue so tests can poll it while the subscription /// keeps handling on background threads. /// - public IReadOnlyCollection Handled => _handled.ToArray(); + public IReadOnlyCollection Handled => [.. _handled]; public override async ValueTask HandleEvent(IMessageConsumeContext context) { await Task.Delay(_delay); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToStream.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToStream.cs index 3b32bc012..27e58a0b1 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToStream.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToStream.cs @@ -91,7 +91,7 @@ async Task> GenerateAndProduceEvents(int count) { var events = commands.Select(ToEvent).ToList(); var streamEvents = events.Select(x => new NewStreamEvent(Guid.NewGuid(), x, new())); - await fixture.EventStore.AppendEvents(streamName, ExpectedStreamVersion.Any, streamEvents.ToArray(), default); + await fixture.EventStore.AppendEvents(streamName, ExpectedStreamVersion.Any, [.. streamEvents], default); return events; } diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs index 721ef9eab..533a77cdb 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionDropBase.cs @@ -82,7 +82,7 @@ protected async Task ShouldResubscribeAfterConnectionDrop(CancellationToken canc () => { var handled = fixture.Handler.Handled; - return expected.All(e => handled.Contains(e)); + return expected.All(handled.Contains); }, DropTimeout, cancellationToken diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionMeasureBase.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionMeasureBase.cs index d60e45311..765e859b8 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionMeasureBase.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionMeasureBase.cs @@ -1,7 +1,6 @@ using DotNet.Testcontainers.Containers; using Eventuous.Subscriptions; using Eventuous.Subscriptions.Checkpoints; -using Eventuous.Subscriptions.Diagnostics; using Eventuous.Sut.App; using Eventuous.Tests.Persistence.Base.Fixtures; diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerBackpressureTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerBackpressureTests.cs index d609476ca..206234303 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerBackpressureTests.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerBackpressureTests.cs @@ -20,14 +20,6 @@ public async Task Commit_awaits_capacity_instead_of_throwing_when_the_channel_is TaskCompletionSource storeGate = new(TaskCreationOptions.RunContinuationsAsynchronously); TaskCompletionSource storeEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); - async ValueTask CommitFn(Checkpoint checkpoint, bool force, CancellationToken ct) { - storeEntered.TrySetResult(); - await storeGate.Task.WaitAsync(ct); - lock (committed) committed.Add(checkpoint.Position!.Value); - - return checkpoint; - } - await using var handler = new CheckpointCommitHandler("backpressure-sub", CommitFn, TimeSpan.FromMilliseconds(10), batchSize: 1); // The gate must open no matter how the test body ends: a failed assertion with the gate @@ -72,26 +64,28 @@ async ValueTask CommitFn(Checkpoint checkpoint, bool force, Cancella await Task.Delay(50, cancellationToken); } - snapshot.ShouldBe(Enumerable.Range(0, 1002).Select(i => (ulong)i).ToList()); + snapshot.ShouldBe([.. Enumerable.Range(0, 1002).Select(i => (ulong)i)]); } finally { storeGate.TrySetResult(); } - } - [Test] - [Timeout(20_000)] - public async Task Dispose_releases_a_backpressured_commit_and_drains_without_hanging(CancellationToken cancellationToken) { - List<(ulong Position, bool Force)> committed = []; - TaskCompletionSource storeGate = new(TaskCreationOptions.RunContinuationsAsynchronously); - TaskCompletionSource storeEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); + return; async ValueTask CommitFn(Checkpoint checkpoint, bool force, CancellationToken ct) { storeEntered.TrySetResult(); await storeGate.Task.WaitAsync(ct); - lock (committed) committed.Add((checkpoint.Position!.Value, force)); + lock (committed) committed.Add(checkpoint.Position!.Value); return checkpoint; } + } + + [Test] + [Timeout(20_000)] + public async Task Dispose_releases_a_backpressured_commit_and_drains_without_hanging(CancellationToken cancellationToken) { + List<(ulong Position, bool Force)> committed = []; + TaskCompletionSource storeGate = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource storeEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); var handler = new CheckpointCommitHandler("backpressure-dispose-sub", CommitFn, TimeSpan.FromMilliseconds(10), batchSize: 1); @@ -146,7 +140,7 @@ async ValueTask CommitFn(Checkpoint checkpoint, bool force, Cancella // The queued positions (0..1000) drained normally, in order, during dispose; the parked // overflow write (1001) never entered the channel, so it must not appear. var normal = snapshot.Where(x => !x.Force).Select(x => x.Position).ToList(); - normal.ShouldBe(Enumerable.Range(0, 1001).Select(i => (ulong)i).ToList()); + normal.ShouldBe([.. Enumerable.Range(0, 1001).Select(i => (ulong)i)]); // CheckpointCommitHandler's OnDispose force-recommits whatever it last successfully stored — // it should match the highest position that drained normally, not something stale or ahead @@ -161,5 +155,15 @@ async ValueTask CommitFn(Checkpoint checkpoint, bool force, Cancella // of stalling the finally block until the test timeout. await Task.WhenAny(disposeTask ?? handler.DisposeAsync().AsTask(), Task.Delay(TimeSpan.FromSeconds(10))); } + + return; + + async ValueTask CommitFn(Checkpoint checkpoint, bool force, CancellationToken ct) { + storeEntered.TrySetResult(); + await storeGate.Task.WaitAsync(ct); + lock (committed) committed.Add((checkpoint.Position!.Value, force)); + + return checkpoint; + } } } diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerConcurrencyTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerConcurrencyTests.cs index eace80a28..196cd174f 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerConcurrencyTests.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions/CheckpointCommitHandlerConcurrencyTests.cs @@ -50,7 +50,7 @@ public async Task Commit_diagnostic_is_not_emitted_on_the_caller_thread(Cancella var callerThreadId = 0; var caller = new Thread(() => { callerThreadId = Environment.CurrentManagedThreadId; - handler.Commit(new CommitPosition(0, 0, DateTime.UtcNow), ct).AsTask().GetAwaiter().GetResult(); + handler.Commit(new(0, 0, DateTime.UtcNow), ct).AsTask().GetAwaiter().GetResult(); }) { IsBackground = true }; caller.Start(); caller.Join(); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/ResubscribeOnHandlerFailureTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/ResubscribeOnHandlerFailureTests.cs index 1ca17fc48..fb7b27e7b 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions/ResubscribeOnHandlerFailureTests.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions/ResubscribeOnHandlerFailureTests.cs @@ -282,7 +282,7 @@ public void TriggerDropped() => Dropped(DropReason.SubscriptionError, new InvalidOperationException("Simulated drop for race test")); protected override ValueTask Subscribe(CancellationToken cancellationToken) { - _runner = new TaskRunner(token => PollEvents(token)).Start(); + _runner = new TaskRunner(PollEvents).Start(); return default; } diff --git a/src/Core/test/Eventuous.Tests/Fixtures/IdGenerator.cs b/src/Core/test/Eventuous.Tests/Fixtures/IdGenerator.cs index 3919fff0f..b5148b025 100644 --- a/src/Core/test/Eventuous.Tests/Fixtures/IdGenerator.cs +++ b/src/Core/test/Eventuous.Tests/Fixtures/IdGenerator.cs @@ -1,5 +1,5 @@ namespace Eventuous.Tests.Fixtures; -public class IdGenerator { +public static class IdGenerator { public static string GetId() => Guid.NewGuid().ToString("N"); } diff --git a/src/Diagnostics/src/Eventuous.Diagnostics.Logging/LoggingEventListener.cs b/src/Diagnostics/src/Eventuous.Diagnostics.Logging/LoggingEventListener.cs index a53a0a4d0..399412a7f 100644 --- a/src/Diagnostics/src/Eventuous.Diagnostics.Logging/LoggingEventListener.cs +++ b/src/Diagnostics/src/Eventuous.Diagnostics.Logging/LoggingEventListener.cs @@ -48,7 +48,7 @@ protected override void OnEventWritten(EventWrittenEventArgs evt) { #pragma warning disable CA2254 if (evt.Payload != null) // ReSharper disable once TemplateIsNotCompileTimeConstantProblem - _log.Log(level, evt.Message, evt.Payload.ToArray()); + _log.Log(level, evt.Message, [.. evt.Payload]); else // ReSharper disable once TemplateIsNotCompileTimeConstantProblem _log.Log(level, evt.Message); diff --git a/src/Diagnostics/test/Eventuous.Tests.OpenTelemetry/Fakes/TestExporter.cs b/src/Diagnostics/test/Eventuous.Tests.OpenTelemetry/Fakes/TestExporter.cs index fb97e6357..ab9007279 100644 --- a/src/Diagnostics/test/Eventuous.Tests.OpenTelemetry/Fakes/TestExporter.cs +++ b/src/Diagnostics/test/Eventuous.Tests.OpenTelemetry/Fakes/TestExporter.cs @@ -35,10 +35,10 @@ public MetricValue[] CollectValues() { _ => throw new ArgumentOutOfRangeException() }; - values.Add(new(metric.Name, tags.Select(x => x.Item1).ToArray(), tags.Select(x => x.Item2).ToArray()!, metricValue)); + values.Add(new(metric.Name, [.. tags.Select(x => x.Item1)], [.. tags.Select(x => x.Item2)!], metricValue)); } } - return values.ToArray(); + return [.. values]; } } diff --git a/src/Experimental/src/ElasticPlayground/Generator.cs b/src/Experimental/src/ElasticPlayground/Generator.cs index d270f0952..ebd9809a7 100644 --- a/src/Experimental/src/ElasticPlayground/Generator.cs +++ b/src/Experimental/src/ElasticPlayground/Generator.cs @@ -3,7 +3,7 @@ namespace ElasticPlayground; -public class Generator{ +public static class Generator{ public static string RandomString() => Guid.NewGuid().ToString(); static readonly Faker Faker = new Faker() diff --git a/src/Experimental/src/Eventuous.ElasticSearch/Producers/ElasticProducer.cs b/src/Experimental/src/Eventuous.ElasticSearch/Producers/ElasticProducer.cs index f09ace90c..32668560c 100644 --- a/src/Experimental/src/Eventuous.ElasticSearch/Producers/ElasticProducer.cs +++ b/src/Experimental/src/Eventuous.ElasticSearch/Producers/ElasticProducer.cs @@ -42,7 +42,7 @@ protected override async Task ProduceMessages( await error.Nack(result.DebugInformation, result.OriginalException).NoContext(); } - messagesList = messagesList.Except(errors).ToList(); + messagesList = [.. messagesList.Except(errors)]; } } diff --git a/src/Experimental/src/Eventuous.ElasticSearch/Store/ElasticEventStore.cs b/src/Experimental/src/Eventuous.ElasticSearch/Store/ElasticEventStore.cs index 5141757dd..11e2ba514 100644 --- a/src/Experimental/src/Eventuous.ElasticSearch/Store/ElasticEventStore.cs +++ b/src/Experimental/src/Eventuous.ElasticSearch/Store/ElasticEventStore.cs @@ -90,18 +90,18 @@ async Task ReadEventsInternal(Func new StreamEvent( - Guid.Parse(x.MessageId), - x.Message, - Metadata.FromHeaders(x.Metadata), - x.ContentType, - x.StreamPosition, - x.Created + return [ + .. response.Documents + .Select(x => new StreamEvent( + Guid.Parse(x.MessageId), + x.Message, + Metadata.FromHeaders(x.Metadata), + x.ContentType, + x.StreamPosition, + x.Created + ) ) - ) - .ToArray(); + ]; } public Task TruncateStream( diff --git a/src/Experimental/src/Eventuous.Spyglass/SpyglassRegistry.cs b/src/Experimental/src/Eventuous.Spyglass/SpyglassRegistry.cs index 64588a6de..84aa37c2c 100644 --- a/src/Experimental/src/Eventuous.Spyglass/SpyglassRegistry.cs +++ b/src/Experimental/src/Eventuous.Spyglass/SpyglassRegistry.cs @@ -3,17 +3,21 @@ namespace Eventuous.Spyglass; +#if NET8_0 +using Lock = object; +#endif + public delegate StreamName SpyglassGetStreamName(StreamNameMap? map, string entityId); public delegate Task SpyglassLoadDelegate(IEventStore eventStore, StreamName streamName, int version); public record SpyglassAggregateInfo( - string? AggregateType, - string StateType, - string[] Methods, - string[] Events, - SpyglassGetStreamName GetStreamName, - SpyglassLoadDelegate LoadDelegate + string? AggregateType, + string StateType, + string[] Methods, + string[] Events, + SpyglassGetStreamName GetStreamName, + SpyglassLoadDelegate LoadDelegate ) { public Guid Id { get; init; } } @@ -28,32 +32,22 @@ public static class SpyglassRegistry { // Module initializers from different assemblies can call Register concurrently when the assemblies // are loaded by parallel test fixtures (or any parallel host startup), so the backing store has to // be thread-safe. Reads also enumerate the snapshot, so we publish a fresh array on every write. - static SpyglassAggregateInfo[] _aggregates = []; - static readonly object _lock = new(); + static SpyglassAggregateInfo[] aggregates = []; + static readonly Lock Lock = new(); public static void Register(SpyglassAggregateInfo info) { - lock (_lock) { + lock (Lock) { var entry = info with { Id = Guid.NewGuid() }; - var next = new SpyglassAggregateInfo[_aggregates.Length + 1]; - Array.Copy(_aggregates, next, _aggregates.Length); - next[^1] = entry; - _aggregates = next; + var next = new SpyglassAggregateInfo[aggregates.Length + 1]; + Array.Copy(aggregates, next, aggregates.Length); + next[^1] = entry; + aggregates = next; } } public static SpyglassAggregateEntry[] GetAggregates() - => _aggregates.Select(a => new SpyglassAggregateEntry(a.Id, a.AggregateType, a.StateType, a.Methods, a.Events)).ToArray(); + => [.. aggregates.Select(a => new SpyglassAggregateEntry(a.Id, a.AggregateType, a.StateType, a.Methods, a.Events))]; public static SpyglassAggregateInfo? FindById(Guid id) - => Array.Find(_aggregates, x => x.Id == id); - - public static SpyglassAggregateInfo? FindByTypeName(string typeName) { - var snapshot = _aggregates; - - return Array.Find(snapshot, x => x.AggregateType == typeName) - ?? Array.Find(snapshot, x => StripStateSuffix(x.StateType) == typeName); - } - - static string StripStateSuffix(string s) - => s.EndsWith("State") && s.Length > 5 ? s[..^5] : s; + => Array.Find(aggregates, x => x.Id == id); } diff --git a/src/Experimental/test/Eventuous.Tests.Spyglass.Generators/CompilationHelper.cs b/src/Experimental/test/Eventuous.Tests.Spyglass.Generators/CompilationHelper.cs index af486ffff..320b80847 100644 --- a/src/Experimental/test/Eventuous.Tests.Spyglass.Generators/CompilationHelper.cs +++ b/src/Experimental/test/Eventuous.Tests.Spyglass.Generators/CompilationHelper.cs @@ -42,7 +42,7 @@ public static (string? GeneratedSource, Diagnostic[] Diagnostics) RunGenerator(C var generatedTree = outputCompilation.SyntaxTrees.FirstOrDefault(t => t.FilePath.Contains("SpyglassModule_")); - return (generatedTree?.GetText().ToString(), diagnostics.ToArray()); + return (generatedTree?.GetText().ToString(), [.. diagnostics]); } static void TryAddRef(List refs, string asmName) { diff --git a/src/Extensions/src/Eventuous.Extensions.AspNetCore/Http/CommandMappingRegistry.cs b/src/Extensions/src/Eventuous.Extensions.AspNetCore/Http/CommandMappingRegistry.cs index df54900ae..791dfdfa3 100644 --- a/src/Extensions/src/Eventuous.Extensions.AspNetCore/Http/CommandMappingRegistry.cs +++ b/src/Extensions/src/Eventuous.Extensions.AspNetCore/Http/CommandMappingRegistry.cs @@ -47,7 +47,7 @@ public static IEnumerable GetForState(Type stateType) { if (!PerState.TryGetValue(stateType, out var list)) yield break; List copy; - lock (list) copy = list.ToList(); + lock (list) copy = [.. list]; foreach (var a in copy) { yield return a; @@ -56,7 +56,7 @@ public static IEnumerable GetForState(Type stateType) { public static IEnumerable GetAll() { List copy; - lock (All) copy = All.ToList(); + lock (All) copy = [.. All]; foreach (var a in copy) { yield return a; diff --git a/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore.Analyzers/AnalyzerTestShared.cs b/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore.Analyzers/AnalyzerTestShared.cs index 11952ec46..0f5d08d83 100644 --- a/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore.Analyzers/AnalyzerTestShared.cs +++ b/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore.Analyzers/AnalyzerTestShared.cs @@ -11,7 +11,7 @@ static async Task GetAnalyzerDiagnosticsAsync(Compilation compilat var withAnalyzers = compilation.WithAnalyzers([analyzer]); var diagnostics = await withAnalyzers.GetAnalyzerDiagnosticsAsync().ConfigureAwait(false); - return diagnostics.Where(d => d.Id is Diags.DiagnosticId or Diags.RouteDiagnosticId).ToArray(); + return [.. diagnostics.Where(d => d.Id is Diags.DiagnosticId or Diags.RouteDiagnosticId)]; } static CSharpCompilation CreateCompilation(string source) { diff --git a/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore/DiscoveredCommandsTests.cs b/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore/DiscoveredCommandsTests.cs index f18148757..3ddbfad32 100644 --- a/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore/DiscoveredCommandsTests.cs +++ b/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore/DiscoveredCommandsTests.cs @@ -16,7 +16,7 @@ public async Task RegisterStateCommands() { var b = app.MapDiscoveredCommands(); - IEnumerable actual = b.DataSources.First().Endpoints.Select(x => x.DisplayName).Order().ToList()!; + IEnumerable actual = [.. b.DataSources.First().Endpoints.Select(x => x.DisplayName).Order()!]; var expected = new[] { "HTTP: POST nested-book", "HTTP: POST import2" }; @@ -31,7 +31,7 @@ public async Task RegisterStatesCommands() { var b = app.MapDiscoveredCommands(typeof(TestCommands.DuplicateCommand)); - IEnumerable actual = b.DataSources.First().Endpoints.Select(x => x.DisplayName).Order().ToList()!; + IEnumerable actual = [.. b.DataSources.First().Endpoints.Select(x => x.DisplayName).Order()!]; var expected = new[] { "HTTP: POST nested-book", "HTTP: POST import2", "HTTP: POST import-wrong" }; diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs b/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs index d8cf4a269..165bac57d 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs @@ -195,6 +195,7 @@ public Task AppendEvents( } ); + // ReSharper disable once AsyncMethodWithoutAwait static async IAsyncEnumerable ToAsyncEnumerable(IEnumerable source) { foreach (var item in source) { yield return item; @@ -362,11 +363,12 @@ StreamEvent AsStreamEvent(object payload) } StreamEvent[] ToStreamEvents(ResolvedEvent[] resolvedEvents) - => resolvedEvents - .Select(ToStreamEvent) - .Where(x => x != null) - .Select(x => x!.Value) - .ToArray(); + => [ + .. resolvedEvents + .Select(ToStreamEvent) + .Where(x => x != null) + .Select(x => x!.Value) + ]; record ErrorInfo(string Message, params object[] Args); diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs index 5ccd34c61..043a43070 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs @@ -59,7 +59,7 @@ public AllStreamSubscription( /// Event serializer /// Metadata serializer public AllStreamSubscription( - KurrentDBClient client, + KurrentDBClient client, AllStreamSubscriptionOptions options, ICheckpointStore checkpointStore, ConsumePipe consumePipe, @@ -112,7 +112,7 @@ protected override async ValueTask Subscribe(CancellationToken cancellationToken } } catch { await messages.DisposeAsync().NoContext(); - subscription.Dispose(); + await subscription.DisposeAsync(); throw; } @@ -205,7 +205,7 @@ CancellationToken cancellationToken // Double disposal on the unsubscribe path is fine; on the dropped path this is the only // cleanup of the underlying call before Resubscribe replaces the subscription. await messages.DisposeAsync().NoContext(); - subscription.Dispose(); + await subscription.DisposeAsync(); } } @@ -225,7 +225,9 @@ CancellationToken cancellationToken protected override async ValueTask Unsubscribe(CancellationToken cancellationToken) { try { Stopping.Cancel(false); - _subscription?.Dispose(); + + if (_subscription != null) + await _subscription.DisposeAsync(); _subscription = null; if (_messagePump is { } pump) { diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs index 280d9fecd..c666cefca 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/CheckpointReachedTests.cs @@ -270,7 +270,7 @@ protected override void GetDependencies(IServiceProvider provider) { public record UnmatchedEvent(int Number) { public const string TypeName = "unmatched-event"; - public static List CreateMany(int count) => Enumerable.Range(0, count).Select(i => new UnmatchedEvent(i)).ToList(); + public static List CreateMany(int count) => [.. Enumerable.Range(0, count).Select(i => new UnmatchedEvent(i))]; } /// diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/PersistentSubscriptionFailureTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/PersistentSubscriptionFailureTests.cs index ba723f13c..5365834d6 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/PersistentSubscriptionFailureTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/PersistentSubscriptionFailureTests.cs @@ -51,7 +51,7 @@ static StreamPersistentSubscription CreateSubscription(string id, string connect StreamName = stream, SubscriptionId = id, ThrowOnError = true, - SubscriptionSettings = new PersistentSubscriptionSettings( + SubscriptionSettings = new( resolveLinkTos: false, messageTimeout: TimeSpan.FromSeconds(2), maxRetryCount: 0 diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/ResolvedLinkCheckpointTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/ResolvedLinkCheckpointTests.cs index 63cfe4b57..c6e573edf 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/ResolvedLinkCheckpointTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/ResolvedLinkCheckpointTests.cs @@ -55,7 +55,7 @@ public async Task CheckpointAdvancesToLinkPositionNotTargetPosition(Cancellation var linkResult = await Client.AppendToStreamAsync( _linkStream, StreamState.Any, - [new EventData(Uuid.NewUuid(), "$>", Encoding.UTF8.GetBytes($"0@{_stream}"), contentType: "application/octet-stream")], + [new(Uuid.NewUuid(), "$>", Encoding.UTF8.GetBytes($"0@{_stream}"), contentType: "application/octet-stream")], cancellationToken: cancellationToken ); diff --git a/src/Postgres/src/Eventuous.Postgresql/Projections/PostgresProjector.cs b/src/Postgres/src/Eventuous.Postgresql/Projections/PostgresProjector.cs index e4d9756e1..e341fc43d 100644 --- a/src/Postgres/src/Eventuous.Postgresql/Projections/PostgresProjector.cs +++ b/src/Postgres/src/Eventuous.Postgresql/Projections/PostgresProjector.cs @@ -39,8 +39,8 @@ protected static NpgsqlCommand Project(NpgsqlConnection connection, string comma return cmd; } - protected static NpgsqlCommand Project(NpgsqlConnection connection, string commandText, params (string Name, object Value)[] parameters) - => Project(connection, commandText, parameters.Select(x => new NpgsqlParameter(x.Name, x.Value)).ToArray()); + protected static NpgsqlCommand Project(NpgsqlConnection connection, string commandText, params (string Name, object Value)[] parameters) + => Project(connection, commandText, [.. parameters.Select(x => new NpgsqlParameter(x.Name, x.Value))]); } public delegate NpgsqlCommand ProjectToPostgres(NpgsqlConnection connection, MessageConsumeContext consumeContext) where T : class; diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Store/TieredStoreTests.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Store/TieredStoreTests.cs index b63616a8f..8010ccffc 100644 --- a/src/Postgres/test/Eventuous.Tests.Postgres/Store/TieredStoreTests.cs +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Store/TieredStoreTests.cs @@ -4,4 +4,5 @@ namespace Eventuous.Tests.Postgres.Store; [InheritsTests] +// ReSharper disable once UnusedType.Global public class TieredStoreTests(StoreFixture storeFixture) : TieredStoreTestsBase(storeFixture); diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapIgnoreTest.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapIgnoreTest.cs index 159019d16..c7e4774e0 100644 --- a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapIgnoreTest.cs +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/GapIgnoreTest.cs @@ -15,12 +15,12 @@ public class GapIgnoreTest() : SubscriptionTestBase(Fixture) { public async Task ShouldIgnoreOldGaps(CancellationToken cancellationToken) { var streamName = new StreamName("test-stream-gap-ignore"); - await Fixture.AppendEvents(streamName, Fixture.CreateEvents(2).Cast().ToArray(), ExpectedStreamVersion.NoStream); + await Fixture.AppendEvents(streamName, [.. Fixture.CreateEvents(2)], ExpectedStreamVersion.NoStream); // Create a gap that will become "old" await Fixture.InsertGap(streamName, 1); - await Fixture.AppendEvents(streamName, Fixture.CreateEvents(3).Cast().ToArray(), ExpectedStreamVersion.Any); + await Fixture.AppendEvents(streamName, [.. Fixture.CreateEvents(3)], ExpectedStreamVersion.Any); await Task.Delay(1000, cancellationToken); diff --git a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/TombstonesCreationTest.cs b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/TombstonesCreationTest.cs index c8e4a7e49..d46bdf8f3 100644 --- a/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/TombstonesCreationTest.cs +++ b/src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/TombstonesCreationTest.cs @@ -16,20 +16,20 @@ public async Task ShouldCreateTombstones(CancellationToken cancellationToken) { var streamName = new StreamName("test-stream"); // create 2 events - await Fixture.AppendEvents(streamName, Fixture.CreateEvents(2).Cast().ToArray(), ExpectedStreamVersion.NoStream); + await Fixture.AppendEvents(streamName, [.. Fixture.CreateEvents(2)], ExpectedStreamVersion.NoStream); // create gap await Fixture.InsertGap(streamName, 1); // create other 2 events - await Fixture.AppendEvents(streamName, Fixture.CreateEvents(2).Cast().ToArray(), ExpectedStreamVersion.Any); + await Fixture.AppendEvents(streamName, [.. Fixture.CreateEvents(2)], ExpectedStreamVersion.Any); // create gaps await Fixture.InsertGap(streamName, 3); await Fixture.InsertGap(streamName, 3); // create other 2 events - await Fixture.AppendEvents(streamName, Fixture.CreateEvents(2).Cast().ToArray(), ExpectedStreamVersion.Any); + await Fixture.AppendEvents(streamName, [.. Fixture.CreateEvents(2)], ExpectedStreamVersion.Any); await Fixture.StartSubscription(); diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index a4a733693..ab7edb3ab 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -46,7 +46,7 @@ public async IAsyncEnumerable ReadEvents(StreamName stream, StreamR throw new StreamNotFound(stream); } - events = result.Select(x => ToStreamEvent(x, _serializer, _metaSerializer)).ToArray(); + events = [.. result.Select(x => ToStreamEvent(x, _serializer, _metaSerializer))]; } catch (InvalidOperationException e) when (e.Message.Contains("Reading is not allowed after reader was completed") || cancellationToken.IsCancellationRequested) { throw new OperationCanceledException("Redis read operation terminated", e, cancellationToken); diff --git a/src/Redis/src/Eventuous.Redis/Subscriptions/RedisAllStreamSubscription.cs b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisAllStreamSubscription.cs index e88fe9361..e66edc43e 100644 --- a/src/Redis/src/Eventuous.Redis/Subscriptions/RedisAllStreamSubscription.cs +++ b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisAllStreamSubscription.cs @@ -45,7 +45,7 @@ protected override async Task ReadEvents(IDatabase database, lo ); } - return persistentEvents.ToArray(); + return [.. persistentEvents]; } } diff --git a/src/Redis/src/Eventuous.Redis/Subscriptions/RedisStreamSubscription.cs b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisStreamSubscription.cs index 605903816..26478d0ba 100644 --- a/src/Redis/src/Eventuous.Redis/Subscriptions/RedisStreamSubscription.cs +++ b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisStreamSubscription.cs @@ -28,8 +28,8 @@ protected override async Task BeforeSubscribe(CancellationToken cancellationToke protected override async Task ReadEvents(IDatabase database, long position) { var events = await database.StreamReadAsync(_streamName, position.ToRedisValue(), Options.MaxPageSize).NoContext(); - return events.Select( - evt => new ReceivedEvent( + return [ + .. events.Select(evt => new ReceivedEvent( Guid.Parse(evt[MessageId].ToString()), evt[MessageType]!, evt.Id.ToLong(), @@ -40,7 +40,7 @@ protected override async Task ReadEvents(IDatabase database, lo _streamName ) ) - .ToArray(); + ]; } readonly string _streamName = options.Stream.ToString(); diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Helpers.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Helpers.cs index ec874b806..413b0fece 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Helpers.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Helpers.cs @@ -27,7 +27,7 @@ CancellationToken cancellationToken ) { var streamEvents = evt.Select(x => new NewStreamEvent(Guid.NewGuid(), x, new())); - return fixture.EventWriter.AppendEvents(stream, version, streamEvents.ToArray(), cancellationToken); + return fixture.EventWriter.AppendEvents(stream, version, [.. streamEvents], cancellationToken); } public Task AppendEvent(StreamName stream, object evt, ExpectedStreamVersion version, CancellationToken cancellationToken) { diff --git a/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToAll.cs b/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToAll.cs index 604354d4e..1c797604a 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToAll.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToAll.cs @@ -90,7 +90,7 @@ static BookingImported ToEvent(ImportBooking cmd) var events = commands.Select(ToEvent).ToList(); var streamEvents = events.Select(x => new NewStreamEvent(Guid.NewGuid(), x, new())); - var result = await _fixture.IntegrationFixture.EventWriter.AppendEvents(_fixture.Stream, ExpectedStreamVersion.Any, streamEvents.ToArray(), default); + var result = await _fixture.IntegrationFixture.EventWriter.AppendEvents(_fixture.Stream, ExpectedStreamVersion.Any, [.. streamEvents], default); return (events, result); } diff --git a/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToStream.cs b/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToStream.cs index d862b40c0..9b485c65a 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToStream.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToStream.cs @@ -90,7 +90,7 @@ async Task> GenerateAndProduceEvents(int count) { var events = commands.Select(ToEvent).ToList(); var streamEvents = events.Select(x => new NewStreamEvent(Guid.NewGuid(), x, new())); - await _fixture.IntegrationFixture.EventWriter.AppendEvents(_fixture.Stream, ExpectedStreamVersion.Any, streamEvents.ToArray(), default); + await _fixture.IntegrationFixture.EventWriter.AppendEvents(_fixture.Stream, ExpectedStreamVersion.Any, [.. streamEvents], default); return events; } diff --git a/src/SignalR/src/Eventuous.SignalR.Client/SignalRSubscriptionClient.cs b/src/SignalR/src/Eventuous.SignalR.Client/SignalRSubscriptionClient.cs index 4329bd38a..c0a8e62b2 100644 --- a/src/SignalR/src/Eventuous.SignalR.Client/SignalRSubscriptionClient.cs +++ b/src/SignalR/src/Eventuous.SignalR.Client/SignalRSubscriptionClient.cs @@ -122,7 +122,7 @@ void OnStreamEvent(StreamEventEnvelope envelope) { void OnStreamError(StreamSubscriptionError error) { if (_subscriptions.TryRemove(error.Stream, out var state)) { - state.Channel.Writer.TryComplete(new Exception(error.Message)); + state.Channel.Writer.TryComplete(new(error.Message)); } } diff --git a/src/SignalR/test/Eventuous.Tests.SignalR/TypedStreamSubscriptionTests.cs b/src/SignalR/test/Eventuous.Tests.SignalR/TypedStreamSubscriptionTests.cs index fab3c9bbb..aa03376f4 100644 --- a/src/SignalR/test/Eventuous.Tests.SignalR/TypedStreamSubscriptionTests.cs +++ b/src/SignalR/test/Eventuous.Tests.SignalR/TypedStreamSubscriptionTests.cs @@ -16,8 +16,8 @@ static HubConnection BuildFakeConnection() [Test] public async Task On_BeforeStart_RegistersHandler() { var connection = BuildFakeConnection(); - var client = new SignalRSubscriptionClient(connection); - var sub = client.SubscribeTyped("test-stream", null); + var client = new SignalRSubscriptionClient(connection); + var sub = client.SubscribeTyped("test-stream", null); // Register two handlers before start — should not throw sub.On((_, _) => ValueTask.CompletedTask); @@ -34,8 +34,8 @@ public async Task On_AfterStart_Throws() { // Since we can't actually start without a server, we test the object // construction and pre-start handler registration. var connection = BuildFakeConnection(); - var client = new SignalRSubscriptionClient(connection); - var sub = client.SubscribeTyped("test-stream", null); + var client = new SignalRSubscriptionClient(connection); + var sub = client.SubscribeTyped("test-stream", null); // Should work fine before start await Assert.That(() => sub.On((_, _) => ValueTask.CompletedTask)).ThrowsNothing(); @@ -46,8 +46,8 @@ public async Task On_AfterStart_Throws() { [Test] public async Task OnError_CanBeChained() { var connection = BuildFakeConnection(); - var client = new SignalRSubscriptionClient(connection); - var sub = client.SubscribeTyped("test-stream", null); + var client = new SignalRSubscriptionClient(connection); + var sub = client.SubscribeTyped("test-stream", null); // Fluent chaining should work var result = sub @@ -61,8 +61,8 @@ public async Task OnError_CanBeChained() { [Test] public async Task DisposeAsync_WithoutStart_IsNoOp() { var connection = BuildFakeConnection(); - var client = new SignalRSubscriptionClient(connection); - var sub = client.SubscribeTyped("test-stream", null); + var client = new SignalRSubscriptionClient(connection); + var sub = client.SubscribeTyped("test-stream", null); // Should not throw even though StartAsync was never called await Assert.That(async () => await sub.DisposeAsync()).ThrowsNothing(); @@ -72,7 +72,7 @@ public async Task DisposeAsync_WithoutStart_IsNoOp() { [Test] public async Task SubscribeTyped_ReturnsNewInstanceEachTime() { var connection = BuildFakeConnection(); - var client = new SignalRSubscriptionClient(connection); + var client = new SignalRSubscriptionClient(connection); var sub1 = client.SubscribeTyped("stream-a", null); var sub2 = client.SubscribeTyped("stream-b", null); diff --git a/src/Sqlite/test/Eventuous.Tests.Sqlite/Store/StoreFixture.cs b/src/Sqlite/test/Eventuous.Tests.Sqlite/Store/StoreFixture.cs index a6a6c2ba4..bdb194340 100644 --- a/src/Sqlite/test/Eventuous.Tests.Sqlite/Store/StoreFixture.cs +++ b/src/Sqlite/test/Eventuous.Tests.Sqlite/Store/StoreFixture.cs @@ -4,7 +4,7 @@ namespace Eventuous.Tests.Sqlite.Store; -public sealed class StoreFixture() : SqliteStoreFixtureBase() { +public sealed class StoreFixture : SqliteStoreFixtureBase { readonly string _schemaName = GetSchemaName(); protected override void SetupServices(IServiceCollection services) { diff --git a/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/SubscribeTests.cs b/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/SubscribeTests.cs index 75473cbc1..619530851 100644 --- a/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/SubscribeTests.cs +++ b/src/Sqlite/test/Eventuous.Tests.Sqlite/Subscriptions/SubscribeTests.cs @@ -22,7 +22,7 @@ public async Task Sqlite_ShouldConsumeProducedEvents(CancellationToken cancellat var testEvents = commands.Select(ToEvent).ToList(); await Fixture.StartSubscription(); - await Fixture.Handler.AssertCollection(TimeSpan.FromSeconds(5), [..testEvents]).Validate(cancellationToken); + await Fixture.Handler.AssertCollection(TimeSpan.FromSeconds(5), [.. testEvents]).Validate(cancellationToken); await Fixture.StopSubscription(); await Assert.That(Fixture.Handler.Count).IsEqualTo(10); } @@ -76,7 +76,7 @@ public async Task Sqlite_ShouldConsumeProducedEvents(CancellationToken cancellat var testEvents = await GenerateAndProduceEvents(count); await Fixture.StartSubscription(); - await Fixture.Handler.AssertCollection(TimeSpan.FromSeconds(5), [..testEvents]).Validate(cancellationToken); + await Fixture.Handler.AssertCollection(TimeSpan.FromSeconds(5), [.. testEvents]).Validate(cancellationToken); await Fixture.StopSubscription(); await Assert.That(Fixture.Handler.Count).IsEqualTo(10); } @@ -106,7 +106,7 @@ static async Task> GenerateAndProduceEvents(int count) { var events = commands.Select(ToEvent).ToList(); var streamEvents = events.Select(x => new NewStreamEvent(Guid.NewGuid(), x, new())); - await Fixture.EventStore.AppendEvents(StreamName, ExpectedStreamVersion.Any, streamEvents.ToArray(), default); + await Fixture.EventStore.AppendEvents(StreamName, ExpectedStreamVersion.Any, [.. streamEvents], default); return events; } diff --git a/src/Testing/src/Eventuous.Testing/InMemoryEventStore.cs b/src/Testing/src/Eventuous.Testing/InMemoryEventStore.cs index b4d9097c3..67d3afe2e 100644 --- a/src/Testing/src/Eventuous.Testing/InMemoryEventStore.cs +++ b/src/Testing/src/Eventuous.Testing/InMemoryEventStore.cs @@ -55,6 +55,7 @@ public async IAsyncEnumerable ReadEvents(StreamName stream, StreamR } /// + // ReSharper disable once AsyncMethodWithoutAwait public async IAsyncEnumerable ReadEventsBackwards(StreamName stream, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) { foreach (var evt in FindStream(stream, true).GetEventsBackwards(start, count)) { yield return evt; diff --git a/test/Eventuous.TestHelpers.TUnit/TestEventListener.cs b/test/Eventuous.TestHelpers.TUnit/TestEventListener.cs index a12c4370e..808219527 100644 --- a/test/Eventuous.TestHelpers.TUnit/TestEventListener.cs +++ b/test/Eventuous.TestHelpers.TUnit/TestEventListener.cs @@ -22,7 +22,7 @@ protected override void OnEventSourceCreated(EventSource? eventSource) { #nullable disable protected override void OnEventWritten(EventWrittenEventArgs evt) { - var message = evt.Message != null && (evt.Payload?.Count ?? 0) > 0 ? string.Format(evt.Message, evt.Payload.ToArray()) : evt.Message; + var message = evt.Message != null && (evt.Payload?.Count ?? 0) > 0 ? string.Format(evt.Message, [.. evt.Payload]) : evt.Message; TestContext.Current?.OutputWriter.WriteLine($"{evt.EventSource.Name} - EventId: [{evt.EventId}], EventName: [{evt.EventName}], Message: [{message}]"); act?.Invoke(evt); } diff --git a/test/Eventuous.TestHelpers/TestHelper.cs b/test/Eventuous.TestHelpers/TestHelper.cs index 95f1c8bb4..e74879c01 100644 --- a/test/Eventuous.TestHelpers/TestHelper.cs +++ b/test/Eventuous.TestHelpers/TestHelper.cs @@ -10,15 +10,20 @@ public static class TestHelper { => GetMember(instanceType, instance, name) as TMember; static object? GetMember(Type instanceType, object instance, string name) { - const BindingFlags flags = BindingFlags.Instance - | BindingFlags.Public - | BindingFlags.NonPublic - | BindingFlags.Static; + while (true) { + const BindingFlags flags = BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Static; - var field = instanceType.GetField(name, flags); - var prop = instanceType.GetProperty(name, flags); - var member = prop?.GetValue(instance) ?? field?.GetValue(instance); + var field = instanceType.GetField(name, flags); + var prop = instanceType.GetProperty(name, flags); + var member = prop?.GetValue(instance) ?? field?.GetValue(instance); - return member == null && instanceType.BaseType != null ? GetMember(instanceType.BaseType, instance, name) : member; + if (member == null && instanceType.BaseType != null) { + instanceType = instanceType.BaseType; + + continue; + } + + return member; + } } } From bab1f59cfacaf5fe7d37a0685117921a6ac16319 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Fri, 31 Jul 2026 19:07:01 +0200 Subject: [PATCH 2/5] Fixing review findings --- .../src/Eventuous.Persistence/EventStore/StoreFunctions.cs | 2 +- src/Core/src/Eventuous.Shared/Tools/TaskRunner.cs | 2 +- .../EventSubscriptionWithCheckpoint.cs | 2 +- .../src/Eventuous.Subscriptions/Filters/ConsumePipe.cs | 2 +- src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs | 7 +++++++ .../src/Eventuous.Kafka/Producers/KafkaBasicProducer.cs | 2 +- .../Subscriptions/AllStreamSubscription.cs | 6 +++--- 7 files changed, 15 insertions(+), 8 deletions(-) diff --git a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs index 44fc54e36..ca2ee79f2 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs @@ -108,7 +108,7 @@ CancellationToken cancellationToken try { var result = new List(); - await foreach (var evt in eventReader.ReadEvents(stream, start, count, cancellationToken).ConfigureAwait(false)) { + await foreach (var evt in eventReader.ReadEvents(stream, start, count, cancellationToken).NoContext(cancellationToken)) { result.Add(evt); } diff --git a/src/Core/src/Eventuous.Shared/Tools/TaskRunner.cs b/src/Core/src/Eventuous.Shared/Tools/TaskRunner.cs index 3e822ff2c..8627cd5b8 100644 --- a/src/Core/src/Eventuous.Shared/Tools/TaskRunner.cs +++ b/src/Core/src/Eventuous.Shared/Tools/TaskRunner.cs @@ -33,7 +33,7 @@ public async ValueTask Stop(CancellationToken cancellationToken) { try { await Task.WhenAny(_runner, state.Task).NoContext(); } finally { - await registration.DisposeAsync(); + await registration.DisposeAsync().NoContext(); } // ReSharper disable once RedundantAssignment diff --git a/src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs b/src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs index 863991299..ce791b81c 100644 --- a/src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs +++ b/src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs @@ -146,6 +146,6 @@ async ValueTask DisposeCommitHandler() { var handler = CheckpointCommitHandler; CheckpointCommitHandler = null; - if (handler != null) await handler.DisposeAsync(); + if (handler != null) await handler.DisposeAsync().NoContext(); } } diff --git a/src/Core/src/Eventuous.Subscriptions/Filters/ConsumePipe.cs b/src/Core/src/Eventuous.Subscriptions/Filters/ConsumePipe.cs index 7ec62b96f..892548f60 100644 --- a/src/Core/src/Eventuous.Subscriptions/Filters/ConsumePipe.cs +++ b/src/Core/src/Eventuous.Subscriptions/Filters/ConsumePipe.cs @@ -53,7 +53,7 @@ public ConsumePipe AddFilterLast(IConsumeFilter filter) public async ValueTask DisposeAsync() { foreach (var filter in _filters) { if (filter is IAsyncDisposable d) { - await d.DisposeAsync(); + await d.DisposeAsync().NoContext(); } } } diff --git a/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs b/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs index ae8fbd857..70ace3618 100644 --- a/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs +++ b/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs @@ -11,6 +11,13 @@ public TestState() { } } +[UsedImplicitly] +file class TestEventHandler : Eventuous.Subscriptions.EventHandler { + public TestEventHandler() { + On(_ => new()); + } +} + [UsedImplicitly] file class TestAggregate : Aggregate { [UsedImplicitly] diff --git a/src/Kafka/src/Eventuous.Kafka/Producers/KafkaBasicProducer.cs b/src/Kafka/src/Eventuous.Kafka/Producers/KafkaBasicProducer.cs index 6a0d00500..708da5608 100644 --- a/src/Kafka/src/Eventuous.Kafka/Producers/KafkaBasicProducer.cs +++ b/src/Kafka/src/Eventuous.Kafka/Producers/KafkaBasicProducer.cs @@ -115,7 +115,7 @@ public async ValueTask DisposeAsync() { static async ValueTask CastAndDispose(IDisposable resource) { if (resource is IAsyncDisposable resourceAsyncDisposable) - await resourceAsyncDisposable.DisposeAsync(); + await resourceAsyncDisposable.DisposeAsync().NoContext(); else resource.Dispose(); } diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs index 043a43070..e4a68cb40 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs @@ -112,7 +112,7 @@ protected override async ValueTask Subscribe(CancellationToken cancellationToken } } catch { await messages.DisposeAsync().NoContext(); - await subscription.DisposeAsync(); + await subscription.DisposeAsync().NoContext(); throw; } @@ -205,7 +205,7 @@ CancellationToken cancellationToken // Double disposal on the unsubscribe path is fine; on the dropped path this is the only // cleanup of the underlying call before Resubscribe replaces the subscription. await messages.DisposeAsync().NoContext(); - await subscription.DisposeAsync(); + await subscription.DisposeAsync().NoContext(); } } @@ -227,7 +227,7 @@ protected override async ValueTask Unsubscribe(CancellationToken cancellationTok Stopping.Cancel(false); if (_subscription != null) - await _subscription.DisposeAsync(); + await _subscription.DisposeAsync().NoContext(); _subscription = null; if (_messagePump is { } pump) { From c4c63bc93322e2e433d1901cb1fc541a7afc59ea Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Sat, 1 Aug 2026 12:20:13 +0200 Subject: [PATCH 3/5] Updated packages, fixed RMQ migration issues, addressed warnings --- Directory.Packages.props | 83 ++++++------ .../Bookings.Payments/Registrations.cs | 5 +- samples/postgres/Bookings/Registrations.cs | 5 +- .../IsSerialisableByServiceBus.cs | 8 +- .../SendAndReceive.cs | 26 ++-- ...SubscriptionMessageProcessingBenchmarks.cs | 4 +- src/Directory.Build.props | 2 +- .../SpyglassApiTests.cs | 2 +- .../DiscoveredCommandsTests.cs | 4 +- .../Fixtures/LegacySubscriptionFixture.cs | 2 +- .../Producers/ExchangeCache.cs | 4 +- .../Producers/RabbitMqProducer.cs | 74 +++++------ .../Shared/RabbitMqExchangeOptions.cs | 2 +- .../Subscriptions/RabbitMqSubscription.cs | 118 ++++++++++-------- .../RabbitMqSubscriptionOptions.cs | 4 +- .../RabbitMqFixture.cs | 2 +- .../SignalRProducerTests.cs | 2 +- .../Eventuous.Sqlite/Eventuous.Sqlite.csproj | 2 + 18 files changed, 179 insertions(+), 170 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index 59f4734dd..11e17d15f 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -3,29 +3,29 @@ true - 10.0.0 - 10.0.0 + 10.0.10 + 10.0.10 - 9.0.10 - 9.0.10 + 9.0.18 + 10.0.10 - 8.0.21 - 9.0.10 + 8.0.29 + 10.0.10 - 4.10.0 + 4.13.0 - 10.0.1 - 0.77.3 + 10.0.3 + 1.63.0 - + - + @@ -39,23 +39,24 @@ - - - - - + + + + + - + - - - + + + + - - + + @@ -72,50 +73,50 @@ - - - + + + - - + + - + - - + + - + - - + + - + - + - + - - + + - - - + + + \ No newline at end of file diff --git a/samples/postgres/Bookings.Payments/Registrations.cs b/samples/postgres/Bookings.Payments/Registrations.cs index 78a4d48f6..0097b6c49 100644 --- a/samples/postgres/Bookings.Payments/Registrations.cs +++ b/samples/postgres/Bookings.Payments/Registrations.cs @@ -12,10 +12,7 @@ namespace Bookings.Payments; public static class Registrations { public static void AddEventuous(this IServiceCollection services, IConfiguration configuration) { - var connectionFactory = new ConnectionFactory { - Uri = new(configuration["RabbitMq:ConnectionString"]!), - DispatchConsumersAsync = true - }; + var connectionFactory = new ConnectionFactory { Uri = new(configuration["RabbitMq:ConnectionString"]!) }; services.AddSingleton(connectionFactory); services.AddEventuousPostgres(configuration.GetSection("Postgres")); services.AddEventStore(); diff --git a/samples/postgres/Bookings/Registrations.cs b/samples/postgres/Bookings/Registrations.cs index ce1f6d3dd..7c3f4ab95 100644 --- a/samples/postgres/Bookings/Registrations.cs +++ b/samples/postgres/Bookings/Registrations.cs @@ -23,10 +23,7 @@ public static void AddEventuous(this IServiceCollection services, IConfiguration new DefaultEventSerializer(new JsonSerializerOptions(JsonSerializerDefaults.Web).ConfigureForNodaTime(DateTimeZoneProviders.Tzdb)) ); - var connectionFactory = new ConnectionFactory { - Uri = new(configuration["RabbitMq:ConnectionString"]!), - DispatchConsumersAsync = true - }; + var connectionFactory = new ConnectionFactory { Uri = new(configuration["RabbitMq:ConnectionString"]!) }; services.AddSingleton(connectionFactory); diff --git a/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/IsSerialisableByServiceBus.cs b/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/IsSerialisableByServiceBus.cs index 8d03c0b0c..fdf6eab4b 100644 --- a/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/IsSerialisableByServiceBus.cs +++ b/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/IsSerialisableByServiceBus.cs @@ -3,6 +3,9 @@ namespace Eventuous.Tests.Azure.ServiceBus; public class IsSerialisableByServiceBus { + // TUnit0046 is a false positive: the data sources already return Func for test isolation, + // and the runtime invokes the Func, but the analyzer can't detect the wrapper when the test parameter is object +#pragma warning disable TUnit0046 public static IEnumerable> PassingTestData() { yield return () => "string"; yield return () => 123; @@ -34,12 +37,13 @@ public class IsSerialisableByServiceBus { yield return () => new Action(() => { }); // delegate yield return () => new WeakReference(new()); // complex type } +#pragma warning restore TUnit0046 [Test] [MethodDataSource(nameof(PassingTestData))] - public async Task Passes(object value) => await Assert.That(IsSerialisableByServiceBus(value)).IsTrue(); + public async Task Passes(object? value) => await Assert.That(IsSerialisableByServiceBus(value)).IsTrue(); [Test] [MethodDataSource(nameof(FailingTestData))] - public async Task Fails(object value) => await Assert.That(IsSerialisableByServiceBus(value)).IsFalse(); + public async Task Fails(object? value) => await Assert.That(IsSerialisableByServiceBus(value)).IsFalse(); } diff --git a/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/SendAndReceive.cs b/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/SendAndReceive.cs index da8240b62..96e8bd378 100644 --- a/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/SendAndReceive.cs +++ b/src/Azure/test/Eventuous.Tests.Azure.ServiceBus/SendAndReceive.cs @@ -7,8 +7,6 @@ namespace Eventuous.Tests.Azure.ServiceBus; [NotInParallel] [TopicAndQueueSource] public class SendAndReceive { - static CancellationToken TestCancellationToken => TestContext.Current!.CancellationToken; - ServiceBusProducer _producer = null!; ServiceBusSubscription _subscription = null!; @@ -31,31 +29,31 @@ public SendAndReceive(AzureServiceBusFixture fixture, ServiceBusProducerOptions [Test] [Retry(3)] - public async Task SingleMessage() { - await _producer.Produce(_streamName, SomeEvent.Create(), _metadata, cancellationToken: TestCancellationToken); + public async Task SingleMessage(CancellationToken cancellationToken) { + await _producer.Produce(_streamName, SomeEvent.Create(), _metadata, cancellationToken: cancellationToken); // Assert await _handler.AssertThat() .Timebox(TimeSpan.FromSeconds(5)) .Single() .Match(evt => evt is SomeEvent) - .Validate(TestCancellationToken); + .Validate(cancellationToken); } [Test] [Retry(3)] - public async Task LoadsOfMessages() { + public async Task LoadsOfMessages(CancellationToken cancellationToken) { const int count = 200; var events = Enumerable.Range(0, count).Select(SomeEvent.Create).ToList(); - await _producer.Produce(_streamName, events, _metadata, cancellationToken: TestCancellationToken); + await _producer.Produce(_streamName, events, _metadata, cancellationToken: cancellationToken); // Assert await _handler.AssertThat() .Timebox(TimeSpan.FromSeconds(20)) .Exactly(count) .Match(evt => evt is SomeEvent) - .Validate(TestCancellationToken); + .Validate(cancellationToken); var handledMessageIds = _handler.Messages .OfType() @@ -66,19 +64,19 @@ await _handler.AssertThat() } [After(Test)] - public async ValueTask CleanUpProducerAndSubscription() { - await _producer.StopAsync(TestCancellationToken); - await _subscription.Unsubscribe(_ => { }, TestCancellationToken); + public async ValueTask CleanUpProducerAndSubscription(CancellationToken cancellationToken) { + await _producer.StopAsync(cancellationToken); + await _subscription.Unsubscribe(_ => { }, cancellationToken); await _subscription.DisposeAsync(); await _producer.DisposeAsync(); } [Before(Test)] - public async Task StartProducerAndSubscription() { + public async Task StartProducerAndSubscription(CancellationToken cancellationToken) { _producer = _fixture.CreateProducer(_serviceBusProducerOptions); _subscription = _fixture.CreateSubscription(_serviceBusSubscriptionOptions, _handler, _correlationId); - await _producer.StartAsync(TestCancellationToken); - await _subscription.Subscribe(_ => { }, (_, _, _) => { }, TestCancellationToken); + await _producer.StartAsync(cancellationToken); + await _subscription.Subscribe(_ => { }, (_, _, _) => { }, cancellationToken); } } diff --git a/src/Benchmarks/Benchmarks/SubscriptionMessageProcessingBenchmarks.cs b/src/Benchmarks/Benchmarks/SubscriptionMessageProcessingBenchmarks.cs index acb42de17..7ea8791a6 100644 --- a/src/Benchmarks/Benchmarks/SubscriptionMessageProcessingBenchmarks.cs +++ b/src/Benchmarks/Benchmarks/SubscriptionMessageProcessingBenchmarks.cs @@ -1,4 +1,5 @@ using BenchmarkDotNet.Attributes; +using Eventuous; using Eventuous.Subscriptions; using Eventuous.Subscriptions.Context; @@ -93,7 +94,8 @@ public async Task CreateContextAndProcess() { await _handler.HandleEvent(ctx); } - class TestEvent { + [EventType("benchmark-test-event")] + internal class TestEvent { public string Value { get; set; } = string.Empty; } diff --git a/src/Directory.Build.props b/src/Directory.Build.props index 5ef0d53ac..47de60401 100644 --- a/src/Directory.Build.props +++ b/src/Directory.Build.props @@ -75,7 +75,7 @@ - + $(MinVerMajor).$(MinVerMinor).$(MinVerPatch) $(MinVerMajor).$(MinVerMinor).$(MinVerPatch) diff --git a/src/Experimental/test/Eventuous.Tests.Spyglass/SpyglassApiTests.cs b/src/Experimental/test/Eventuous.Tests.Spyglass/SpyglassApiTests.cs index 716d15e28..ebe50c603 100644 --- a/src/Experimental/test/Eventuous.Tests.Spyglass/SpyglassApiTests.cs +++ b/src/Experimental/test/Eventuous.Tests.Spyglass/SpyglassApiTests.cs @@ -143,7 +143,7 @@ await eventStore.AppendEvents( // State should reflect the loaded events var state = root.GetProperty("state"); await Assert.That(state.GetProperty("guestId").GetString()).IsEqualTo("guest-1"); - await Assert.That(state.GetProperty("paid").GetBoolean()).IsEqualTo(false); + await Assert.That(state.GetProperty("paid").GetBoolean()).IsFalse(); // Events should contain both events var events = root.GetProperty("events"); diff --git a/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore/DiscoveredCommandsTests.cs b/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore/DiscoveredCommandsTests.cs index 3ddbfad32..781d68dba 100644 --- a/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore/DiscoveredCommandsTests.cs +++ b/src/Extensions/test/Eventuous.Tests.Extensions.AspNetCore/DiscoveredCommandsTests.cs @@ -29,7 +29,7 @@ public async Task RegisterStatesCommands() { await using var app = builder.Build(); - var b = app.MapDiscoveredCommands(typeof(TestCommands.DuplicateCommand)); + var b = app.MapDiscoveredCommands([typeof(TestCommands.DuplicateCommand)]); IEnumerable actual = [.. b.DataSources.First().Endpoints.Select(x => x.DisplayName).Order()!]; @@ -43,7 +43,7 @@ public async Task CallDiscoveredCommandRoute() { var fixture = new ServerFixture( factory, _ => { }, - app => app.MapDiscoveredCommands(typeof(TestCommands.ImportBookingHttp3)) + app => app.MapDiscoveredCommands([typeof(TestCommands.ImportBookingHttp3)]) ); var cmd = ServerFixture.GetNestedBookRoom(new DateTime(2023, 10, 1)); diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/Fixtures/LegacySubscriptionFixture.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/Fixtures/LegacySubscriptionFixture.cs index 592ab9db3..52d50d7f6 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/Fixtures/LegacySubscriptionFixture.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/Fixtures/LegacySubscriptionFixture.cs @@ -9,7 +9,7 @@ namespace Eventuous.Tests.KurrentDB.Subscriptions.Fixtures; public abstract class LegacySubscriptionFixture : IAsyncInitializer, IAsyncDisposable where T : class, IEventHandler { protected StreamName Stream { get; } = new($"test-{Guid.NewGuid():N}"); - protected StoreFixture StoreFixture { get; } + public StoreFixture StoreFixture { get; } protected T Handler { get; } protected KurrentDBProducer Producer { get; private set; } = null!; protected ILogger Log { get; } diff --git a/src/RabbitMq/src/Eventuous.RabbitMq/Producers/ExchangeCache.cs b/src/RabbitMq/src/Eventuous.RabbitMq/Producers/ExchangeCache.cs index 5e3cace99..42592b94c 100644 --- a/src/RabbitMq/src/Eventuous.RabbitMq/Producers/ExchangeCache.cs +++ b/src/RabbitMq/src/Eventuous.RabbitMq/Producers/ExchangeCache.cs @@ -6,12 +6,12 @@ namespace Eventuous.RabbitMq.Producers; class ExchangeCache(ILogger? log) { - public void EnsureExchange(string name, Action createExchange) { + public async Task EnsureExchange(string name, Func createExchange) { if (_exchanges.Contains(name)) return; try { log?.LogInformation("Ensuring exchange {ExchangeName}", name); - createExchange(); + await createExchange().NoContext(); } catch (Exception e) { log?.LogError(e, "Failed to ensure exchange {ExchangeName}: {ErrorMessage}", name, e.Message); diff --git a/src/RabbitMq/src/Eventuous.RabbitMq/Producers/RabbitMqProducer.cs b/src/RabbitMq/src/Eventuous.RabbitMq/Producers/RabbitMqProducer.cs index 275d62840..a4a7562bf 100644 --- a/src/RabbitMq/src/Eventuous.RabbitMq/Producers/RabbitMqProducer.cs +++ b/src/RabbitMq/src/Eventuous.RabbitMq/Producers/RabbitMqProducer.cs @@ -22,7 +22,7 @@ public class RabbitMqProducer : BaseProducer, IHostedPro readonly ExchangeCache _exchangeCache; IConnection? _connection; - IModel? _channel; + IChannel? _channel; /// /// Creates a RabbitMQ producer instance @@ -45,13 +45,11 @@ public RabbitMqProducer( _exchangeCache = new(_log); } - public Task StartAsync(CancellationToken cancellationToken = default) { - _connection = _connectionFactory.CreateConnection(); - _channel = _connection.CreateModel(); - _channel.ConfirmSelect(); - Ready = true; - - return Task.CompletedTask; + public async Task StartAsync(CancellationToken cancellationToken = default) { + var channelOptions = new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true); + _connection = await _connectionFactory.CreateConnectionAsync(cancellationToken).NoContext(); + _channel = await _connection.CreateChannelAsync(channelOptions, cancellationToken).NoContext(); + Ready = true; } static readonly ProducerTracingOptions TracingOptions = new() { @@ -66,17 +64,22 @@ protected override async Task ProduceMessages( RabbitMqProduceOptions? options, CancellationToken cancellationToken = default ) { - EnsureExchange(stream); + await EnsureExchange(stream, cancellationToken).NoContext(); var produced = new List(); var failed = new List<(ProducedMessage Msg, Exception Ex)>(); + var pending = new List<(ProducedMessage Msg, Task Publish)>(); foreach (var message in messages) { if (Activity.Current is { IsAllDataRequested: true }) { Activity.Current.SetTag(RabbitMqTelemetryTags.RoutingKey, options?.RoutingKey); } + pending.Add((message, Publish(stream, message, options, cancellationToken))); + } + + foreach (var (message, publish) in pending) { try { - Publish(stream, message, options); + await publish.NoContext(); produced.Add(message); } catch (Exception e) { _log?.LogError(e, "Failed to produce message to RabbitMQ"); @@ -84,8 +87,6 @@ protected override async Task ProduceMessages( } } - await Confirm(cancellationToken).NoContext(); - await produced.Select(x => x.Ack()).WhenAll().NoContext(); await failed @@ -94,19 +95,21 @@ await failed .NoContext(); } - void Publish(string stream, ProducedMessage message, RabbitMqProduceOptions? options) { + async Task Publish(string stream, ProducedMessage message, RabbitMqProduceOptions? options, CancellationToken cancellationToken) { if (_channel == null) throw new InvalidOperationException("Producer hasn't been initialized, call Initialize"); var (msg, metadata) = (message.Message, message.Metadata); var (eventType, contentType, payload) = _serializer.SerializeEvent(msg); SetActivityMessageType(eventType); - var prop = _channel.CreateBasicProperties(); - prop.ContentType = contentType; - prop.Persistent = options?.Persisted != false; - prop.Type = eventType; - prop.CorrelationId = metadata!.GetCorrelationId(); - prop.MessageId = message.MessageId.ToString(); + + var prop = new BasicProperties { + ContentType = contentType, + Persistent = options?.Persisted != false, + Type = eventType, + CorrelationId = metadata!.GetCorrelationId(), + MessageId = message.MessageId.ToString() + }; metadata!.Remove(MetaTags.MessageId); prop.Headers = metadata.ToDictionary(x => x.Key, x => x.Value); @@ -118,38 +121,35 @@ void Publish(string stream, ProducedMessage message, RabbitMqProduceOptions? opt prop.ReplyTo = options.ReplyTo; } - _channel.BasicPublish(stream, options?.RoutingKey ?? "", true, prop, payload); + await _channel.BasicPublishAsync(stream, options?.RoutingKey ?? "", true, prop, payload, cancellationToken).NoContext(); } - void EnsureExchange(string exchange) + Task EnsureExchange(string exchange, CancellationToken cancellationToken) => _exchangeCache.EnsureExchange( exchange, () => - _channel!.ExchangeDeclare( + _channel!.ExchangeDeclareAsync( exchange, _options?.Type ?? ExchangeType.Fanout, _options?.Durable ?? true, _options?.AutoDelete ?? false, - _options?.Arguments + _options?.Arguments, + cancellationToken: cancellationToken ) ); - async Task Confirm(CancellationToken cancellationToken) { - while (!_channel!.WaitForConfirms(ConfirmTimeout) && !cancellationToken.IsCancellationRequested) { - await Task.Delay(ConfirmIdle, cancellationToken).NoContext(); + public async Task StopAsync(CancellationToken cancellationToken = default) { + if (_channel != null) { + await _channel.CloseAsync(cancellationToken).NoContext(); + _channel.Dispose(); + _channel = null; } - } - - static readonly TimeSpan ConfirmTimeout = TimeSpan.FromSeconds(1); - static readonly TimeSpan ConfirmIdle = TimeSpan.FromMilliseconds(100); - public Task StopAsync(CancellationToken cancellationToken = default) { - _channel?.Close(); - _channel?.Dispose(); - _connection?.Close(); - _connection?.Dispose(); - - return Task.CompletedTask; + if (_connection != null) { + await _connection.CloseAsync(cancellationToken: cancellationToken).NoContext(); + _connection.Dispose(); + _connection = null; + } } public bool Ready { get; private set; } diff --git a/src/RabbitMq/src/Eventuous.RabbitMq/Shared/RabbitMqExchangeOptions.cs b/src/RabbitMq/src/Eventuous.RabbitMq/Shared/RabbitMqExchangeOptions.cs index d426d2881..ff046ce70 100644 --- a/src/RabbitMq/src/Eventuous.RabbitMq/Shared/RabbitMqExchangeOptions.cs +++ b/src/RabbitMq/src/Eventuous.RabbitMq/Shared/RabbitMqExchangeOptions.cs @@ -8,5 +8,5 @@ public class RabbitMqExchangeOptions { public bool Durable { get; init; } = true; public bool AutoDelete { get; init; } - public IDictionary? Arguments { get; init; } + public IDictionary Arguments { get; init; } = new Dictionary(); } \ No newline at end of file diff --git a/src/RabbitMq/src/Eventuous.RabbitMq/Subscriptions/RabbitMqSubscription.cs b/src/RabbitMq/src/Eventuous.RabbitMq/Subscriptions/RabbitMqSubscription.cs index 1b2bd7379..de9f4e7ef 100644 --- a/src/RabbitMq/src/Eventuous.RabbitMq/Subscriptions/RabbitMqSubscription.cs +++ b/src/RabbitMq/src/Eventuous.RabbitMq/Subscriptions/RabbitMqSubscription.cs @@ -15,11 +15,13 @@ namespace Eventuous.RabbitMq.Subscriptions; /// [PublicAPI] public class RabbitMqSubscription : EventSubscription { - public delegate void HandleEventProcessingFailure(IModel channel, BasicDeliverEventArgs message, Exception? exception); + public delegate ValueTask HandleEventProcessingFailure(IChannel channel, BasicDeliverEventArgs message, Exception? exception); readonly HandleEventProcessingFailure _failureHandler; - readonly IConnection _connection; - readonly IModel _channel; + readonly ConnectionFactory _connectionFactory; + + IConnection? _connection; + IChannel? _channel; /// /// Creates RabbitMQ subscription service instance @@ -58,13 +60,8 @@ public RabbitMqSubscription( loggerFactory, eventSerializer ) { - _failureHandler = options.FailureHandler ?? DefaultEventFailureHandler; - _connection = Ensure.NotNull(connectionFactory).CreateConnection(); - _channel = _connection.CreateModel(); - - var prefetch = options.PrefetchCount > 0 ? options.PrefetchCount : options.ConcurrencyLimit * 2; - - _channel.BasicQos(0, (ushort)prefetch, false); + _failureHandler = options.FailureHandler ?? DefaultEventFailureHandler; + _connectionFactory = Ensure.NotNull(connectionFactory); if (options is { FailureHandler: not null, ThrowOnError: false }) Log.ThrowOnErrorIncompatible(); } @@ -93,7 +90,13 @@ public RabbitMqSubscription( eventSerializer ) { } - protected override ValueTask Subscribe(CancellationToken cancellationToken) { + protected override async ValueTask Subscribe(CancellationToken cancellationToken) { + _connection = await _connectionFactory.CreateConnectionAsync(cancellationToken).NoContext(); + _channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken).NoContext(); + + var prefetch = Options.PrefetchCount > 0 ? Options.PrefetchCount : Options.ConcurrencyLimit * 2; + await _channel.BasicQosAsync(0, (ushort)prefetch, false, cancellationToken).NoContext(); + var exchange = Ensure.NotEmptyString(Options.Exchange); Log.InfoLog?.Log("Ensuring exchange {Exchange}", exchange); @@ -102,40 +105,44 @@ protected override ValueTask Subscribe(CancellationToken cancellationToken) { Log.WarnLog?.Log("Fan-out exchange doesn't support routing keys"); } - _channel.ExchangeDeclare( - exchange, - Options.ExchangeOptions.Type, - Options.ExchangeOptions.Durable, - Options.ExchangeOptions.AutoDelete, - Options.ExchangeOptions.Arguments - ); + await _channel.ExchangeDeclareAsync( + exchange, + Options.ExchangeOptions.Type, + Options.ExchangeOptions.Durable, + Options.ExchangeOptions.AutoDelete, + Options.ExchangeOptions.Arguments, + cancellationToken: cancellationToken + ) + .NoContext(); var queue = Options.QueueOptions.Queue ?? Options.SubscriptionId; Log.InfoLog?.Log("Ensuring queue {Queue}", queue); - _channel.QueueDeclare( - queue, - Options.QueueOptions.Durable, - Options.QueueOptions.Exclusive, - Options.QueueOptions.AutoDelete, - Options.QueueOptions.Arguments - ); + await _channel.QueueDeclareAsync( + queue, + Options.QueueOptions.Durable, + Options.QueueOptions.Exclusive, + Options.QueueOptions.AutoDelete, + Options.QueueOptions.Arguments, + cancellationToken: cancellationToken + ) + .NoContext(); Log.InfoLog?.Log("Binding exchange {Exchange} to queue {Queue}", exchange, queue); - _channel.QueueBind( - queue, - exchange, - Options.BindingOptions.RoutingKey, - Options.BindingOptions.Arguments - ); + await _channel.QueueBindAsync( + queue, + exchange, + Options.BindingOptions.RoutingKey, + Options.BindingOptions.Arguments, + cancellationToken: cancellationToken + ) + .NoContext(); var consumer = new AsyncEventingBasicConsumer(_channel); - consumer.Received += HandleReceived; - - _channel.BasicConsume(consumer, queue); + consumer.ReceivedAsync += HandleReceived; - return default; + await _channel.BasicConsumeAsync(queue, false, consumer, cancellationToken).NoContext(); } const string ReceivedMessageKey = "receivedMessage"; @@ -152,33 +159,29 @@ async Task HandleReceived(object sender, BasicDeliverEventArgs received) { } } - ValueTask Ack(IMessageConsumeContext ctx) { + async ValueTask Ack(IMessageConsumeContext ctx) { var received = ctx.Items.GetItem(ReceivedMessageKey)!; - _channel.BasicAck(received.DeliveryTag, false); - - return default; + await _channel!.BasicAckAsync(received.DeliveryTag, false).NoContext(); } - ValueTask Nack(IMessageConsumeContext ctx, Exception exception) { + async ValueTask Nack(IMessageConsumeContext ctx, Exception exception) { if (Options.ThrowOnError) throw exception; var received = ctx.Items.GetItem(ReceivedMessageKey)!; - _failureHandler(_channel, received, exception); - - return default; + await _failureHandler(_channel!, received, exception).NoContext(); } MessageConsumeContext CreateContext(object sender, BasicDeliverEventArgs received) { - var evt = DeserializeData(received.BasicProperties.ContentType, received.BasicProperties.Type, received.Body, received.Exchange); + var evt = DeserializeData(received.BasicProperties.ContentType!, received.BasicProperties.Type!, received.Body, received.Exchange); var meta = received.BasicProperties.Headers != null ? new Metadata(received.BasicProperties.Headers.ToDictionary(x => x.Key, x => x.Value)!) : null; return new( - received.BasicProperties.MessageId, - received.BasicProperties.Type, - received.BasicProperties.ContentType, + received.BasicProperties.MessageId!, + received.BasicProperties.Type!, + received.BasicProperties.ContentType!, received.Exchange, 0, 0, @@ -192,18 +195,23 @@ MessageConsumeContext CreateContext(object sender, BasicDeliverEventArgs receive ); } - protected override ValueTask Unsubscribe(CancellationToken cancellationToken) { - _channel.Close(); - _channel.Dispose(); - _connection.Close(); - _connection.Dispose(); + protected override async ValueTask Unsubscribe(CancellationToken cancellationToken) { + if (_channel != null) { + await _channel.CloseAsync(cancellationToken).NoContext(); + _channel.Dispose(); + _channel = null; + } - return default; + if (_connection != null) { + await _connection.CloseAsync(cancellationToken: cancellationToken).NoContext(); + _connection.Dispose(); + _connection = null; + } } - void DefaultEventFailureHandler(IModel channel, BasicDeliverEventArgs message, Exception? exception) { + async ValueTask DefaultEventFailureHandler(IChannel channel, BasicDeliverEventArgs message, Exception? exception) { Log.WarnLog?.Log("Error in the consumer, will redeliver", exception?.ToString() ?? "Unknown error"); - _channel.BasicReject(message.DeliveryTag, true); + await channel.BasicRejectAsync(message.DeliveryTag, true).NoContext(); } record Event(BasicDeliverEventArgs Original, IMessageConsumeContext Context); diff --git a/src/RabbitMq/src/Eventuous.RabbitMq/Subscriptions/RabbitMqSubscriptionOptions.cs b/src/RabbitMq/src/Eventuous.RabbitMq/Subscriptions/RabbitMqSubscriptionOptions.cs index 574cf4b84..4690b2bbf 100644 --- a/src/RabbitMq/src/Eventuous.RabbitMq/Subscriptions/RabbitMqSubscriptionOptions.cs +++ b/src/RabbitMq/src/Eventuous.RabbitMq/Subscriptions/RabbitMqSubscriptionOptions.cs @@ -55,13 +55,13 @@ public record RabbitMqQueueOptions { public bool Exclusive { get; set; } public bool AutoDelete { get; set; } - public IDictionary? Arguments { get; set; } + public IDictionary Arguments { get; set; } = new Dictionary(); } [PublicAPI] public record RabbitMqBindingOptions { public string RoutingKey { get; set; } = ""; - public IDictionary? Arguments { get; set; } + public IDictionary Arguments { get; set; } = new Dictionary(); } } diff --git a/src/RabbitMq/test/Eventuous.Tests.RabbitMq/RabbitMqFixture.cs b/src/RabbitMq/test/Eventuous.Tests.RabbitMq/RabbitMqFixture.cs index a14c2b8a2..18e9b66f9 100644 --- a/src/RabbitMq/test/Eventuous.Tests.RabbitMq/RabbitMqFixture.cs +++ b/src/RabbitMq/test/Eventuous.Tests.RabbitMq/RabbitMqFixture.cs @@ -11,7 +11,7 @@ public class RabbitMqFixture : IAsyncInitializer, IAsyncDisposable { public async Task InitializeAsync() { _rabbitMq = new RabbitMqBuilder().Build(); await _rabbitMq.StartAsync(); - ConnectionFactory = new() { Uri = new(_rabbitMq.GetConnectionString()), DispatchConsumersAsync = true }; + ConnectionFactory = new() { Uri = new(_rabbitMq.GetConnectionString()) }; } public async ValueTask DisposeAsync() => await _rabbitMq.DisposeAsync(); diff --git a/src/SignalR/test/Eventuous.Tests.SignalR/SignalRProducerTests.cs b/src/SignalR/test/Eventuous.Tests.SignalR/SignalRProducerTests.cs index f3e85909e..25fdfd2ef 100644 --- a/src/SignalR/test/Eventuous.Tests.SignalR/SignalRProducerTests.cs +++ b/src/SignalR/test/Eventuous.Tests.SignalR/SignalRProducerTests.cs @@ -33,7 +33,7 @@ public async Task ProduceMessages_SendsEnvelopeToCorrectConnection() { await clientProxy.Received(1) .SendCoreAsync( SignalRSubscriptionMethods.StreamEvent, - Arg.Is(args => args.Length == 1 && args[0] is StreamEventEnvelope), + Arg.Is(args => args != null && args.Length == 1 && args[0] is StreamEventEnvelope), Arg.Any() ) .ConfigureAwait(false); diff --git a/src/Sqlite/src/Eventuous.Sqlite/Eventuous.Sqlite.csproj b/src/Sqlite/src/Eventuous.Sqlite/Eventuous.Sqlite.csproj index 35715748b..2d2d57dca 100644 --- a/src/Sqlite/src/Eventuous.Sqlite/Eventuous.Sqlite.csproj +++ b/src/Sqlite/src/Eventuous.Sqlite/Eventuous.Sqlite.csproj @@ -7,6 +7,8 @@ + + From 7e95530cb5a575eed884b139cd09d13e86364121 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Mon, 3 Aug 2026 15:55:53 +0200 Subject: [PATCH 4/5] fix: repair test suites broken by the package upgrades Three deterministic failures surfaced by the dependency bumps: Bump Verify.TUnit to 31.28.0. The 31.0.5 build was compiled against the old TUnit and threw MethodAccessException on TestContext.TestDetails after the TUnit 0.77 -> 1.63 upgrade, failing the AspNetCore extension tests. Allow admin commands on the Redis test connection. The teardown flushes the database with a raw FLUSHDB through Execute, and StackExchange.Redis 3.x enforces the admin gate for raw commands too. Only trust Activity.Current as the consumer span parent when Eventuous created it. TUnit 1.63 wraps every test in its own activity, which the subscription captured as the parent instead of the remote context from the message metadata, breaking ShouldPropagateRemoteContext. The same hijack applies to any foreign ambient activity in production, e.g. a client library's delivery span, so the metadata now wins over foreign activities, with Activity.Current kept as the last resort for messages that carry no tracing metadata. Co-Authored-By: Claude Fable 5 --- Directory.Packages.props | 2 +- .../Diagnostics/SubscriptionActivity.cs | 13 +++++++++++-- .../Fixtures/IntegrationFixture.cs | 4 +++- 3 files changed, 15 insertions(+), 4 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index 11e17d15f..a5a221986 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -94,7 +94,7 @@ - + diff --git a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionActivity.cs b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionActivity.cs index 5ec2655c1..53dde9ca9 100644 --- a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionActivity.cs +++ b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionActivity.cs @@ -43,7 +43,14 @@ static class SubscriptionActivity { } static ActivityContext? GetParentContext(IBaseConsumeContext context) { - if (Activity.Current != null) return Activity.Current.Context; + // The current activity is only trusted as a parent when Eventuous created it: that's the + // message's own pipeline activity (handler, then filters nested under it). A foreign ambient + // activity — a test framework's per-test span, a client library's delivery span — must not + // override the remote context propagated in the message metadata, or the consumer span gets + // detached from the producer trace. + if (Activity.Current?.Source.Name.StartsWith(EventuousDiagnostics.InstrumentationName, StringComparison.Ordinal) == true) { + return Activity.Current.Context; + } if (context.Items.TryGetItem(ContextItemKeys.Activity, out var parentActivity)) { return parentActivity?.Context; @@ -51,7 +58,9 @@ static class SubscriptionActivity { var tracingData = context.Metadata?.GetTracingMeta(); - return tracingData?.ToActivityContext(true); + if (tracingData?.ToActivityContext(true) is { } remoteContext) return remoteContext; + + return Activity.Current?.Context; } public static Activity? Create( diff --git a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs index 4e0ec7ea5..e237509d3 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs @@ -35,7 +35,9 @@ public async Task InitializeAsync() { return; IDatabase GetDb() { - var muxer = ConnectionMultiplexer.Connect(connString); + // FLUSHDB in test teardown is an admin command; StackExchange.Redis 3.x enforces the + // admin gate for raw commands issued through Execute as well. + var muxer = ConnectionMultiplexer.Connect($"{connString},allowAdmin=true"); return muxer.GetDatabase(); } From 962ccb08ead59a4d4a7fef3d2b62b32621ae0c6f Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Mon, 3 Aug 2026 16:12:12 +0200 Subject: [PATCH 5/5] fix: address Copilot review comments Make the exchange cache safe for concurrent producers: the HashSet was unsynchronized and the async EnsureExchange widened the race across the await. Duplicate declarations are idempotent on the broker, so the cache is a ConcurrentDictionary without a gate, and failed declarations stay out of it to be retried. Use OfType instead of Where with a null-forgiving operator when loading aggregate events, drop an unused local in the amendment test, and rename the Spyglass registry lock field so it doesn't collide with the Lock type. Co-Authored-By: Claude Fable 5 --- src/Core/src/Eventuous.Domain/Aggregate.cs | 2 +- .../ServiceTestBase.Amendments.cs | 2 +- .../src/Eventuous.Spyglass/SpyglassRegistry.cs | 4 ++-- .../src/Eventuous.RabbitMq/Producers/ExchangeCache.cs | 10 +++++++--- 4 files changed, 11 insertions(+), 7 deletions(-) diff --git a/src/Core/src/Eventuous.Domain/Aggregate.cs b/src/Core/src/Eventuous.Domain/Aggregate.cs index 10c0fd519..5548d5acc 100644 --- a/src/Core/src/Eventuous.Domain/Aggregate.cs +++ b/src/Core/src/Eventuous.Domain/Aggregate.cs @@ -79,7 +79,7 @@ protected void EnsureExists(Func? getException = null) { } public void Load(long version, IEnumerable events) { - Original = [.. events.Where(x => x != null)!]; + Original = [.. events.OfType()]; OriginalVersion = version; State = Original.Aggregate(State, Fold); diff --git a/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.Amendments.cs b/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.Amendments.cs index 13c6af7b2..060be14f2 100644 --- a/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.Amendments.cs +++ b/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.Amendments.cs @@ -9,7 +9,7 @@ public async Task Should_amend_event_from_command(CancellationToken cancellation var service = CreateService(amendEvent: AmendEvent); var cmd = CreateCommand(); - var dummy = await service.Handle(cmd, cancellationToken); + await service.Handle(cmd, cancellationToken); var stream = await Store.ReadStream(StreamName.For(cmd.BookingId), StreamReadPosition.Start, cancellationToken: cancellationToken); await Assert.That(stream[0].Metadata["userId"]).IsEqualTo(cmd.ImportedBy); diff --git a/src/Experimental/src/Eventuous.Spyglass/SpyglassRegistry.cs b/src/Experimental/src/Eventuous.Spyglass/SpyglassRegistry.cs index 84aa37c2c..2bed5f1c6 100644 --- a/src/Experimental/src/Eventuous.Spyglass/SpyglassRegistry.cs +++ b/src/Experimental/src/Eventuous.Spyglass/SpyglassRegistry.cs @@ -33,10 +33,10 @@ public static class SpyglassRegistry { // are loaded by parallel test fixtures (or any parallel host startup), so the backing store has to // be thread-safe. Reads also enumerate the snapshot, so we publish a fresh array on every write. static SpyglassAggregateInfo[] aggregates = []; - static readonly Lock Lock = new(); + static readonly Lock SyncRoot = new(); public static void Register(SpyglassAggregateInfo info) { - lock (Lock) { + lock (SyncRoot) { var entry = info with { Id = Guid.NewGuid() }; var next = new SpyglassAggregateInfo[aggregates.Length + 1]; Array.Copy(aggregates, next, aggregates.Length); diff --git a/src/RabbitMq/src/Eventuous.RabbitMq/Producers/ExchangeCache.cs b/src/RabbitMq/src/Eventuous.RabbitMq/Producers/ExchangeCache.cs index 42592b94c..e0184a453 100644 --- a/src/RabbitMq/src/Eventuous.RabbitMq/Producers/ExchangeCache.cs +++ b/src/RabbitMq/src/Eventuous.RabbitMq/Producers/ExchangeCache.cs @@ -1,13 +1,14 @@ // Copyright (C) Eventuous HQ OÜ. All rights reserved // Licensed under the Apache License, Version 2.0. +using System.Collections.Concurrent; using Microsoft.Extensions.Logging; namespace Eventuous.RabbitMq.Producers; class ExchangeCache(ILogger? log) { public async Task EnsureExchange(string name, Func createExchange) { - if (_exchanges.Contains(name)) return; + if (_exchanges.ContainsKey(name)) return; try { log?.LogInformation("Ensuring exchange {ExchangeName}", name); @@ -18,8 +19,11 @@ public async Task EnsureExchange(string name, Func createExchange) { throw; } - _exchanges.Add(name); + // Concurrent producers can race here and declare the same exchange more than once. The + // declaration is idempotent on the broker (identical arguments), so the race only costs an + // extra round trip, while a failed declaration stays out of the cache and is retried. + _exchanges.TryAdd(name, true); } - readonly HashSet _exchanges = []; + readonly ConcurrentDictionary _exchanges = new(); }