TaskCompletionSource: Use TaskCreationOptions.RunContinuationsAsynchronously (#228)

Author: pglombardo

Source/HiveMQtt/Client/HiveMQClient.cs

@@ -81,7 +81,7 @@ public async Task<ConnectResult> ConnectAsync()
 
         await this.Connection.ConnectAsync().ConfigureAwait(true);
 
-        var taskCompletionSource = new TaskCompletionSource<ConnAckPacket>();
+        var taskCompletionSource = new TaskCompletionSource<ConnAckPacket>(TaskCreationOptions.RunContinuationsAsynchronously);
         void TaskHandler(object? sender, OnConnAckReceivedEventArgs args) => taskCompletionSource.SetResult(args.ConnAckPacket);
 
         EventHandler<OnConnAckReceivedEventArgs> eventHandler = TaskHandler;
@@ -161,7 +161,7 @@ public async Task<bool> DisconnectAsync(DisconnectOptions? options = null)
         // Once this is set, no more incoming packets or outgoing will be accepted
         this.Connection.State = ConnectState.Disconnecting;
 
-        var taskCompletionSource = new TaskCompletionSource<DisconnectPacket>();
+        var taskCompletionSource = new TaskCompletionSource<DisconnectPacket>(TaskCreationOptions.RunContinuationsAsynchronously);
         void TaskHandler(object? sender, OnDisconnectSentEventArgs args) => taskCompletionSource.SetResult(args.DisconnectPacket);
         EventHandler<OnDisconnectSentEventArgs> eventHandler = TaskHandler;
         this.OnDisconnectSent += eventHandler;
@@ -315,7 +315,7 @@ public async Task<SubscribeResult> SubscribeAsync(SubscribeOptions options)
         var subscribePacket = new SubscribePacket(options, (ushort)packetIdentifier);
 
         // Setup the task completion source to wait for the SUBACK
-        var taskCompletionSource = new TaskCompletionSource<SubAckPacket>();
+        var taskCompletionSource = new TaskCompletionSource<SubAckPacket>(TaskCreationOptions.RunContinuationsAsynchronously);
         void TaskHandler(object? sender, OnSubAckReceivedEventArgs args)
         {
             if (args.SubAckPacket.PacketIdentifier == subscribePacket.PacketIdentifier)
@@ -429,7 +429,7 @@ public async Task<UnsubscribeResult> UnsubscribeAsync(UnsubscribeOptions unsubOp
         var packetIdentifier = await this.Connection.PacketIDManager.GetAvailablePacketIDAsync().ConfigureAwait(false);
         var unsubscribePacket = new UnsubscribePacket(unsubOptions, (ushort)packetIdentifier);
 
-        var taskCompletionSource = new TaskCompletionSource<UnsubAckPacket>();
+        var taskCompletionSource = new TaskCompletionSource<UnsubAckPacket>(TaskCreationOptions.RunContinuationsAsynchronously);
         void TaskHandler(object? sender, OnUnsubAckReceivedEventArgs args)
         {
             if (args.UnsubAckPacket.PacketIdentifier == unsubscribePacket.PacketIdentifier)

Source/HiveMQtt/MQTT5/Packets/PublishPacket.cs

@@ -119,7 +119,7 @@ internal virtual void OnPublishQoS1CompleteEventLauncher(PubAckPacket packet)
     /// Valid for outgoing Publish messages QoS 1.  A TaskCompletionSource that is set when the QoS 1 publish transaction is complete.
     /// </para>
     /// </summary>
-    public TaskCompletionSource<PubAckPacket> OnPublishQoS1CompleteTCS { get; } = new();
+    public TaskCompletionSource<PubAckPacket> OnPublishQoS1CompleteTCS { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
 
     /// <summary>
     /// Valid for outgoing Publish messages QoS 2.  An event that is fired after the the QoS 2 PubComp is received.
@@ -150,7 +150,7 @@ internal virtual void OnPublishQoS2CompleteEventLauncher(List<ControlPacket> pac
     /// Valid for outgoing Publish messages QoS 2.  A TaskCompletionSource that is set when the QoS 2 publish transaction is complete.
     /// </para>
     /// </summary>
-    public TaskCompletionSource<List<ControlPacket>> OnPublishQoS2CompleteTCS { get; } = new();
+    public TaskCompletionSource<List<ControlPacket>> OnPublishQoS2CompleteTCS { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
 
     /// <summary>
     /// Decode the received MQTT Publish packet.

Source/HiveMQtt/MQTT5/Packets/SubscribePacket.cs

@@ -96,7 +96,7 @@ internal virtual void OnCompleteEventLauncher(SubAckPacket packet)
     /// Valid for outgoing subscribe packets.  A TaskCompletionSource that is set when the subscribe transaction is complete.
     /// </para>
     /// </summary>
-    public TaskCompletionSource<SubAckPacket> OnCompleteTCS { get; } = new();
+    public TaskCompletionSource<SubAckPacket> OnCompleteTCS { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
 
     /// <summary>
     /// Encode this packet to be sent on the wire.

Tests/HiveMQtt.Test/HiveMQClient/ConnectTest.cs

@@ -62,7 +62,7 @@ public async Task TestConnectEventsAsync()
         client.OnConnAckReceived += OnConnAckReceivedHandler;
 
         // Set up TaskCompletionSource to wait for event handlers to finish
-        var taskCompletionSource = new TaskCompletionSource<bool>();
+        var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
         client.OnDisconnectSent += (sender, args) => taskCompletionSource.SetResult(true);
 
         // Connect and Disconnect
@@ -105,7 +105,7 @@ public async Task Test_AfterDisconnectEvent_Async()
         client.AfterDisconnect += AfterDisconnectHandler;
 
         // Set up TaskCompletionSource to wait for event handlers to finish
-        var taskCompletionSource = new TaskCompletionSource<bool>();
+        var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
         client.OnDisconnectSent += (sender, args) => taskCompletionSource.SetResult(true);
 
         // Connect and Disconnect

Tests/HiveMQtt.Test/HiveMQClient/LWTTest.cs

@@ -20,7 +20,7 @@ public async Task Last_Will_With_Properties_Async()
         Assert.True(listenerClient.IsConnected());
 
         var messagesReceived = 0;
-        var taskLWTReceived = new TaskCompletionSource<bool>();
+        var taskLWTReceived = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
 
         // Set the event handler for the message received event
         listenerClient.OnMessageReceived += (sender, args) =>

Tests/HiveMQtt.Test/HiveMQClient/LastWillAndTestamentBuilderTest.cs

@@ -40,7 +40,7 @@ public async Task Last_Will_With_Properties_Async()
         Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);
         Assert.True(listenerClient.IsConnected());
 
-        var taskLWTReceived = new TaskCompletionSource<bool>();
+        var taskLWTReceived = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
         var correlationData = new byte[] { 1, 2, 3, 4, 5 };
 
         // Set the event handler for the message received event

Tests/HiveMQtt.Test/HiveMQClient/Plan/PacketRestrictionsTest.cs

@@ -46,7 +46,6 @@ public async Task Unsubscribe_From_NonExistent_Topic_Should_Return_ReasonCode_17
         result.Should().NotBeNull();
         // result.ReasonCodes.Should().Contain(ReasonCode.NoSubscriptionExisted);
         // result.Subscriptions.Count.Should().Be(1);
-
         var disconnectResult = await client.DisconnectAsync().ConfigureAwait(false);
         disconnectResult.Should().BeTrue();
     }

Tests/HiveMQtt.Test/HiveMQClient/PubSubTest.cs

@@ -15,7 +15,7 @@ public async Task MostBasicPubSubAsync()
         var testPayload = new string(/*lang=json,strict*/ "{\"interference\": \"1029384\"}");
         var options = new HiveMQClientOptionsBuilder().WithClientId("MostBasicPubSubAsync").Build();
         var client = new HiveMQClient(options);
-        var taskCompletionSource = new TaskCompletionSource<bool>();
+        var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
 
         var connectResult = await client.ConnectAsync().ConfigureAwait(false);
         Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);
@@ -59,7 +59,7 @@ public async Task QoS1PubSubAsync()
         var testPayload = new string(/*lang=json,strict*/ "{\"interference\": \"1029384\"}");
         var options = new HiveMQClientOptionsBuilder().WithClientId("MostBasicPubSubAsync").Build();
         var client = new HiveMQClient(options);
-        var taskCompletionSource = new TaskCompletionSource<bool>();
+        var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
         var messagesReceived = 0;
 
         var connectResult = await client.ConnectAsync().ConfigureAwait(false);
@@ -109,7 +109,7 @@ public async Task QoS2PubSubAsync()
 
         var options = new HiveMQClientOptionsBuilder().WithClientId("QoS2PubSubAsync").Build();
         var client = new HiveMQClient(options);
-        var taskCompletionSource = new TaskCompletionSource<bool>();
+        var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
         var connectResult = await client.ConnectAsync().ConfigureAwait(false);
         Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);
         var messagesReceived = 0;
@@ -159,7 +159,7 @@ public async Task LargeMessagePubSubAsync()
         var testPayload = "1. A delusion starts like any other idea, as an egg. Identical on the outside, perfectly formed. From the shell, you'd never know anything was wrong. It's what's inside that matters. 2. A delusion starts like any other idea, as an egg. Identical on the outside, perfectly formed. From the shell, you'd never know anything was wrong. It's what's inside that matters. 3. A delusion starts like any other idea, as an egg. Identical on the outside, perfectly formed. From the shell, you'd never know anything was wrong. It's what's inside that matters.";
         var options = new HiveMQClientOptionsBuilder().WithClientId("LargeMessagePubSubAsync").Build();
         var client = new HiveMQClient(options);
-        var taskCompletionSource = new TaskCompletionSource<bool>();
+        var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
 
         var connectResult = await client.ConnectAsync().ConfigureAwait(false);
         Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);
@@ -212,7 +212,7 @@ public async Task OneMBMessagePubSubAsync()
 
         var options = new HiveMQClientOptionsBuilder().WithClientId("OneMBMessagePubSubAsync").Build();
         var client = new HiveMQClient(options);
-        var taskCompletionSource = new TaskCompletionSource<bool>();
+        var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
 
         var connectResult = await client.ConnectAsync().ConfigureAwait(false);
         Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);

Tests/HiveMQtt.Test/HiveMQClient/PublishBuilderTest.cs

@@ -217,7 +217,7 @@ public async Task PublishPayloadFormatIndicatorAsync()
             .WithPayloadFormatIndicator(MQTT5PayloadFormatIndicator.UTF8Encoded)
             .Build();
 
-        var taskCompletionSource = new TaskCompletionSource<bool>();
+        var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
         client.OnPublishSent += (sender, args) =>
         {
             Assert.Equal(MQTT5PayloadFormatIndicator.UTF8Encoded, args.PublishPacket.Message.PayloadFormatIndicator);

Tests/HiveMQtt.Test/HiveMQClient/PublishTest.cs

@@ -188,7 +188,7 @@ public async Task PublishPayloadFormatIndicatorAsync()
             Payload = Encoding.ASCII.GetBytes("blah"),
         };
 
-        var taskCompletionSource = new TaskCompletionSource<bool>();
+        var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
         client.OnPublishSent += (sender, args) =>
         {
             Assert.Equal(MQTT5PayloadFormatIndicator.UTF8Encoded, args.PublishPacket.Message.PayloadFormatIndicator);