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/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/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/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/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/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/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/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..ca2ee79f2 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).NoContext(cancellationToken)) { + 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); @@ -37,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.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/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/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..70ace3618 100644 --- a/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs +++ b/src/Core/test/Eventuous.Tests.Shared.Analyzers/Analyzed.cs @@ -12,17 +12,18 @@ public TestState() { } [UsedImplicitly] -file class TestAggregate : Aggregate { - [UsedImplicitly] - 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()); + On(_ => new()); } } +[UsedImplicitly] +file class TestAggregate : Aggregate { + [UsedImplicitly] + public void Process() => Apply(new Events.RoomBooked("1", DateTime.Now, DateTime.Now.AddDays(1), 100)); +} + 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/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/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/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/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..781d68dba 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" }; @@ -29,9 +29,9 @@ 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().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" }; @@ -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/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/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..e4a68cb40 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().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(); - subscription.Dispose(); + await subscription.DisposeAsync().NoContext(); } } @@ -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().NoContext(); _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/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/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/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/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/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/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/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 @@ + + 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; + } } }