Draft: TUnit Example

Directory.Packages.props

@@ -39,10 +39,6 @@
     <PackageVersion Include="Confluent.Kafka" Version="2.14.0" />
     <PackageVersion Include="Confluent.SchemaRegistry" Version="2.14.0" />
     <PackageVersion Include="Confluent.SchemaRegistry.Serdes.Json" Version="2.14.0" />
-    <PackageVersion Include="coverlet.collector" Version="10.0.0">
-      <PrivateAssets>all</PrivateAssets>
-      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
-    </PackageVersion>
     <PackageVersion Include="Dapper" Version="2.1.72" />
     <PackageVersion Include="Dapper.Contrib" Version="2.0.78" />
     <PackageVersion Include="DapperExtensions" Version="1.7.0" />
@@ -55,7 +51,6 @@
     <PackageVersion Include="FluentMigrator.Runner.SQLite" Version="8.0.1" />
     <PackageVersion Include="FluentMigrator.Runner.SqlServer" Version="8.0.1" />
     <PackageVersion Include="Fluid.Core" Version="2.31.0" />
-    <PackageVersion Include="GitHubActionsTestLogger" Version="3.0.4" />
     <PackageVersion Include="Google.Cloud.Firestore.V1" Version="4.2.0" />
     <PackageVersion Include="Google.Cloud.PubSub.V1" Version="3.34.0" />
     <PackageVersion Include="Google.Cloud.ResourceManager.V3" Version="2.6.0" />
@@ -100,7 +95,6 @@
     <PackageVersion Include="Microsoft.Extensions.Options" Version="10.0.7" />
     <PackageVersion Include="Microsoft.Extensions.ServiceDiscovery" Version="10.5.0" />
     <PackageVersion Include="Microsoft.Extensions.TimeProvider.Testing" Version="10.5.0" />
-    <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="18.5.1" />
     <PackageVersion Include="MongoDB.Driver" Version="3.8.1" />
     <PackageVersion Include="MQTTnet" Version="4.3.7.1207" />
     <PackageVersion Include="MySqlConnector" Version="2.5.0" />
@@ -109,12 +103,6 @@
     <PackageVersion Include="Neuroglia.AsyncApi.IO" Version="3.0.6" />
     <PackageVersion Include="NJsonSchema" Version="11.6.1" />
     <PackageVersion Include="NJsonSchema.NewtonsoftJson" Version="11.6.1" />
-    <PackageVersion Include="NUnit" Version="4.6.0" />
-    <PackageVersion Include="NUnit.Analyzers" Version="4.13.0">
-      <PrivateAssets>all</PrivateAssets>
-      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
-    </PackageVersion>
-    <PackageVersion Include="NUnit3TestAdapter" Version="6.2.0" />
     <PackageVersion Include="OpenTelemetry" Version="1.15.3" />
     <PackageVersion Include="OpenTelemetry.Instrumentation.Process" Version="1.14.0-beta.2" />
     <PackageVersion Include="OpenTelemetry.Api.ProviderBuilderExtensions" Version="1.15.3" />
@@ -160,16 +148,10 @@
     <PackageVersion Include="System.Text.Json" Version="10.0.7" />
     <PackageVersion Include="System.Text.RegularExpressions" Version="4.3.1" />
     <PackageVersion Include="System.Threading.Channels" Version="10.0.7" />
-    <PackageVersion Include="TUnit" Version="0.67.10" />
+    <PackageVersion Include="TUnit" Version="1.40.0" />
+    <PackageVersion Include="Microsoft.Testing.Extensions.HangDump" Version="2.2.1" />
     <PackageVersion Include="YamlDotNet" Version="16.3.0" />
-    <PackageVersion Include="xunit" Version="2.9.3" />
-    <PackageVersion Include="xunit.runner.visualstudio" Version="3.1.5">
-      <PrivateAssets>all</PrivateAssets>
-      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
-    </PackageVersion>
     <PackageVersion Include="FakeItEasy" Version="9.0.1" />
-    <PackageVersion Include="xunit.v3" Version="3.1.0" />
-    <PackageVersion Include="xunit.v3.runner.msbuild" Version="3.1.0" />
   </ItemGroup>
   <ItemGroup Condition=" '$(TargetFramework)' == 'net8.0' ">
     <PackageVersion Include="Microsoft.EntityFrameworkCore" Version="8.0.22" />
@@ -221,4 +203,4 @@
       <IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
     </GlobalPackageReference>
   </ItemGroup>
-</Project>
+</Project>

global.json

@@ -0,0 +1,5 @@
+{
+  "test": {
+    "runner": "Microsoft.Testing.Platform"
+  }
+}

samples/TaskQueue/AWSTaskQueue/GreetingsSender/Program.cs

@@ -64,7 +64,7 @@ static async Task Main(string[] args)
                     }
                 });
 
-                var producerRegistry = new SnsProducerRegistryFactory(
+                var producerRegistry = await new SnsProducerRegistryFactory(
                     awsConnection,
                     [
                         new SnsPublication<GreetingEvent>
@@ -80,7 +80,7 @@ static async Task Main(string[] args)
                             }
                         }
                     ]
-                ).Create();
+                ).CreateAsync();
 
                 serviceCollection
                     .AddBrighter()

src/Paramore.Brighter.MessageScheduler.Hangfire/HangfireMessageSchedulerFactory.cs

@@ -12,10 +12,17 @@ public class HangfireMessageSchedulerFactory : IAmAMessageSchedulerFactory, IAmA
     /// </summary>
     public string? Queue { get; set; }
 
+    private IBackgroundJobClientV2? _client;
+
     /// <summary>
-    /// The <see cref="IBackgroundJobClientV2"/>.
+    /// The <see cref="IBackgroundJobClientV2"/>. Lazily defaults to a <see cref="BackgroundJobClient"/>
+    /// bound to <see cref="JobStorage.Current"/> only when first accessed without an explicit assignment.
     /// </summary>
-    public IBackgroundJobClientV2 Client { get; set; } = new BackgroundJobClient();
+    public IBackgroundJobClientV2 Client
+    {
+        get => _client ??= new BackgroundJobClient();
+        set => _client = value;
+    }
 
     /// <summary>
     /// The <see cref="System.TimeProvider"/>

src/Paramore.Brighter.MessagingGateway.AWSSQS.V4/SqsMessageConsumer.cs

@@ -435,12 +435,12 @@ public async ValueTask DisposeAsync()
         if (_deadLetterProducer?.IsValueCreated == true && _deadLetterProducer.Value is IAsyncDisposable deadLetterAsync)
             await deadLetterAsync.DisposeAsync();
         else if (_deadLetterProducer?.IsValueCreated == true)
-            _deadLetterProducer.Value?.Dispose();
+            await _deadLetterProducer.Value!.DisposeAsync();
 
         if (_invalidMessageProducer?.IsValueCreated == true && _invalidMessageProducer.Value is IAsyncDisposable invalidAsync)
             await invalidAsync.DisposeAsync();
         else if (_invalidMessageProducer?.IsValueCreated == true)
-            _invalidMessageProducer.Value?.Dispose();
+            await _invalidMessageProducer.Value!.DisposeAsync();
 
         GC.SuppressFinalize(this);
     }

src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsMessageConsumer.cs

@@ -442,12 +442,12 @@ public async ValueTask DisposeAsync()
             if (_deadLetterProducer?.IsValueCreated == true && _deadLetterProducer.Value is IAsyncDisposable deadLetterAsync)
                 await deadLetterAsync.DisposeAsync();
             else if (_deadLetterProducer?.IsValueCreated == true)
-                _deadLetterProducer.Value?.Dispose();
+                await _deadLetterProducer.Value!.DisposeAsync();
 
             if (_invalidMessageProducer?.IsValueCreated == true && _invalidMessageProducer.Value is IAsyncDisposable invalidAsync)
                 await invalidAsync.DisposeAsync();
             else if (_invalidMessageProducer?.IsValueCreated == true)
-                _invalidMessageProducer.Value?.Dispose();
+                await _invalidMessageProducer.Value!.DisposeAsync();
 
             GC.SuppressFinalize(this);
         }

src/Paramore.Brighter.MessagingGateway.Redis/RedisMessageConsumer.cs

@@ -279,10 +279,12 @@ public Message[] Receive(TimeSpan? timeOut = null)
         {
             Log.RetrievingNextMessage(s_logger, _queueName, Topic);
 
+            Console.WriteLine($"[RDX] consumer ReceiveAsync entry queue='{_queueName}' topic='{Topic}' inflight=[{string.Join(",", _inflight.Keys)}] tid={Environment.CurrentManagedThreadId}");
             if (_inflight.Any())
             {
                 Log.UnackedMessageInFlight(s_logger, _queueName);
-                throw new ChannelFailureException($"Unacked message still in flight with id: {_inflight.Keys.First()}");   
+                Console.WriteLine($"[RDX] consumer THROW ChannelFailureException unacked id='{_inflight.Keys.First()}' queue='{_queueName}'");
+                throw new ChannelFailureException($"Unacked message still in flight with id: {_inflight.Keys.First()}");
             }
 
             timeOut ??= TimeSpan.FromSeconds(1);
@@ -295,8 +297,11 @@ public Message[] Receive(TimeSpan? timeOut = null)
                 await EnsureConnectionAsync(client);
                 (string? msgId, string rawMsg) redisMessage = await ReadMessageAsync(client, timeOut.Value);
                 if (redisMessage.msgId == null || string.IsNullOrEmpty(redisMessage.rawMsg))
+                {
+                    Console.WriteLine($"[RDX] consumer ReceiveAsync RETURN empty queue='{_queueName}' msgIdNull={redisMessage.msgId == null} rawMsgEmpty={string.IsNullOrEmpty(redisMessage.rawMsg)}");
                     return [];
-
+                }
+
                 var message = RedisMessageCreator.CreateMessage(redisMessage.rawMsg);
 
                 if (message.Header.MessageType != MessageType.MT_NONE && message.Header.MessageType != MessageType.MT_UNACCEPTABLE)
@@ -683,8 +688,11 @@ private async Task EnsureConnectionAsync(IRedisClientAsync client)
             Log.CreatingQueue(s_logger, _queueName);
             //what is the queue list key
             var key = Topic + "." + QUEUES;
-            //subscribe us 
+            //subscribe us
             await client.AddItemToSetAsync(key, _queueName);
+            Console.WriteLine($"[RDX] consumer EnsureConnectionAsync SADD key='{key}' queue='{_queueName}' tid={Environment.CurrentManagedThreadId} pool={Pool.Value.GetHashCode()}");
+            var members = await client.GetAllItemsFromSetAsync(key);
+            Console.WriteLine($"[RDX] consumer post-SADD set='{key}' members=[{string.Join(",", members)}]");
         }
 
         private (string? msgId, string rawMsg) ReadMessage(IRedisClient client, TimeSpan timeOut)
@@ -708,28 +716,34 @@ private async Task EnsureConnectionAsync(IRedisClientAsync client)
         {
             var msg = string.Empty;
             string? latestId = null;
+            var preBlpop = await client.GetAllItemsFromListAsync(_queueName);
+            Console.WriteLine($"[RDX] consumer ReadMessageAsync BLPOP queue='{_queueName}' topic='{Topic}' timeout={timeOut.TotalMilliseconds}ms tid={Environment.CurrentManagedThreadId} pool={Pool.Value.GetHashCode()} clientHash={client.GetHashCode()} pre-BLPOP-list=[{string.Join(",", preBlpop)}]");
             try
             {
                 // Give the server-side BLPOP timeout a chance to fire first; if the client cancels first
                 // a freshly-pushed item can be delivered to the lingering BLPOP and lost.
                 using var cts = new CancellationTokenSource(timeOut + TimeSpan.FromSeconds(1));
                 latestId = await client.BlockingRemoveStartFromListAsync(_queueName, timeOut, cts.Token);
+                Console.WriteLine($"[RDX] consumer BLPOP returned id='{latestId ?? "<null>"}' queue='{_queueName}'");
                 if (latestId != null)
                 {
                     var key = Topic + "." + latestId;
-                    msg = await client.GetValueAsync(key);
+                    msg = await client.GetValueAsync(key) ?? string.Empty;
+                    Console.WriteLine($"[RDX] consumer GET key='{key}' rawMsgLen={msg.Length}");
                     Log.ReceivedMessageFromQueue(s_logger, _queueName, Topic, JsonSerializer.Serialize(msg, JsonSerialisationOptions.Options));
                 }
             }
             catch (OperationCanceledException)
             {
+                Console.WriteLine($"[RDX] consumer BLPOP cancelled queue='{_queueName}'");
                 Log.TimeoutWithoutReceivingMessage(s_logger, _queueName, Topic);
             }
             catch (RedisException re) when (re.InnerException is OperationCanceledException)
             {
+                Console.WriteLine($"[RDX] consumer BLPOP redis-ex (cancelled) queue='{_queueName}' ex={re.Message}");
                 Log.TimeoutWithoutReceivingMessage(s_logger, _queueName, Topic);
             }
-            
+
             return (latestId, msg);
         }
 

src/Paramore.Brighter.MessagingGateway.Redis/RedisMessageProducer.cs

@@ -200,8 +200,10 @@ public async Task SendWithDelayAsync(Message message, TimeSpan? delay, Cancellat
             Log.PublishingMessage(s_logger, message.Header.Topic.Value, message.Id.ToString(), message.Body.Value);
             //increment a counter to get the next message id
             var nextMsgId = await IncrementMessageCounterAsync(client, cancellationToken);
+            Console.WriteLine($"[RDX] producer SendAsync topic='{Topic}' messageId='{message.Id}' nextMsgId={nextMsgId} tid={Environment.CurrentManagedThreadId} pool={Pool.Value.GetHashCode()}");
             //store the message, against that id
             await StoreMessageAsync(client, redisMessage, nextMsgId);
+            Console.WriteLine($"[RDX] producer SET key='{Topic}.{nextMsgId}' bodyLen={redisMessage.Length}");
             //If there are subscriber queues, push the message to the subscriber queues
             var pushedTo = await PushToQueuesAsync(client, nextMsgId, cancellationToken);
             Log.PublishedMessage(s_logger, message.Header.Topic.Value, message.Id.ToString(), message.Body.Value, string.Join(", ", pushedTo));
@@ -227,6 +229,8 @@ private async Task<HashSet<string>> PushToQueuesAsync(IRedisClientAsync client,
             {
                 //First add to the queue itself
                 await client.AddItemToListAsync(queue, nextMsgId.ToString(), cancellationToken);
+                var contents = await client.GetAllItemsFromListAsync(queue, cancellationToken);
+                Console.WriteLine($"[RDX] producer post-RPUSH '{nextMsgId}' to queue='{queue}' contents=[{string.Join(",", contents)}] clientHash={client.GetHashCode()}");
             }
             return queues;
         }

src/Paramore.Brighter.ServiceActivator/Performer.cs

@@ -61,6 +61,9 @@ public void Stop(RoutingKey routingKey)
         /// <returns>Task.</returns>
         public Task Run()
         {
+            // Pin to TaskScheduler.Default so ambient schedulers (TUnit/async test hosts,
+            // limited-concurrency schedulers) cannot queue the pump behind other work
+            // and starve it indefinitely.
             return Task.Factory.StartNew(
                 () => _messagePump.Run(),
                 CancellationToken.None,

src/Paramore.Brighter/InternalBus.cs

@@ -1,4 +1,4 @@
-#region Licence
+#region Licence
 /* The MIT License (MIT)
 Copyright © 2014 Ian Cooper <ian_hammond_cooper@yahoo.co.uk>
 
@@ -48,11 +48,15 @@ public class InternalBus(int boundedCapacity = -1) : IAmABus
     public void Enqueue(Message message, TimeSpan? timeout = null)
     {
         timeout ??= TimeSpan.FromMilliseconds(-1);
-        
+
         ValidateMillisecondsTimeout(timeout.Value);
 
         var topic = message.Header.Topic;
 
+        // GetOrAdd guarantees we enqueue to the collection that's actually stored in
+        // the dictionary — the previous TryGetValue/TryAdd pair lost messages under
+        // concurrent first-write contention because the losing thread enqueued to its
+        // own dangling BlockingCollection.
         var blockingCollection = _messages.GetOrAdd(topic, _ => boundedCapacity > 0
             ? new BlockingCollection<Message>(boundedCapacity)
             : new BlockingCollection<Message>());

src/Paramore.Brighter/JsonConverters/IdConverter.cs

@@ -30,7 +30,17 @@ namespace Paramore.Brighter.JsonConverters;
 
 public class IdConverter : JsonConverter<Id>
 {
-    public override Id Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) => new(reader.GetString()!);
+    public override bool HandleNull => true;
+
+    public override Id? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
+    {
+        if (reader.TokenType == JsonTokenType.Null)
+        {
+            return null;
+        }
+
+        return new Id(reader.GetString()!);
+    }
 
     public override void Write(Utf8JsonWriter writer, Id? value, JsonSerializerOptions options)
     {

src/Paramore.Brighter/Observability/Baggage.cs

@@ -139,5 +139,34 @@ public static Baggage FromString(string baggageString)
         baggage.LoadBaggage(baggageString);
         return baggage;
     }
+
+    /// <summary>
+    /// Two Baggage instances are equal when they contain the same set of key-value
+    /// pairs. W3C baggage is an unordered set, so iteration order is not significant.
+    /// </summary>
+    public bool Equals(Baggage? other)
+    {
+        if (other is null) return false;
+        if (ReferenceEquals(this, other)) return true;
+        if (_entries.Count != other._entries.Count) return false;
+
+        foreach (var kvp in _entries)
+        {
+            if (!other._entries.TryGetValue(kvp.Key, out var otherValue)) return false;
+            if (!string.Equals(kvp.Value, otherValue, StringComparison.Ordinal)) return false;
+        }
+
+        return true;
+    }
+
+    public override bool Equals(object? obj) => obj is Baggage other && Equals(other);
+
+    public override int GetHashCode()
+    {
+        var hash = 0;
+        foreach (var kvp in _entries)
+            hash ^= HashCode.Combine(kvp.Key, kvp.Value);
+        return hash;
+    }
 }
 

src/Paramore.Brighter/RequestHandler.cs

@@ -73,6 +73,9 @@ public abstract partial class RequestHandler<TRequest>(InstrumentationOptions in
         /// <param name="successor">The successor.</param>
         public void SetSuccessor(IHandleRequests<TRequest> successor)
         {
+            if (this == successor)
+                return;
+
             _successor = successor;
         }
 

src/Paramore.Brighter/Transforms/Transformers/CompressPayloadTransformer.cs

@@ -122,7 +122,6 @@ public async Task<Message> WrapAsync(Message message, Publication publication, C
 
         var originalContentType = message.Header.ContentType ?? new ContentType(MediaTypeNames.Text.Plain){CharSet = CharacterEncoding.UTF8.FromCharacterEncoding()};
         var contentType = new ContentType(mimeType);
-        contentType.CharSet = message.Header.ContentType?.CharSet ?? CharacterEncoding.UTF8.FromCharacterEncoding();
         message.Header.ContentType = contentType;
         message.Header.Bag.Add(ORIGINAL_CONTENTTYPE_HEADER, originalContentType.ToString());
 

tests/Directory.Build.props

@@ -1,10 +1,15 @@
 <Project>
   <PropertyGroup>
     <IsPackable>false</IsPackable>
+    <ImplicitUsings>enable</ImplicitUsings>
     <BrighterTestTargetFrameworks>net9.0;net10.0</BrighterTestTargetFrameworks>
     <BrighterTestNineOnlyTargetFrameworks>net9.0</BrighterTestNineOnlyTargetFrameworks>
     <DisableMsCoverageReferencedPathMaps Condition="'$(CollectCoverage)' != 'true'">true</DisableMsCoverageReferencedPathMaps>
     <EnableSourceLink>false</EnableSourceLink>
     <EnableSourceControlManagerQueries>false</EnableSourceControlManagerQueries>
   </PropertyGroup>
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Testing.Extensions.HangDump" />
+    <Compile Include="$(MSBuildThisFileDirectory)TestExceptionRecorder.cs" Link="TestExceptionRecorder.cs" />
+  </ItemGroup>
 </Project>