From d2ce9d607a0058d53bc46bb594b6af44e9a1925c Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sun, 23 Oct 2022 08:16:45 +0200 Subject: [PATCH] Fix freezes in _AsyncLock_ (#1549) --- .github/workflows/ReleaseNotes.md | 1 + Samples/Client/Client_Connection_Samples.cs | 63 +++++- .../MqttConnectionContext.cs | 2 +- .../MQTTnet.Benchmarks/AsyncLockBenchmark.cs | 4 +- .../ManagedMqttClient.cs | 10 +- .../ManagedMqttClientStorageManager.cs | 4 +- .../MQTTnet.Extensions.Rpc/MqttRpcClient.cs | 6 +- .../WebSocket4NetMqttChannel.cs | 5 +- Source/MQTTnet.TestApp/AsyncLockTest.cs | 2 +- Source/MQTTnet.TestApp/ClientTest.cs | 4 +- Source/MQTTnet.TestApp/ManagedClientTest.cs | 6 +- .../MQTTnet.TestApp/MessageThroughputTest.cs | 3 +- Source/MQTTnet.TestApp/PublicBrokerTest.cs | 4 +- Source/MQTTnet.TestApp/ServerTest.cs | 20 +- .../Client/ManagedMqttClient_Tests.cs | 14 +- .../MQTTnet.Tests/Client/MqttClient_Tests.cs | 38 ++-- .../Diagnostics/PacketInspection_Tests.cs | 4 +- .../Factory/MqttFactory_Tests.cs | 4 +- .../Internal/AsyncEvent_Tests.cs | 49 ++++- .../MQTTnet.Tests/Internal/AsyncLock_Tests.cs | 178 ++++++++++++++-- Source/MQTTnet.Tests/MQTTnet.Tests.csproj | 2 +- Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs | 7 +- Source/MQTTnet.Tests/MQTTv5/Server_Tests.cs | 19 +- .../Mockups/MemoryMqttChannel.cs | 5 +- .../TestApplicationMessageReceivedHandler.cs | 4 +- .../MQTTnet.Tests/Mockups/TestEnvironment.cs | 6 +- Source/MQTTnet.Tests/RPC_Tests.cs | 3 +- Source/MQTTnet.Tests/RoundtripTime_Tests.cs | 4 +- .../Server/Assigned_Client_ID_Tests.cs | 10 +- Source/MQTTnet.Tests/Server/Events_Tests.cs | 16 +- Source/MQTTnet.Tests/Server/General.cs | 65 +++--- Source/MQTTnet.Tests/Server/Load_Tests.cs | 8 +- Source/MQTTnet.Tests/Server/QoS_Tests.cs | 8 +- Source/MQTTnet.Tests/Server/Security_Tests.cs | 8 +- .../Server/Server_Reference_Tests.cs | 3 +- Source/MQTTnet.Tests/Server/Session_Tests.cs | 72 +++---- .../MQTTnet.Tests/Server/Subscribe_Tests.cs | 24 +-- .../MQTTnet.Tests/Server/Topic_Alias_Tests.cs | 4 +- .../Server/User_Properties_Tests.cs | 3 +- Source/MQTTnet.Tests/Server/Will_Tests.cs | 6 +- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 4 +- Source/MQTTnet/Client/IMqttClient.cs | 1 + Source/MQTTnet/Client/MqttClient.cs | 19 +- Source/MQTTnet/Formatter/MqttBufferReader.cs | 8 +- Source/MQTTnet/Formatter/MqttPacketBuffer.cs | 3 +- .../Implementations/CrossPlatformSocket.cs | 193 ++++++++++++------ .../MQTTnet/Implementations/MqttTcpChannel.cs | 3 +- .../Implementations/MqttTcpServerAdapter.cs | 7 +- .../Implementations/MqttWebSocketChannel.cs | 2 +- .../PlatformAbstractionLayer.cs | 43 ---- Source/MQTTnet/Internal/AsyncEvent.cs | 9 +- .../MQTTnet/Internal/AsyncEventInvocator.cs | 2 +- Source/MQTTnet/Internal/AsyncLock.cs | 188 ++++++++--------- .../Internal/AsyncTaskCompletionSource.cs | 25 ++- Source/MQTTnet/Internal/CompletedTask.cs | 17 ++ Source/MQTTnet/Internal/EmptyBuffer.cs | 19 ++ Source/MQTTnet/MQTTnet.csproj | 1 - .../Events/ValidatingConnectionEventArgs.cs | 3 +- Source/MQTTnet/Server/Internal/MqttClient.cs | 6 +- .../Internal/MqttClientSessionsManager.cs | 6 +- .../Internal/MqttRetainedMessagesManager.cs | 8 +- .../Internal/MqttServerKeepAliveMonitor.cs | 41 +++- Source/MQTTnet/Server/MqttServer.cs | 3 +- .../Server/Status/MqttSessionStatus.cs | 5 +- 64 files changed, 798 insertions(+), 516 deletions(-) delete mode 100644 Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs create mode 100644 Source/MQTTnet/Internal/CompletedTask.cs create mode 100644 Source/MQTTnet/Internal/EmptyBuffer.cs diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index caaab403..8dee1663 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1 +1,2 @@ +* [Core] Fixed dead lock and race conditions in new _AsyncLock_ implementation (#1542). * [Client] Fixed connection freeze when using Xamarin etc. \ No newline at end of file diff --git a/Samples/Client/Client_Connection_Samples.cs b/Samples/Client/Client_Connection_Samples.cs index 18b619c4..e344ab30 100644 --- a/Samples/Client/Client_Connection_Samples.cs +++ b/Samples/Client/Client_Connection_Samples.cs @@ -119,7 +119,7 @@ public static class Client_Connection_Samples // The used public broker sometimes has invalid certificates. This sample accepts all // certificates. This should not be used in live environments. o.CertificateValidationHandler = _ => true; - + // The default value is determined by the OS. Set manually to force version. o.SslProtocol = SslProtocols.Tls12; }) @@ -134,15 +134,17 @@ public static class Client_Connection_Samples } } - public static async Task Connect_Client_Using_WebSockets() + public static async Task Connect_Client_Using_WebSocket4Net() { /* * This sample creates a simple MQTT client and connects to a public broker using a WebSocket connection. + * Instead of the .NET implementation of WebSockets the implementaion from WebSocket4Net is used. It provides more + * encryption algorithms and supports more platforms. * * This is a modified version of the sample _Connect_Client_! See other sample for more details. */ - var mqttFactory = new MqttFactory(); + var mqttFactory = new MqttFactory().UseWebSocket4Net(); using (var mqttClient = mqttFactory.CreateMqttClient()) { @@ -155,18 +157,16 @@ public static class Client_Connection_Samples response.DumpToConsole(); } } - - public static async Task Connect_Client_Using_WebSocket4Net() + + public static async Task Connect_Client_Using_WebSockets() { /* * This sample creates a simple MQTT client and connects to a public broker using a WebSocket connection. - * Instead of the .NET implementation of WebSockets the implementaion from WebSocket4Net is used. It provides more - * encryption algorithms and supports more platforms. * * This is a modified version of the sample _Connect_Client_! See other sample for more details. */ - var mqttFactory = new MqttFactory().UseWebSocket4Net(); + var mqttFactory = new MqttFactory(); using (var mqttClient = mqttFactory.CreateMqttClient()) { @@ -206,7 +206,7 @@ public static class Client_Connection_Samples using (var timeout = new CancellationTokenSource(5000)) { var response = await mqttClient.ConnectAsync(mqttClientOptions, timeout.Token); - + Console.WriteLine("The MQTT client is connected."); response.DumpToConsole(); @@ -214,6 +214,49 @@ public static class Client_Connection_Samples } } + public static async Task Disconnect_Clean() + { + /* + * This sample disconnects from the server with sending a DISCONNECT packet. + * This way of disconnecting is treated as a clean disconnect which will not + * trigger sending the last will etc. + */ + + var mqttFactory = new MqttFactory(); + + using (var mqttClient = mqttFactory.CreateMqttClient()) + { + var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build(); + + await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); + + // Calling _DisconnectAsync_ will send a DISCONNECT packet before closing the connection. + // Using a reason code requires MQTT version 5.0.0! + await mqttClient.DisconnectAsync(MqttClientDisconnectReason.ImplementationSpecificError); + } + } + + public static async Task Disconnect_Non_Clean() + { + /* + * This sample disconnects from the server without sending a DISCONNECT packet. + * This way of disconnecting is treated as a non clean disconnect which will + * trigger sending the last will etc. + */ + + var mqttFactory = new MqttFactory(); + + var mqttClient = mqttFactory.CreateMqttClient(); + + var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build(); + + await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); + + // Calling _Dispose_ or use of a _using_ statement will close the transport connection + // without sending a DISCONNECT packet to the server. + mqttClient.Dispose(); + } + public static async Task Inspect_Certificate_Validation_Errors() { /* @@ -248,7 +291,7 @@ public static class Client_Connection_Samples } } } - + public static async Task Ping_Server() { /* diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index 1268d25e..90ee5ce6 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -171,7 +171,7 @@ namespace MQTTnet.AspNetCore { var formatter = PacketFormatterAdapter; - using (await _writerLock.WaitAsync(cancellationToken).ConfigureAwait(false)) + using (await _writerLock.EnterAsync(cancellationToken).ConfigureAwait(false)) { var buffer = formatter.Encode(packet); var msg = buffer.Join().AsMemory(); diff --git a/Source/MQTTnet.Benchmarks/AsyncLockBenchmark.cs b/Source/MQTTnet.Benchmarks/AsyncLockBenchmark.cs index 5b50c3e3..27a348d4 100644 --- a/Source/MQTTnet.Benchmarks/AsyncLockBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/AsyncLockBenchmark.cs @@ -29,7 +29,7 @@ namespace MQTTnet.Benchmarks tasks[i] = Task.Run( async () => { - using (await asyncLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) + using (await asyncLock.EnterAsync().ConfigureAwait(false)) { var localI = globalI; await Task.Delay(5).ConfigureAwait(false); // Increase the chance for wrong data. @@ -56,7 +56,7 @@ namespace MQTTnet.Benchmarks { for (var i = 0; i < 100000; i++) { - using (await asyncLock.WaitAsync(cancellationToken.Token).ConfigureAwait(false)) + using (await asyncLock.EnterAsync(cancellationToken.Token).ConfigureAwait(false)) { } } diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index b4fa02c2..2583a605 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -167,7 +167,7 @@ namespace MQTTnet.Extensions.ManagedClient try { - using (await _messageQueueLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _messageQueueLock.EnterAsync().ConfigureAwait(false)) { if (_messageQueue.Count >= Options.MaxPendingMessages) { @@ -296,7 +296,7 @@ namespace MQTTnet.Extensions.ManagedClient _subscriptionsQueuedSignal.Release(); - return Task.FromResult(0); + return CompletedTask.Instance; } public Task UnsubscribeAsync(ICollection topics) @@ -319,7 +319,7 @@ namespace MQTTnet.Extensions.ManagedClient _subscriptionsQueuedSignal.Release(); - return Task.FromResult(0); + return CompletedTask.Instance; } protected override void Dispose(bool disposing) @@ -693,7 +693,7 @@ namespace MQTTnet.Extensions.ManagedClient { await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false); - using (await _messageQueueLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync + using (await _messageQueueLock.EnterAsync().ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync { // While publishing this message, this.PublishAsync could have booted this // message off the queue to make room for another (when using a cap @@ -724,7 +724,7 @@ namespace MQTTnet.Extensions.ManagedClient //contradict the expected behavior of QoS 1 and 2, that's also true //for the usage of a message queue cap, so it's still consistent //with prior behavior in that way. - using (await _messageQueueLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync + using (await _messageQueueLock.EnterAsync().ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync { _messageQueue.RemoveFirst(i => i.Id.Equals(message.Id)); diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientStorageManager.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientStorageManager.cs index 1b812d8d..bd1c48c2 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientStorageManager.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientStorageManager.cs @@ -34,7 +34,7 @@ namespace MQTTnet.Extensions.ManagedClient { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - using (await _messagesLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _messagesLock.EnterAsync().ConfigureAwait(false)) { _messages.Add(applicationMessage); await SaveAsync().ConfigureAwait(false); @@ -45,7 +45,7 @@ namespace MQTTnet.Extensions.ManagedClient { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - using (await _messagesLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _messagesLock.EnterAsync().ConfigureAwait(false)) { var index = _messages.IndexOf(applicationMessage); if (index == -1) diff --git a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs index 29a9d6f6..205e1591 100644 --- a/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs +++ b/Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs @@ -9,7 +9,7 @@ using System; using System.Collections.Concurrent; using System.Threading; using System.Threading.Tasks; -using MQTTnet.Implementations; +using MQTTnet.Internal; namespace MQTTnet.Extensions.Rpc { @@ -116,7 +116,7 @@ namespace MQTTnet.Extensions.Rpc { if (!_waitingCalls.TryRemove(eventArgs.ApplicationMessage.Topic, out var awaitable)) { - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } #if NET452 @@ -128,7 +128,7 @@ namespace MQTTnet.Extensions.Rpc // Set this message to handled to that other code can avoid execution etc. eventArgs.IsHandled = true; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } public void Dispose() diff --git a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs index bbbdc81f..d656a92f 100644 --- a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs +++ b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs @@ -14,6 +14,7 @@ using System.Threading.Tasks; using MQTTnet.Channel; using MQTTnet.Client; using MQTTnet.Exceptions; +using MQTTnet.Internal; using SuperSocket.ClientEngine; using WebSocket4Net; @@ -124,7 +125,7 @@ namespace MQTTnet.Extensions.WebSocket4Net _webSocket.Close(); } - return Task.FromResult(0); + return CompletedTask.Instance; } public Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) @@ -157,7 +158,7 @@ namespace MQTTnet.Extensions.WebSocket4Net public Task WriteAsync(ArraySegment buffer, bool isEndOfPacket, CancellationToken cancellationToken) { _webSocket.Send(buffer.Array, buffer.Offset, buffer.Count); - return Task.FromResult(0); + return CompletedTask.Instance; } public void Dispose() diff --git a/Source/MQTTnet.TestApp/AsyncLockTest.cs b/Source/MQTTnet.TestApp/AsyncLockTest.cs index 4f943b19..035f3d46 100644 --- a/Source/MQTTnet.TestApp/AsyncLockTest.cs +++ b/Source/MQTTnet.TestApp/AsyncLockTest.cs @@ -18,7 +18,7 @@ namespace MQTTnet.TestApp { for (var i = 0; i < 100000; i++) { - using (await asyncLock.WaitAsync(cancellationToken.Token).ConfigureAwait(false)) + using (await asyncLock.EnterAsync(cancellationToken.Token).ConfigureAwait(false)) { } } diff --git a/Source/MQTTnet.TestApp/ClientTest.cs b/Source/MQTTnet.TestApp/ClientTest.cs index 33469acb..9f233256 100644 --- a/Source/MQTTnet.TestApp/ClientTest.cs +++ b/Source/MQTTnet.TestApp/ClientTest.cs @@ -7,7 +7,7 @@ using System; using System.Text; using System.Threading.Tasks; using MQTTnet.Diagnostics; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; namespace MQTTnet.TestApp @@ -40,7 +40,7 @@ namespace MQTTnet.TestApp Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}"); Console.WriteLine(); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; client.ConnectedAsync += async e => diff --git a/Source/MQTTnet.TestApp/ManagedClientTest.cs b/Source/MQTTnet.TestApp/ManagedClientTest.cs index 88125ff8..976c17e6 100644 --- a/Source/MQTTnet.TestApp/ManagedClientTest.cs +++ b/Source/MQTTnet.TestApp/ManagedClientTest.cs @@ -9,7 +9,7 @@ using Newtonsoft.Json; using System.Collections.Generic; using MQTTnet.Client; using MQTTnet.Extensions.ManagedClient; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; namespace MQTTnet.TestApp @@ -42,7 +42,7 @@ namespace MQTTnet.TestApp managedClient.ApplicationMessageReceivedAsync += e => { Console.WriteLine(">> RECEIVED: " + e.ApplicationMessage.Topic); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await managedClient.StartAsync(options); @@ -85,7 +85,7 @@ namespace MQTTnet.TestApp public Task SaveQueuedMessagesAsync(IList messages) { File.WriteAllText(Filename, JsonConvert.SerializeObject(messages)); - return Task.FromResult(0); + return CompletedTask.Instance; } public Task> LoadQueuedMessagesAsync() diff --git a/Source/MQTTnet.TestApp/MessageThroughputTest.cs b/Source/MQTTnet.TestApp/MessageThroughputTest.cs index 67232f10..a7b36df9 100644 --- a/Source/MQTTnet.TestApp/MessageThroughputTest.cs +++ b/Source/MQTTnet.TestApp/MessageThroughputTest.cs @@ -5,6 +5,7 @@ using System; using System.Threading; using System.Threading.Tasks; using System.Linq; +using MQTTnet.Internal; namespace MQTTnet.TestApp { @@ -124,7 +125,7 @@ namespace MQTTnet.TestApp { _cancellationTokenSource.Cancel(); } - return Implementations.PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } public async Task Cleanup() diff --git a/Source/MQTTnet.TestApp/PublicBrokerTest.cs b/Source/MQTTnet.TestApp/PublicBrokerTest.cs index 65aaa31c..768ba874 100644 --- a/Source/MQTTnet.TestApp/PublicBrokerTest.cs +++ b/Source/MQTTnet.TestApp/PublicBrokerTest.cs @@ -8,7 +8,7 @@ using System.Security.Authentication; using System.Threading; using System.Threading.Tasks; using MQTTnet.Formatter; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; namespace MQTTnet.TestApp @@ -144,7 +144,7 @@ namespace MQTTnet.TestApp client.ApplicationMessageReceivedAsync += e => { receivedMessage = e.ApplicationMessage; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client.ConnectAsync(options); diff --git a/Source/MQTTnet.TestApp/ServerTest.cs b/Source/MQTTnet.TestApp/ServerTest.cs index e71f9761..a0925def 100644 --- a/Source/MQTTnet.TestApp/ServerTest.cs +++ b/Source/MQTTnet.TestApp/ServerTest.cs @@ -4,13 +4,11 @@ using System; using System.Collections.Generic; -using System.Diagnostics; using System.IO; using System.Text; using System.Threading.Tasks; using MQTTnet.Diagnostics; -using MQTTnet.Extensions.ManagedClient; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; using MQTTnet.Server; using Newtonsoft.Json; @@ -69,13 +67,13 @@ namespace MQTTnet.TestApp } File.WriteAllText(Filename, JsonConvert.SerializeObject(e.StoredRetainedMessages)); - return Task.FromResult(0); + return CompletedTask.Instance; }; mqttServer.RetainedMessagesClearedAsync += e => { File.Delete(Filename); - return Task.FromResult(0); + return CompletedTask.Instance; }; mqttServer.LoadingRetainedMessageAsync += e => @@ -93,7 +91,7 @@ namespace MQTTnet.TestApp e.LoadedRetainedMessages = retainedMessages; - return Task.FromResult(0); + return CompletedTask.Instance; }; mqttServer.InterceptingPublishAsync += e => @@ -111,7 +109,7 @@ namespace MQTTnet.TestApp e.CloseConnection = true; } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; mqttServer.ValidatingConnectionAsync += e => @@ -124,7 +122,7 @@ namespace MQTTnet.TestApp } } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; mqttServer.InterceptingSubscriptionAsync += e => @@ -140,7 +138,7 @@ namespace MQTTnet.TestApp e.CloseConnection = true; } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; mqttServer.InterceptingPublishAsync += e => @@ -149,7 +147,7 @@ namespace MQTTnet.TestApp $"'{e.ClientId}' reported '{e.ApplicationMessage.Topic}' > '{Encoding.UTF8.GetString(e.ApplicationMessage.Payload ?? new byte[0])}'", ConsoleColor.Magenta); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; //options.ApplicationMessageInterceptor = c => @@ -177,7 +175,7 @@ namespace MQTTnet.TestApp mqttServer.ClientConnectedAsync += e => { Console.Write("Client disconnected event fired."); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await mqttServer.StartAsync(); diff --git a/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs b/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs index a590027d..e161b212 100644 --- a/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Client/ManagedMqttClient_Tests.cs @@ -11,12 +11,10 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Extensions.ManagedClient; -using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.Protocol; using MQTTnet.Server; using MQTTnet.Tests.Mockups; -using MqttClient = MQTTnet.Client.MqttClient; namespace MQTTnet.Tests.Client { @@ -39,7 +37,7 @@ namespace MQTTnet.Tests.Client managedClient.ConnectingFailedAsync += args => { connectingFailedEventArgs = args; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await managedClient.StartAsync( @@ -117,7 +115,7 @@ namespace MQTTnet.Tests.Client receivingClient.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await receivingClient.SubscribeAsync("My/last/will"); @@ -294,7 +292,7 @@ namespace MQTTnet.Tests.Client managedClient.ApplicationMessageReceivedAsync += e => { receivedManagedMessages.Add(e.ApplicationMessage); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await managedClient.SubscribeAsync("topic"); @@ -428,7 +426,7 @@ namespace MQTTnet.Tests.Client managedClient.ConnectedAsync += e => { connected.TrySetResult(true); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; return connected.Task; @@ -457,7 +455,7 @@ namespace MQTTnet.Tests.Client result.TrySetResult(receivedMessages); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; return result.Task; @@ -486,7 +484,7 @@ namespace MQTTnet.Tests.Client public Task SaveQueuedMessagesAsync(IList messages) { _messages = messages; - return Task.FromResult(0); + return CompletedTask.Instance; } } } \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs b/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs index b073c5fe..8553ebd0 100644 --- a/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Client/MqttClient_Tests.cs @@ -14,7 +14,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Exceptions; using MQTTnet.Formatter; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Tests.Mockups; @@ -58,7 +58,7 @@ namespace MQTTnet.Tests.Client } _ = InvokeInternal(); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var publishes = Task.WhenAll(publisher.PublishStringAsync("a", null, qos), publisher.PublishStringAsync("b", null, qos)); @@ -109,7 +109,7 @@ namespace MQTTnet.Tests.Client client.DisconnectedAsync += args => { disconnectHandlerCalled = true; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").Build()); @@ -138,7 +138,7 @@ namespace MQTTnet.Tests.Client client.DisconnectedAsync += e => { ex = e.Exception; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; try @@ -170,7 +170,7 @@ namespace MQTTnet.Tests.Client server.InterceptingPublishAsync += c => { i++; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client.SendAsync( @@ -211,7 +211,7 @@ namespace MQTTnet.Tests.Client client.DisconnectedAsync += e => { handlerFired = true; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await server.StopAsync(); @@ -255,7 +255,7 @@ namespace MQTTnet.Tests.Client receiveClient.ApplicationMessageReceivedAsync += e => { receivedPayload = e.ApplicationMessage.ConvertPayloadToString(); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await receiveClient.SubscribeAsync("x"); @@ -313,7 +313,7 @@ namespace MQTTnet.Tests.Client receiver.ApplicationMessageReceivedAsync += e => { receivedMessage = e.ApplicationMessage; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await sender.PublishAsync(message.Build(), CancellationToken.None); @@ -456,7 +456,7 @@ namespace MQTTnet.Tests.Client receivedValues.Add(value); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } client1.ApplicationMessageReceivedAsync += Handler1; @@ -534,7 +534,7 @@ namespace MQTTnet.Tests.Client client2.ApplicationMessageReceivedAsync += e => { client2TopicResults.Add(Encoding.UTF8.GetString(e.ApplicationMessage.Payload)); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client2.SubscribeAsync(client2Topic); @@ -569,7 +569,7 @@ namespace MQTTnet.Tests.Client receivedMessages.Add(e.ApplicationMessage); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client1.SubscribeAsync("a"); @@ -728,7 +728,7 @@ namespace MQTTnet.Tests.Client replies.Add(eventArgs.ApplicationMessage.Topic); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } client2.ApplicationMessageReceivedAsync += Handler2; @@ -766,7 +766,7 @@ namespace MQTTnet.Tests.Client replyReceived = true; } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; client2.ApplicationMessageReceivedAsync += async e => @@ -817,7 +817,7 @@ namespace MQTTnet.Tests.Client replyReceived = true; } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client.PublishStringAsync("request", null, MqttQualityOfServiceLevel.AtLeastOnce); @@ -840,7 +840,7 @@ namespace MQTTnet.Tests.Client client.DisconnectedAsync += e => { Assert.IsTrue(e.ClientWasConnected); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client.DisconnectAsync(); @@ -860,7 +860,7 @@ namespace MQTTnet.Tests.Client client.DisconnectedAsync += e => { Assert.IsTrue(e.ClientWasConnected); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await server.StopAsync(); @@ -897,7 +897,7 @@ namespace MQTTnet.Tests.Client receivedMessages.Add(e.ApplicationMessage); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).Build()); @@ -922,14 +922,14 @@ namespace MQTTnet.Tests.Client client1.DisconnectedAsync += c => { disconnectedFired = true; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var messageReceived = false; client1.ApplicationMessageReceivedAsync += e => { messageReceived = true; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("topic1").WithExactlyOnceQoS().Build()); diff --git a/Source/MQTTnet.Tests/Diagnostics/PacketInspection_Tests.cs b/Source/MQTTnet.Tests/Diagnostics/PacketInspection_Tests.cs index 225f2cec..5589a37d 100644 --- a/Source/MQTTnet.Tests/Diagnostics/PacketInspection_Tests.cs +++ b/Source/MQTTnet.Tests/Diagnostics/PacketInspection_Tests.cs @@ -7,7 +7,7 @@ using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; -using MQTTnet.Implementations; +using MQTTnet.Internal; namespace MQTTnet.Tests.Diagnostics { @@ -33,7 +33,7 @@ namespace MQTTnet.Tests.Diagnostics mqttClient.InspectPackage += eventArgs => { packets.Add(eventArgs.Direction + ":" + Convert.ToBase64String(eventArgs.Buffer)); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); diff --git a/Source/MQTTnet.Tests/Factory/MqttFactory_Tests.cs b/Source/MQTTnet.Tests/Factory/MqttFactory_Tests.cs index b34d555e..68c0fc61 100644 --- a/Source/MQTTnet.Tests/Factory/MqttFactory_Tests.cs +++ b/Source/MQTTnet.Tests/Factory/MqttFactory_Tests.cs @@ -8,7 +8,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Diagnostics; using MQTTnet.Extensions.ManagedClient; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Server; namespace MQTTnet.Tests.Factory @@ -56,7 +56,7 @@ namespace MQTTnet.Tests.Factory managedClient.ConnectingFailedAsync += e => { tcs.TrySetResult(null); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await Task.WhenAny(Task.Delay(managedClient.Options.ClientOptions.Timeout), tcs.Task); diff --git a/Source/MQTTnet.Tests/Internal/AsyncEvent_Tests.cs b/Source/MQTTnet.Tests/Internal/AsyncEvent_Tests.cs index 54442bf6..94a960e5 100644 --- a/Source/MQTTnet.Tests/Internal/AsyncEvent_Tests.cs +++ b/Source/MQTTnet.Tests/Internal/AsyncEvent_Tests.cs @@ -6,7 +6,6 @@ using System; using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; -using MQTTnet.Implementations; using MQTTnet.Internal; namespace MQTTnet.Tests.Internal @@ -16,6 +15,28 @@ namespace MQTTnet.Tests.Internal { int _testEventAsyncCount; + [TestMethod] + public async Task Attach_and_Detach_Sync_And_Async_EventHandler() + { + var testClass = new TestClass(); + + testClass.TestEventAsync += OnTestEventAsync; + testClass.TestEvent += OnTestEvent; + + await testClass.FireEventAsync(EventArgs.Empty); + Assert.AreEqual(2, _testEventAsyncCount); + + testClass.TestEvent -= OnTestEvent; + + await testClass.FireEventAsync(EventArgs.Empty); + Assert.AreEqual(3, _testEventAsyncCount); + + testClass.TestEventAsync -= OnTestEventAsync; + + await testClass.FireEventAsync(EventArgs.Empty); + Assert.AreEqual(3, _testEventAsyncCount); + } + [TestMethod] public async Task Attach_EventHandler() { @@ -28,6 +49,19 @@ namespace MQTTnet.Tests.Internal Assert.AreEqual(1, _testEventAsyncCount); } + [TestMethod] + public async Task Attach_Sync_And_Async_EventHandler() + { + var testClass = new TestClass(); + + testClass.TestEventAsync += OnTestEventAsync; + testClass.TestEvent += OnTestEvent; + + await testClass.FireEventAsync(EventArgs.Empty); + + Assert.AreEqual(2, _testEventAsyncCount); + } + [TestMethod] public async Task Detach_EventHandler() { @@ -68,16 +102,27 @@ namespace MQTTnet.Tests.Internal Assert.AreEqual(false, testClass.HasTestHandlers); } + void OnTestEvent(EventArgs arg) + { + Interlocked.Increment(ref _testEventAsyncCount); + } + Task OnTestEventAsync(EventArgs arg) { Interlocked.Increment(ref _testEventAsyncCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } sealed class TestClass { readonly AsyncEvent _testEvent = new AsyncEvent(); + public event Action TestEvent + { + add => _testEvent.AddHandler(value); + remove => _testEvent.RemoveHandler(value); + } + public event Func TestEventAsync { add => _testEvent.AddHandler(value); diff --git a/Source/MQTTnet.Tests/Internal/AsyncLock_Tests.cs b/Source/MQTTnet.Tests/Internal/AsyncLock_Tests.cs index fae81227..e30fd63d 100644 --- a/Source/MQTTnet.Tests/Internal/AsyncLock_Tests.cs +++ b/Source/MQTTnet.Tests/Internal/AsyncLock_Tests.cs @@ -12,26 +12,130 @@ using MQTTnet.Internal; namespace MQTTnet.Tests.Internal { [TestClass] - public class AsyncLock_Tests + public sealed class AsyncLock_Tests { [TestMethod] - public void Lock_10_Parallel_Tasks() + public async Task Cancellation_Of_Awaiter() { - const int ThreadsCount = 10; - - var threads = new Task[ThreadsCount]; var @lock = new AsyncLock(); + + // This call will not yet "release" the lock due to missing _using_. + var releaser = await @lock.EnterAsync().ConfigureAwait(false); + + var counter = 0; + + Debug.WriteLine("Prepared locked lock."); + + _ = Task.Run( + async () => + { + // SHOULD GET TIMEOUT! + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1))) + { + using (await @lock.EnterAsync(timeout.Token)) + { + Debug.WriteLine("Task 1 incremented"); + counter++; + } + } + }); + + _ = Task.Run( + async () => + { + // SHOULD GET TIMEOUT! + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(2))) + { + using (await @lock.EnterAsync(timeout.Token)) + { + Debug.WriteLine("Task 2 incremented"); + counter++; + } + } + }); + + _ = Task.Run( + async () => + { + // SHOULD GET TIMEOUT! + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(3))) + { + using (await @lock.EnterAsync(timeout.Token)) + { + Debug.WriteLine("Task 3 incremented"); + counter++; + } + } + }); + + _ = Task.Run( + async () => + { + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(4))) + { + using (await @lock.EnterAsync(timeout.Token)) + { + Debug.WriteLine("Task 4 incremented"); + counter++; + } + } + }); + + _ = Task.Run( + async () => + { + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5))) + { + using (await @lock.EnterAsync(timeout.Token)) + { + Debug.WriteLine("Task 5 incremented"); + counter++; + } + } + }); + + _ = Task.Run( + async () => + { + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(6))) + { + using (await @lock.EnterAsync(timeout.Token)) + { + Debug.WriteLine("Task 6 incremented"); + counter++; + } + } + }); + + Debug.WriteLine("Delay before release..."); + await Task.Delay(TimeSpan.FromSeconds(3.1)); + releaser.Dispose(); + + Debug.WriteLine("Wait for all tasks..."); + await Task.Delay(TimeSpan.FromSeconds(6.1)); + + Assert.AreEqual(3, counter); + } + + [TestMethod] + public void Lock_Parallel_Tasks() + { + const int taskCount = 50; + + var @lock = new AsyncLock(); + + var tasks = new Task[taskCount]; var globalI = 0; - for (var i = 0; i < ThreadsCount; i++) + for (var i = 0; i < taskCount; i++) { #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed - threads[i] = Task.Run( + tasks[i] = Task.Run( async () => { - using (await @lock.WaitAsync(CancellationToken.None)) + using (await @lock.EnterAsync()) { var localI = globalI; - await Task.Delay(10); // Increase the chance for wrong data. + await Task.Delay(5); // Increase the chance for wrong data. localI++; globalI = localI; } @@ -39,8 +143,8 @@ namespace MQTTnet.Tests.Internal #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed } - Task.WaitAll(threads); - Assert.AreEqual(ThreadsCount, globalI); + Task.WaitAll(tasks); + Assert.AreEqual(taskCount, globalI); } [TestMethod] @@ -57,7 +161,7 @@ namespace MQTTnet.Tests.Internal threads[i] = Task.Run( async () => { - using (await @lock.WaitAsync(CancellationToken.None)) + using (await @lock.EnterAsync()) { var localI = globalI; await Task.Delay(10); // Increase the chance for wrong data. @@ -96,7 +200,7 @@ namespace MQTTnet.Tests.Internal var @lock = new AsyncLock(); for (var i = 0; i < 100; i++) { - using (await @lock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) + using (await @lock.EnterAsync().ConfigureAwait(false)) { sum++; } @@ -112,11 +216,11 @@ namespace MQTTnet.Tests.Internal var @lock = new AsyncLock(); // This call will never "release" the lock due to missing _using_. - await @lock.WaitAsync(CancellationToken.None).ConfigureAwait(false); + await @lock.EnterAsync().ConfigureAwait(false); using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3))) { - await @lock.WaitAsync(cts.Token).ConfigureAwait(false); + await @lock.EnterAsync(cts.Token).ConfigureAwait(false); } } @@ -125,15 +229,15 @@ namespace MQTTnet.Tests.Internal { var asyncLock = new AsyncLock(); - var releaser = await asyncLock.WaitAsync(CancellationToken.None).ConfigureAwait(false); + var releaser = await asyncLock.EnterAsync().ConfigureAwait(false); try { using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(3))) { - await asyncLock.WaitAsync(timeout.Token).ConfigureAwait(false); + await asyncLock.EnterAsync(timeout.Token).ConfigureAwait(false); } - + Assert.Fail("Exception should be thrown!"); } catch (OperationCanceledException) @@ -142,10 +246,46 @@ namespace MQTTnet.Tests.Internal releaser.Dispose(); - using (await asyncLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) + using (await asyncLock.EnterAsync(CancellationToken.None).ConfigureAwait(false)) { // When the method finished, the thread got access. } } + + [TestMethod] + public async Task Use_After_Cancellation() + { + var @lock = new AsyncLock(); + + // This call will not yet "release" the lock due to missing _using_. + var releaser = await @lock.EnterAsync().ConfigureAwait(false); + + try + { + using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3))) + { + await @lock.EnterAsync(cts.Token).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + // Expected exception! + } + + releaser.Dispose(); + + // Regular usage after cancellation. + using (await @lock.EnterAsync().ConfigureAwait(false)) + { + } + + using (await @lock.EnterAsync().ConfigureAwait(false)) + { + } + + using (await @lock.EnterAsync().ConfigureAwait(false)) + { + } + } } } \ No newline at end of file diff --git a/Source/MQTTnet.Tests/MQTTnet.Tests.csproj b/Source/MQTTnet.Tests/MQTTnet.Tests.csproj index 2e95c15d..d9a7ce1e 100644 --- a/Source/MQTTnet.Tests/MQTTnet.Tests.csproj +++ b/Source/MQTTnet.Tests/MQTTnet.Tests.csproj @@ -1,7 +1,7 @@  - net452;net461;netcoreapp3.1;net6.0 + net452;net461;net48;netcoreapp3.1;net6.0 false 7.3 false diff --git a/Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs b/Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs index b87ca6af..ac7b8792 100644 --- a/Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs +++ b/Source/MQTTnet.Tests/MQTTv5/Client_Tests.cs @@ -8,10 +8,9 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Formatter; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; -using MQTTnet.Tests.Mockups; namespace MQTTnet.Tests.MQTTv5 { @@ -43,7 +42,7 @@ namespace MQTTnet.Tests.MQTTv5 client.ApplicationMessageReceivedAsync += e => { receivedMessage = e.ApplicationMessage; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client.PublishAsync(new MqttApplicationMessageBuilder() @@ -245,7 +244,7 @@ namespace MQTTnet.Tests.MQTTv5 receiver.ApplicationMessageReceivedAsync += e => { receivedMessage = e.ApplicationMessage; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var sender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500)); diff --git a/Source/MQTTnet.Tests/MQTTv5/Server_Tests.cs b/Source/MQTTnet.Tests/MQTTv5/Server_Tests.cs index 180117b0..a187972d 100644 --- a/Source/MQTTnet.Tests/MQTTv5/Server_Tests.cs +++ b/Source/MQTTnet.Tests/MQTTv5/Server_Tests.cs @@ -8,7 +8,7 @@ using MQTTnet.Formatter; using MQTTnet.Tests.Mockups; using System.Threading; using System.Threading.Tasks; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; namespace MQTTnet.Tests.MQTTv5 @@ -31,7 +31,7 @@ namespace MQTTnet.Tests.MQTTv5 c1.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); @@ -171,7 +171,6 @@ namespace MQTTnet.Tests.MQTTv5 // Session should be present Assert.IsTrue(!result3.IsSessionPresent, "Session is present when it should not"); - } } @@ -180,7 +179,7 @@ namespace MQTTnet.Tests.MQTTv5 { using (var testEnvironment = CreateTestEnvironment()) { - var disconnectReason = MQTTnet.Client.MqttClientDisconnectReason.UnspecifiedError; + var disconnectReason = MqttClientDisconnectReason.UnspecifiedError; string testClientId = null; @@ -189,15 +188,15 @@ namespace MQTTnet.Tests.MQTTv5 testEnvironment.Server.ClientConnectedAsync += e => { testClientId = e.ClientId; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; - var client = await testEnvironment.ConnectClient(new MQTTnet.Client.MqttClientOptionsBuilder().WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)); + var client = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500)); client.DisconnectedAsync += e => { disconnectReason = e.Reason; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await LongTestDelay(); @@ -208,17 +207,17 @@ namespace MQTTnet.Tests.MQTTv5 // Have the server disconnect the client with AdministrativeAction reason - await testEnvironment.Server.DisconnectClientAsync(testClientId, Protocol.MqttDisconnectReasonCode.AdministrativeAction); + await testEnvironment.Server.DisconnectClientAsync(testClientId, MqttDisconnectReasonCode.AdministrativeAction); await LongTestDelay(); // The reason should be returned to the client in the DISCONNECT packet - Assert.AreEqual(MQTTnet.Client.MqttClientDisconnectReason.AdministrativeAction, disconnectReason); + Assert.AreEqual(MqttClientDisconnectReason.AdministrativeAction, disconnectReason); } } - MqttClientOptions CreateClientOptions(TestEnvironment testEnvironment, string clientId, bool cleanSession, uint sessionExpiryInterval) + static MqttClientOptions CreateClientOptions(TestEnvironment testEnvironment, string clientId, bool cleanSession, uint sessionExpiryInterval) { return testEnvironment.Factory.CreateClientOptionsBuilder() .WithProtocolVersion(MqttProtocolVersion.V500) diff --git a/Source/MQTTnet.Tests/Mockups/MemoryMqttChannel.cs b/Source/MQTTnet.Tests/Mockups/MemoryMqttChannel.cs index 7bdf97de..5cb8ead6 100644 --- a/Source/MQTTnet.Tests/Mockups/MemoryMqttChannel.cs +++ b/Source/MQTTnet.Tests/Mockups/MemoryMqttChannel.cs @@ -8,6 +8,7 @@ using System.Security.Cryptography.X509Certificates; using System.Threading; using System.Threading.Tasks; using MQTTnet.Channel; +using MQTTnet.Internal; namespace MQTTnet.Tests.Mockups { @@ -33,12 +34,12 @@ namespace MQTTnet.Tests.Mockups public Task ConnectAsync(CancellationToken cancellationToken) { - return Task.FromResult(0); + return CompletedTask.Instance; } public Task DisconnectAsync(CancellationToken cancellationToken) { - return Task.FromResult(0); + return CompletedTask.Instance; } public Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) diff --git a/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs b/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs index 81bca57b..c1a7c482 100644 --- a/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs +++ b/Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs @@ -7,7 +7,7 @@ using System.Linq; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; -using MQTTnet.Implementations; +using MQTTnet.Internal; namespace MQTTnet.Tests.Mockups { @@ -46,7 +46,7 @@ namespace MQTTnet.Tests.Mockups _receivedEventArgs.Add(eventArgs); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } } } \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs index c3da9717..6b79a963 100644 --- a/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs +++ b/Source/MQTTnet.Tests/Mockups/TestEnvironment.cs @@ -13,7 +13,7 @@ using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Extensions.Rpc; using MQTTnet.Formatter; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.LowLevelClient; using MQTTnet.Protocol; using MQTTnet.Server; @@ -220,7 +220,7 @@ namespace MQTTnet.Tests.Mockups } } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; return client; @@ -261,7 +261,7 @@ namespace MQTTnet.Tests.Mockups } } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; return Server; diff --git a/Source/MQTTnet.Tests/RPC_Tests.cs b/Source/MQTTnet.Tests/RPC_Tests.cs index e37996ea..f266f5a6 100644 --- a/Source/MQTTnet.Tests/RPC_Tests.cs +++ b/Source/MQTTnet.Tests/RPC_Tests.cs @@ -13,6 +13,7 @@ using MQTTnet.Extensions.Rpc; using MQTTnet.Protocol; using MQTTnet.Formatter; using MQTTnet.Implementations; +using MQTTnet.Internal; namespace MQTTnet.Tests { @@ -189,7 +190,7 @@ namespace MQTTnet.Tests responseSender.ApplicationMessageReceivedAsync += async e => { Assert.IsNull(e.ApplicationMessage.ResponseTopic); - await PlatformAbstractionLayer.CompletedTask; + await CompletedTask.Instance; }; var requestSender = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500)); diff --git a/Source/MQTTnet.Tests/RoundtripTime_Tests.cs b/Source/MQTTnet.Tests/RoundtripTime_Tests.cs index 6ebbab04..01cb9a93 100644 --- a/Source/MQTTnet.Tests/RoundtripTime_Tests.cs +++ b/Source/MQTTnet.Tests/RoundtripTime_Tests.cs @@ -8,7 +8,7 @@ using System.Diagnostics; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Tests.Mockups; namespace MQTTnet.Tests @@ -33,7 +33,7 @@ namespace MQTTnet.Tests receiverClient.ApplicationMessageReceivedAsync += e => { response?.TrySetResult(e.ApplicationMessage.ConvertPayloadToString()); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await receiverClient.SubscribeAsync("#"); diff --git a/Source/MQTTnet.Tests/Server/Assigned_Client_ID_Tests.cs b/Source/MQTTnet.Tests/Server/Assigned_Client_ID_Tests.cs index 8e983be6..cce7c299 100644 --- a/Source/MQTTnet.Tests/Server/Assigned_Client_ID_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Assigned_Client_ID_Tests.cs @@ -7,7 +7,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Formatter; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; namespace MQTTnet.Tests.Server @@ -47,20 +47,20 @@ namespace MQTTnet.Tests.Server e.ReasonCode = MqttConnectReasonCode.Success; } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; testEnvironment.Server.ClientConnectedAsync += args => { serverConnectedClientId = args.ClientId; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; testEnvironment.Server.ClientDisconnectedAsync += args => { serverDisconnectedClientId = args.ClientId; disconnectedMre.Set(); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; // Arrange client @@ -68,7 +68,7 @@ namespace MQTTnet.Tests.Server client.ConnectedAsync += args => { clientAssignedClientId = args.ConnectResult.AssignedClientIdentifier; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; // Act diff --git a/Source/MQTTnet.Tests/Server/Events_Tests.cs b/Source/MQTTnet.Tests/Server/Events_Tests.cs index a8d23361..db453c64 100644 --- a/Source/MQTTnet.Tests/Server/Events_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Events_Tests.cs @@ -7,7 +7,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Formatter; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; using MQTTnet.Server; @@ -27,7 +27,7 @@ namespace MQTTnet.Tests.Server server.ClientConnectedAsync += e => { eventArgs = e; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await testEnvironment.ConnectClient(o => o.WithCredentials("TheUser")); @@ -54,7 +54,7 @@ namespace MQTTnet.Tests.Server server.ClientDisconnectedAsync += e => { eventArgs = e; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client = await testEnvironment.ConnectClient(o => o.WithCredentials("TheUser")); @@ -81,7 +81,7 @@ namespace MQTTnet.Tests.Server server.ClientSubscribedTopicAsync += e => { eventArgs = e; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client = await testEnvironment.ConnectClient(); @@ -108,7 +108,7 @@ namespace MQTTnet.Tests.Server server.ClientUnsubscribedTopicAsync += e => { eventArgs = e; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client = await testEnvironment.ConnectClient(); @@ -134,7 +134,7 @@ namespace MQTTnet.Tests.Server server.InterceptingPublishAsync += e => { eventArgs = e; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client = await testEnvironment.ConnectClient(); @@ -161,7 +161,7 @@ namespace MQTTnet.Tests.Server server.StartedAsync += e => { eventArgs = e; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await server.StartAsync(); @@ -183,7 +183,7 @@ namespace MQTTnet.Tests.Server server.StoppedAsync += e => { eventArgs = e; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await server.StopAsync(); diff --git a/Source/MQTTnet.Tests/Server/General.cs b/Source/MQTTnet.Tests/Server/General.cs index 148bf98c..eeb75fc6 100644 --- a/Source/MQTTnet.Tests/Server/General.cs +++ b/Source/MQTTnet.Tests/Server/General.cs @@ -11,8 +11,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Adapter; using MQTTnet.Client; -using MQTTnet.Formatter; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server; @@ -99,7 +98,7 @@ namespace MQTTnet.Tests.Server server.ValidatingConnectionAsync += e => { e.ReasonCode = MqttConnectReasonCode.NotAuthorized; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var connectingFailedException = await Assert.ThrowsExceptionAsync(() => testEnvironment.ConnectClient()); @@ -122,7 +121,7 @@ namespace MQTTnet.Tests.Server e.Response.ReasonCode = MqttSubscribeReasonCode.UnspecifiedError; } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; // Prepare some retained messages. @@ -146,7 +145,7 @@ namespace MQTTnet.Tests.Server buffer.Append(e.ApplicationMessage.Topic); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client2.SubscribeAsync("y"); @@ -172,13 +171,13 @@ namespace MQTTnet.Tests.Server server.ClientConnectedAsync += e => { Interlocked.Increment(ref clientConnectedCalled); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; server.ClientDisconnectedAsync += e => { Interlocked.Increment(ref clientDisconnectedCalled); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder()); @@ -225,7 +224,7 @@ namespace MQTTnet.Tests.Server // Clear retained message. await client.PublishAsync( new MqttApplicationMessageBuilder().WithTopic("r" + i2) - .WithPayload(PlatformAbstractionLayer.EmptyByteArray) + .WithPayload(EmptyBuffer.Array) .WithRetainFlag() .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build()); @@ -276,7 +275,7 @@ namespace MQTTnet.Tests.Server { e.ApplicationMessage = new MqttApplicationMessage { Topic = "new_topic" }; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; string receivedTopic = null; @@ -286,7 +285,7 @@ namespace MQTTnet.Tests.Server c1.ApplicationMessageReceivedAsync += a => { receivedTopic = a.ApplicationMessage.Topic; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("original_topic").Build()); @@ -305,7 +304,7 @@ namespace MQTTnet.Tests.Server server.InterceptingPublishAsync += e => { e.ApplicationMessage.Payload = Encoding.ASCII.GetBytes("extended"); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var c1 = await testEnvironment.ConnectClient(); @@ -316,7 +315,7 @@ namespace MQTTnet.Tests.Server c2.ApplicationMessageReceivedAsync += e => { isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(e.ApplicationMessage.Payload), StringComparison.Ordinal) == 0; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("test").Build()); @@ -339,7 +338,7 @@ namespace MQTTnet.Tests.Server server.ApplicationMessageNotConsumedAsync += e => { undeliverd = e.ApplicationMessage.Topic; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client = await testEnvironment.ConnectClient(); @@ -376,7 +375,7 @@ namespace MQTTnet.Tests.Server receivedMessages.Add(e.ApplicationMessage); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await Task.Delay(500); @@ -400,7 +399,7 @@ namespace MQTTnet.Tests.Server s.RetainedMessageChangedAsync += e => { savedRetainedMessages = e.StoredRetainedMessages; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var c1 = await testEnvironment.ConnectClient(); @@ -440,7 +439,7 @@ namespace MQTTnet.Tests.Server c1.ApplicationMessageReceivedAsync += e => { receivedTopic = e.ApplicationMessage.Topic; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.SubscribeAsync("#"); @@ -487,7 +486,7 @@ namespace MQTTnet.Tests.Server client.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var message = new MqttApplicationMessageBuilder().WithTopic("a").WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce).Build(); @@ -521,13 +520,13 @@ namespace MQTTnet.Tests.Server c2.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; c3.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c2.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }).ConfigureAwait(false); @@ -581,7 +580,7 @@ namespace MQTTnet.Tests.Server events.Add("c"); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; server.ClientDisconnectedAsync += e => @@ -591,7 +590,7 @@ namespace MQTTnet.Tests.Server events.Add("d"); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var clientOptionsBuilder = new MqttClientOptionsBuilder().WithClientId(Guid.NewGuid().ToString()); @@ -621,7 +620,7 @@ namespace MQTTnet.Tests.Server events.Add("r"); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c2.SubscribeAsync("topic"); @@ -669,7 +668,7 @@ namespace MQTTnet.Tests.Server server.ValidatingConnectionAsync += e => { ConnectionValidationHandler(e); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var events = new List(); @@ -681,7 +680,7 @@ namespace MQTTnet.Tests.Server events.Add("c"); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; server.ClientDisconnectedAsync += e => @@ -691,7 +690,7 @@ namespace MQTTnet.Tests.Server events.Add("d"); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var clientOptions = new MqttClientOptionsBuilder().WithClientId("same_id"); @@ -706,7 +705,7 @@ namespace MQTTnet.Tests.Server events.Add("x"); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; c1.ApplicationMessageReceivedAsync += e => @@ -716,7 +715,7 @@ namespace MQTTnet.Tests.Server events.Add("r"); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; c1.SubscribeAsync("topic").Wait(); @@ -785,7 +784,7 @@ namespace MQTTnet.Tests.Server client1.ApplicationMessageReceivedAsync += e => { receivedBody = e.ApplicationMessage.Payload; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client1.SubscribeAsync("string"); @@ -822,7 +821,7 @@ namespace MQTTnet.Tests.Server receivedMessages.Add(e.ApplicationMessage); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await Task.Delay(500); @@ -850,7 +849,7 @@ namespace MQTTnet.Tests.Server c1.DisconnectedAsync += e => { disconnectCalled++; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await Task.Delay(100); @@ -950,7 +949,7 @@ namespace MQTTnet.Tests.Server testEnvironment.Server.ClientConnectedAsync += e => { testClientId = e.ClientId; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client = testEnvironment.CreateClient(); @@ -959,12 +958,12 @@ namespace MQTTnet.Tests.Server { if (e.Buffer.Length > 0) { - if (e.Buffer[0] == (byte)Protocol.MqttControlPacketType.Disconnect << 4) + if (e.Buffer[0] == (byte)MqttControlPacketType.Disconnect << 4) { disconnectPacketReceived = true; } } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client.ConnectAsync( diff --git a/Source/MQTTnet.Tests/Server/Load_Tests.cs b/Source/MQTTnet.Tests/Server/Load_Tests.cs index b4ddfe47..7cba9c12 100644 --- a/Source/MQTTnet.Tests/Server/Load_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Load_Tests.cs @@ -3,7 +3,7 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -26,7 +26,7 @@ namespace MQTTnet.Tests.Server receiverClient.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessages); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await receiverClient.SubscribeAsync("t/+"); @@ -74,7 +74,7 @@ namespace MQTTnet.Tests.Server server.InterceptingPublishAsync += e => { Interlocked.Increment(ref receivedMessages); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; for (var i = 0; i < 100; i++) @@ -137,7 +137,7 @@ namespace MQTTnet.Tests.Server server.InterceptingPublishAsync += e => { Interlocked.Increment(ref receivedMessages); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; for (var i = 0; i < 100; i++) diff --git a/Source/MQTTnet.Tests/Server/QoS_Tests.cs b/Source/MQTTnet.Tests/Server/QoS_Tests.cs index 483c8d07..08bc6219 100644 --- a/Source/MQTTnet.Tests/Server/QoS_Tests.cs +++ b/Source/MQTTnet.Tests/Server/QoS_Tests.cs @@ -6,7 +6,7 @@ using System.Collections.Generic; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; using MQTTnet.Server; @@ -26,7 +26,7 @@ namespace MQTTnet.Tests.Server server.ClientAcknowledgedPublishPacketAsync += args => { eventArgs = args; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client1 = await testEnvironment.ConnectClient(); @@ -53,7 +53,7 @@ namespace MQTTnet.Tests.Server server.ClientAcknowledgedPublishPacketAsync += args => { eventArgs = args; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client1 = await testEnvironment.ConnectClient(); @@ -88,7 +88,7 @@ namespace MQTTnet.Tests.Server eventArgs.Add(args); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client1 = await testEnvironment.ConnectClient(); diff --git a/Source/MQTTnet.Tests/Server/Security_Tests.cs b/Source/MQTTnet.Tests/Server/Security_Tests.cs index 0e144df8..bc03e9c3 100644 --- a/Source/MQTTnet.Tests/Server/Security_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Security_Tests.cs @@ -8,7 +8,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Adapter; using MQTTnet.Client; using MQTTnet.Exceptions; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; namespace MQTTnet.Tests.Server @@ -33,7 +33,7 @@ namespace MQTTnet.Tests.Server publishedApplicationMessages.Add(eventArgs.ApplicationMessage); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; testEnvironment.Server.ValidatingConnectionAsync += eventArgs => @@ -47,7 +47,7 @@ namespace MQTTnet.Tests.Server eventArgs.ReasonCode = MqttConnectReasonCode.NotAuthorized; } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; using (var validClient = testEnvironment.CreateClient()) @@ -157,7 +157,7 @@ namespace MQTTnet.Tests.Server e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client = testEnvironment.CreateClient(); diff --git a/Source/MQTTnet.Tests/Server/Server_Reference_Tests.cs b/Source/MQTTnet.Tests/Server/Server_Reference_Tests.cs index 19885d96..f8f76c18 100644 --- a/Source/MQTTnet.Tests/Server/Server_Reference_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Server_Reference_Tests.cs @@ -8,6 +8,7 @@ using MQTTnet.Adapter; using MQTTnet.Client; using MQTTnet.Formatter; using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; namespace MQTTnet.Tests.Server @@ -28,7 +29,7 @@ namespace MQTTnet.Tests.Server { e.ReasonCode = MqttConnectReasonCode.ServerMoved; e.ServerReference = "new_server"; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; try diff --git a/Source/MQTTnet.Tests/Server/Session_Tests.cs b/Source/MQTTnet.Tests/Server/Session_Tests.cs index 772b8484..d86d2e8d 100644 --- a/Source/MQTTnet.Tests/Server/Session_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Session_Tests.cs @@ -10,9 +10,8 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Formatter; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; -using MQTTnet.Server; using MQTTnet.Tests.Mockups; namespace MQTTnet.Tests.Server @@ -29,7 +28,7 @@ namespace MQTTnet.Tests.Server await testEnvironment.StartServer(o => o.WithPersistentSessions()); - const string ClientId = "Client1"; + const string clientId = "Client1"; // Create client with clean session and long session expiry interval @@ -38,7 +37,7 @@ namespace MQTTnet.Tests.Server .WithTcpServer("127.0.0.1", testEnvironment.ServerPort) .WithSessionExpiryInterval(9999) // not relevant for v311 but testing impact .WithCleanSession() // start and end with clean session - .WithClientId(ClientId) + .WithClientId(clientId) .Build()); // Disconnect; empty session should be removed from server @@ -57,7 +56,7 @@ namespace MQTTnet.Tests.Server .WithTcpServer("127.0.0.1", testEnvironment.ServerPort) .WithSessionExpiryInterval(9999) // not relevant for v311 but testing impact .WithCleanSession(false) // see if there is a session - .WithClientId(ClientId) + .WithClientId(clientId) .Build(); @@ -83,7 +82,7 @@ namespace MQTTnet.Tests.Server server.SessionDeletedAsync += e => { deletedEventFired = true; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client = await testEnvironment.ConnectClient(); @@ -111,7 +110,7 @@ namespace MQTTnet.Tests.Server e.SessionItems["can_subscribe_x"] = true; e.SessionItems["default_payload"] = "Hello World"; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await testEnvironment.ConnectClient(); @@ -123,6 +122,8 @@ namespace MQTTnet.Tests.Server } } + #if !NET452 + [TestMethod] [DataRow(MqttProtocolVersion.V311)] [DataRow(MqttProtocolVersion.V500)] @@ -134,7 +135,7 @@ namespace MQTTnet.Tests.Server await testEnvironment.StartServer(); - var options = new MqttClientOptionsBuilder().WithClientId("1").WithTimeout(TimeSpan.FromSeconds(1)).WithProtocolVersion(protocolVersion).WithKeepAlivePeriod(TimeSpan.FromSeconds(5)); + var options = new MqttClientOptionsBuilder().WithClientId("1").WithTimeout(TimeSpan.FromSeconds(1)).WithProtocolVersion(protocolVersion); var hasReceive = false; @@ -149,14 +150,14 @@ namespace MQTTnet.Tests.Server // Try to connect 50 clients at the same time. var clients = await Task.WhenAll(Enumerable.Range(0, 50).Select(i => ConnectAndSubscribe(testEnvironment, options, OnReceive))); + await Task.Delay(5000); + + var connectedClients = clients.Where(c => c?.TryPingAsync().GetAwaiter().GetResult() == true).ToList(); + await LongTestDelay(); - var connectedClients = clients.Where(c => c != null).Where(c => c.TryPingAsync().GetAwaiter().GetResult()).ToList(); - Assert.AreEqual(1, connectedClients.Count); - await Task.Delay(5000); - var option2 = new MqttClientOptionsBuilder().WithClientId("2").WithKeepAlivePeriod(TimeSpan.FromSeconds(10)); var sendClient = await testEnvironment.ConnectClient(option2); await sendClient.PublishStringAsync("aaa", "1"); @@ -166,6 +167,7 @@ namespace MQTTnet.Tests.Server Assert.AreEqual(true, hasReceive); } } +#endif [DataTestMethod] [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)] @@ -186,7 +188,7 @@ namespace MQTTnet.Tests.Server c.AutoAcknowledge = false; ++count; Console.WriteLine("process"); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await subscriber.SubscribeAsync("#", qos); @@ -223,7 +225,7 @@ namespace MQTTnet.Tests.Server client1.DisconnectedAsync += c => { disconnectReason = c.Reason; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client2 = await testEnvironment.ConnectClient(options); @@ -249,7 +251,7 @@ namespace MQTTnet.Tests.Server e.SessionItems["can_subscribe_x"] = true; e.SessionItems["default_payload"] = "Hello World"; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; server.InterceptingSubscriptionAsync += e => @@ -262,13 +264,13 @@ namespace MQTTnet.Tests.Server } } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; server.InterceptingPublishAsync += e => { e.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(e.SessionItems["default_payload"] as string ?? string.Empty); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; string receivedPayload = null; @@ -277,7 +279,7 @@ namespace MQTTnet.Tests.Server client.ApplicationMessageReceivedAsync += e => { receivedPayload = e.ApplicationMessage.ConvertPayloadToString(); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var subscribeResult = await client.SubscribeAsync("x"); @@ -327,7 +329,7 @@ namespace MQTTnet.Tests.Server c1.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); @@ -335,10 +337,10 @@ namespace MQTTnet.Tests.Server // C2 has the last will defined. var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithClientId("WillOwner"); - var c2 = await testEnvironment.ConnectClient(clientOptions); + await testEnvironment.ConnectClient(clientOptions); // C3 will do the connection takeover. - var c3 = await testEnvironment.ConnectClient(clientOptions); + await testEnvironment.ConnectClient(clientOptions); await Task.Delay(1000); @@ -350,34 +352,22 @@ namespace MQTTnet.Tests.Server { try { - var sendClient = await testEnvironment.ConnectClient(options); + var client = await testEnvironment.ConnectClient(options).ConfigureAwait(false); - sendClient.ApplicationMessageReceivedAsync += e => + client.ApplicationMessageReceivedAsync += e => { onReceive(); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; - - using (var subscribeTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5))) + + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5))) { - await sendClient.SubscribeAsync("aaa", MqttQualityOfServiceLevel.AtMostOnce, subscribeTimeout.Token).ConfigureAwait(false); + await client.SubscribeAsync("aaa", MqttQualityOfServiceLevel.AtMostOnce, timeout.Token).ConfigureAwait(false); } - return sendClient; + return client; } - catch - { - return null; - } - } - - static async Task TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options) - { - try - { - return await testEnvironment.ConnectClient(options); - } - catch + catch (Exception) { return null; } diff --git a/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs b/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs index 862c2202..b67adea9 100644 --- a/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs @@ -9,7 +9,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Formatter; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server; @@ -33,7 +33,7 @@ namespace MQTTnet.Tests.Server e.Response.ReasonCode = MqttSubscribeReasonCode.TopicFilterInvalid; } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client = await testEnvironment.ConnectClient(); @@ -56,7 +56,7 @@ namespace MQTTnet.Tests.Server server.ClientSubscribedTopicAsync += e => { server.InjectApplicationMessage(new InjectedMqttApplicationMessage(new MqttApplicationMessageBuilder().WithTopic("test_topic").Build())); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var client = await testEnvironment.ConnectClient(); @@ -81,7 +81,7 @@ namespace MQTTnet.Tests.Server { // Set the topic to "a" regards what the client wants to subscribe. e.TopicFilter.Topic = "a"; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var topicAReceived = false; @@ -99,7 +99,7 @@ namespace MQTTnet.Tests.Server topicBReceived = true; } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await client.SubscribeAsync("b"); @@ -146,7 +146,7 @@ namespace MQTTnet.Tests.Server c1.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; for (var i = 0; i < 500; i++) @@ -189,7 +189,7 @@ namespace MQTTnet.Tests.Server c1.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var optionsBuilder = new MqttClientSubscribeOptionsBuilder(); @@ -229,7 +229,7 @@ namespace MQTTnet.Tests.Server c1.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.SubscribeAsync(new MqttClientSubscribeOptionsBuilder().WithTopicFilter("a").Build()); @@ -267,7 +267,7 @@ namespace MQTTnet.Tests.Server c1.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.SubscribeAsync(new MqttClientSubscribeOptionsBuilder().WithTopicFilter("a").WithTopicFilter("b").WithTopicFilter("c").Build()); @@ -301,7 +301,7 @@ namespace MQTTnet.Tests.Server c1.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; var c2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("c2")); @@ -316,7 +316,7 @@ namespace MQTTnet.Tests.Server server.ClientSubscribedTopicAsync += e => { subscribeEventCalled = e.TopicFilter.Topic == "a" && e.ClientId == c1.Options.ClientId; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); @@ -331,7 +331,7 @@ namespace MQTTnet.Tests.Server server.ClientUnsubscribedTopicAsync += e => { unsubscribeEventCalled = e.TopicFilter == "a" && e.ClientId == c1.Options.ClientId; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.UnsubscribeAsync("a"); diff --git a/Source/MQTTnet.Tests/Server/Topic_Alias_Tests.cs b/Source/MQTTnet.Tests/Server/Topic_Alias_Tests.cs index b15c3b1a..a92b023e 100644 --- a/Source/MQTTnet.Tests/Server/Topic_Alias_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Topic_Alias_Tests.cs @@ -8,7 +8,7 @@ using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Formatter; -using MQTTnet.Implementations; +using MQTTnet.Internal; namespace MQTTnet.Tests.Server { @@ -50,7 +50,7 @@ namespace MQTTnet.Tests.Server receivedTopics.Add(e.ApplicationMessage.Topic); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.SubscribeAsync("#"); diff --git a/Source/MQTTnet.Tests/Server/User_Properties_Tests.cs b/Source/MQTTnet.Tests/Server/User_Properties_Tests.cs index 27f3e72d..0bbc7030 100644 --- a/Source/MQTTnet.Tests/Server/User_Properties_Tests.cs +++ b/Source/MQTTnet.Tests/Server/User_Properties_Tests.cs @@ -9,6 +9,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Formatter; using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Tests.Mockups; @@ -48,7 +49,7 @@ namespace MQTTnet.Tests.Server receiver.ApplicationMessageReceivedAsync += e => { receivedMessage = e.ApplicationMessage; - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await sender.PublishAsync(message.Build(), CancellationToken.None); diff --git a/Source/MQTTnet.Tests/Server/Will_Tests.cs b/Source/MQTTnet.Tests/Server/Will_Tests.cs index d464cd7e..12b24f63 100644 --- a/Source/MQTTnet.Tests/Server/Will_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Will_Tests.cs @@ -2,7 +2,7 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; -using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Protocol; namespace MQTTnet.Tests.Server @@ -26,7 +26,7 @@ namespace MQTTnet.Tests.Server c1.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); @@ -53,7 +53,7 @@ namespace MQTTnet.Tests.Server c1.ApplicationMessageReceivedAsync += e => { Interlocked.Increment(ref receivedMessagesCount); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; }; await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build()); diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 5af179f2..7a9db45d 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -195,7 +195,7 @@ namespace MQTTnet.Adapter // This lock makes sure that multiple threads can send packets at the same time. // This is required when a disconnect is sent from another thread while the // worker thread is still sending publish packets etc. - using (await _syncRoot.WaitAsync(cancellationToken).ConfigureAwait(false)) + using (await _syncRoot.EnterAsync(cancellationToken).ConfigureAwait(false)) { // Check for cancellation here again because "WaitAsync" might take some time. cancellationToken.ThrowIfCancellationRequested(); @@ -363,7 +363,7 @@ namespace MQTTnet.Adapter var fixedHeader = readFixedHeaderResult.FixedHeader; if (fixedHeader.RemainingLength == 0) { - return new ReceivedMqttPacket(fixedHeader.Flags, PlatformAbstractionLayer.EmptyByteArraySegment, 2); + return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2); } var bodyLength = fixedHeader.RemainingLength; diff --git a/Source/MQTTnet/Client/IMqttClient.cs b/Source/MQTTnet/Client/IMqttClient.cs index ada0ad0f..849bff12 100644 --- a/Source/MQTTnet/Client/IMqttClient.cs +++ b/Source/MQTTnet/Client/IMqttClient.cs @@ -8,6 +8,7 @@ namespace MQTTnet.Client public interface IMqttClient : IDisposable { event Func ApplicationMessageReceivedAsync; + event Func ConnectedAsync; event Func ConnectingAsync; diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 00b0e3aa..92004fe8 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -9,7 +9,6 @@ using MQTTnet.Adapter; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Formatter; -using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.PacketDispatcher; using MQTTnet.Packets; @@ -224,17 +223,17 @@ namespace MQTTnet.Client public async Task PingAsync(CancellationToken cancellationToken = default) { - if (!cancellationToken.CanBeCanceled) + if (cancellationToken.CanBeCanceled) + { + await SendAndReceiveAsync(MqttPingReqPacket.Instance, cancellationToken).ConfigureAwait(false); + } + else { using (var timeout = new CancellationTokenSource(Options.Timeout)) { await SendAndReceiveAsync(MqttPingReqPacket.Instance, timeout.Token).ConfigureAwait(false); } } - else - { - await SendAndReceiveAsync(MqttPingReqPacket.Instance, cancellationToken).ConfigureAwait(false); - } } public Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default) @@ -369,7 +368,7 @@ namespace MQTTnet.Client throw new MqttProtocolViolationException("Received a not supported QoS level."); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } async Task AuthenticateAsync(MqttClientOptions options, CancellationToken cancellationToken) @@ -490,7 +489,7 @@ namespace MQTTnet.Client return DisconnectCoreAsync(sender, exception, connectResult, clientWasConnected); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } bool DisconnectIsPendingOrFinished() @@ -550,7 +549,7 @@ namespace MQTTnet.Client return extendedAuthenticationExchangeHandler.HandleRequestAsync(new MqttExtendedAuthenticationExchangeContext(authPacket, this)); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } Task ProcessReceivedDisconnectPacket(MqttDisconnectPacket disconnectPacket) @@ -604,7 +603,7 @@ namespace MQTTnet.Client return SendAsync(pubRelPacket, cancellationToken); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) diff --git a/Source/MQTTnet/Formatter/MqttBufferReader.cs b/Source/MQTTnet/Formatter/MqttBufferReader.cs index 4c5cc36e..665186ab 100644 --- a/Source/MQTTnet/Formatter/MqttBufferReader.cs +++ b/Source/MQTTnet/Formatter/MqttBufferReader.cs @@ -6,13 +6,13 @@ using System; using System.Runtime.CompilerServices; using System.Text; using MQTTnet.Exceptions; -using MQTTnet.Implementations; +using MQTTnet.Internal; namespace MQTTnet.Formatter { public sealed class MqttBufferReader { - byte[] _buffer = PlatformAbstractionLayer.EmptyByteArray; + byte[] _buffer = EmptyBuffer.Array; int _initialPosition; int _length; int _position; @@ -29,7 +29,7 @@ namespace MQTTnet.Formatter if (length == 0) { - return PlatformAbstractionLayer.EmptyByteArray; + return EmptyBuffer.Array; } ValidateReceiveBuffer(length); @@ -66,7 +66,7 @@ namespace MQTTnet.Formatter if (bufferLength == 0) { - return PlatformAbstractionLayer.EmptyByteArray; + return EmptyBuffer.Array; } var buffer = new byte[bufferLength]; diff --git a/Source/MQTTnet/Formatter/MqttPacketBuffer.cs b/Source/MQTTnet/Formatter/MqttPacketBuffer.cs index 4fdd77ee..fb8692f5 100644 --- a/Source/MQTTnet/Formatter/MqttPacketBuffer.cs +++ b/Source/MQTTnet/Formatter/MqttPacketBuffer.cs @@ -5,12 +5,13 @@ using System; using System.Linq; using MQTTnet.Implementations; +using MQTTnet.Internal; namespace MQTTnet.Formatter { public readonly struct MqttPacketBuffer { - static readonly ArraySegment EmptyPayload = PlatformAbstractionLayer.EmptyByteArraySegment; + static readonly ArraySegment EmptyPayload = EmptyBuffer.ArraySegment; public MqttPacketBuffer(ArraySegment packet, ArraySegment payload) { diff --git a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs index 1d14ffcd..f60bdab6 100644 --- a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs +++ b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs @@ -15,14 +15,20 @@ namespace MQTTnet.Implementations public sealed class CrossPlatformSocket : IDisposable { readonly Socket _socket; + +#if !NET5_0_OR_GREATER readonly Action _socketDisposeAction; +#endif NetworkStream _networkStream; public CrossPlatformSocket(AddressFamily addressFamily) { _socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp); + +#if !NET5_0_OR_GREATER _socketDisposeAction = _socket.Dispose; +#endif } public CrossPlatformSocket() @@ -31,17 +37,76 @@ namespace MQTTnet.Implementations // will make use of dual mode in the .net framework. _socket = new Socket(SocketType.Stream, ProtocolType.Tcp); +#if !NET5_0_OR_GREATER _socketDisposeAction = _socket.Dispose; +#endif } - public CrossPlatformSocket(Socket socket) + CrossPlatformSocket(Socket socket) { _socket = socket ?? throw new ArgumentNullException(nameof(socket)); _networkStream = new NetworkStream(socket, true); +#if !NET5_0_OR_GREATER _socketDisposeAction = _socket.Dispose; +#endif } + public bool DualMode + { + get => _socket.DualMode; + set => _socket.DualMode = value; + } + + public bool IsConnected => _socket.Connected; + + public bool KeepAlive + { + get => _socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive) as int? == 1; + set => _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, value ? 1 : 0); + } + + public int TcpKeepAliveInterval + { +#if NETCOREAPP3_0_OR_GREATER + get => _socket.GetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveInterval) as int? ?? 0; + set => _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveInterval, value); +#else + get { throw new NotSupportedException("TcpKeepAliveInterval requires at least net5.0."); } + set { throw new NotSupportedException("TcpKeepAliveInterval requires at least net5.0."); } +#endif + } + + public int TcpKeepAliveRetryCount + { +#if NETCOREAPP3_0_OR_GREATER + get => _socket.GetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveRetryCount) as int? ?? 0; + set => _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveRetryCount, value); +#else + get { throw new NotSupportedException("TcpKeepAliveRetryCount requires at least net5.0."); } + set { throw new NotSupportedException("TcpKeepAliveRetryCount requires at least net5.0."); } +#endif + } + + public int TcpKeepAliveTime + { +#if NETCOREAPP3_0_OR_GREATER + get => _socket.GetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime) as int? ?? 0; + set => _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, value); +#else + get { throw new NotSupportedException("TcpKeepAliveTime requires at least net5.0."); } + set { throw new NotSupportedException("TcpKeepAliveTime requires at least net5.0."); } +#endif + } + + public LingerOption LingerState + { + get => _socket.LingerState; + set => _socket.LingerState = value; + } + + public EndPoint LocalEndPoint => _socket.LocalEndPoint; + public bool NoDelay { // We cannot use the _NoDelay_ property from the socket because there is an issue in .NET 4.5.2, 4.6. @@ -51,24 +116,20 @@ namespace MQTTnet.Implementations set => _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, value ? 1 : 0); } - public LingerOption LingerState - { - get => _socket.LingerState; - set => _socket.LingerState = value; - } - - public bool DualMode - { - get => _socket.DualMode; - set => _socket.DualMode = value; - } - public int ReceiveBufferSize { get => _socket.ReceiveBufferSize; set => _socket.ReceiveBufferSize = value; } + public EndPoint RemoteEndPoint => _socket.RemoteEndPoint; + + public bool ReuseAddress + { + get => _socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress) as int? != 0; + set => _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, value ? 1 : 0); + } + public int SendBufferSize { get => _socket.SendBufferSize; @@ -81,16 +142,6 @@ namespace MQTTnet.Implementations set => _socket.SendTimeout = value; } - public EndPoint RemoteEndPoint => _socket.RemoteEndPoint; - - public bool ReuseAddress - { - get => (int)_socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress) != 0; - set => _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, value ? 1 : 0); - } - - public bool IsConnected => _socket.Connected; - public async Task AcceptAsync() { try @@ -109,8 +160,6 @@ namespace MQTTnet.Implementations } } - public EndPoint LocalEndPoint => _socket.LocalEndPoint; - public void Bind(EndPoint localEndPoint) { if (localEndPoint is null) @@ -121,11 +170,6 @@ namespace MQTTnet.Implementations _socket.Bind(localEndPoint); } - public void Listen(int connectionBacklog) - { - _socket.Listen(connectionBacklog); - } - public async Task ConnectAsync(string host, int port, CancellationToken cancellationToken) { if (host is null) @@ -137,11 +181,17 @@ namespace MQTTnet.Implementations try { - _networkStream?.Dispose(); - #if NET5_0_OR_GREATER + + if (_networkStream != null) + { + await _networkStream.DisposeAsync().ConfigureAwait(false); + } + await _socket.ConnectAsync(host, port, cancellationToken).ConfigureAwait(false); #else + _networkStream?.Dispose(); + // Workaround for: https://github.com/dotnet/corefx/issues/24430 using (cancellationToken.Register(_socketDisposeAction)) { @@ -175,20 +225,26 @@ namespace MQTTnet.Implementations } } - public async Task SendAsync(ArraySegment buffer, SocketFlags socketFlags) + public void Dispose() { - try + _networkStream?.Dispose(); + _socket?.Dispose(); + } + + public NetworkStream GetStream() + { + var networkStream = _networkStream; + if (networkStream == null) { -#if NET452 || NET461 - await Task.Factory.FromAsync(SocketWrapper.BeginSend, _socket.EndSend, new SocketWrapper(_socket, buffer, socketFlags)).ConfigureAwait(false); -#else - await _socket.SendAsync(buffer, socketFlags).ConfigureAwait(false); -#endif - } - catch (ObjectDisposedException) - { - // This will happen when _socket.EndConnect gets called by Task library but the socket is already disposed. + throw new IOException("The socket is not connected."); } + + return networkStream; + } + + public void Listen(int connectionBacklog) + { + _socket.Listen(connectionBacklog); } public async Task ReceiveAsync(ArraySegment buffer, SocketFlags socketFlags) @@ -208,28 +264,27 @@ namespace MQTTnet.Implementations } } - public NetworkStream GetStream() + public async Task SendAsync(ArraySegment buffer, SocketFlags socketFlags) { - var networkStream = _networkStream; - if (networkStream == null) + try { - throw new IOException("The socket is not connected."); +#if NET452 || NET461 + await Task.Factory.FromAsync(SocketWrapper.BeginSend, _socket.EndSend, new SocketWrapper(_socket, buffer, socketFlags)).ConfigureAwait(false); +#else + await _socket.SendAsync(buffer, socketFlags).ConfigureAwait(false); +#endif + } + catch (ObjectDisposedException) + { + // This will happen when _socket.EndConnect gets called by Task library but the socket is already disposed. } - - return networkStream; - } - - public void Dispose() - { - _networkStream?.Dispose(); - _socket?.Dispose(); } #if NET452 || NET461 sealed class SocketWrapper { - readonly Socket _socket; readonly ArraySegment _buffer; + readonly Socket _socket; readonly SocketFlags _socketFlags; public SocketWrapper(Socket socket, ArraySegment buffer, SocketFlags socketFlags) @@ -239,16 +294,28 @@ namespace MQTTnet.Implementations _socketFlags = socketFlags; } - public static IAsyncResult BeginSend(AsyncCallback callback, object state) - { - var socketWrapper = (SocketWrapper)state; - return socketWrapper._socket.BeginSend(socketWrapper._buffer.Array, socketWrapper._buffer.Offset, socketWrapper._buffer.Count, socketWrapper._socketFlags, callback, state); - } - public static IAsyncResult BeginReceive(AsyncCallback callback, object state) { var socketWrapper = (SocketWrapper)state; - return socketWrapper._socket.BeginReceive(socketWrapper._buffer.Array, socketWrapper._buffer.Offset, socketWrapper._buffer.Count, socketWrapper._socketFlags, callback, state); + return socketWrapper._socket.BeginReceive( + socketWrapper._buffer.Array, + socketWrapper._buffer.Offset, + socketWrapper._buffer.Count, + socketWrapper._socketFlags, + callback, + state); + } + + public static IAsyncResult BeginSend(AsyncCallback callback, object state) + { + var socketWrapper = (SocketWrapper)state; + return socketWrapper._socket.BeginSend( + socketWrapper._buffer.Array, + socketWrapper._buffer.Offset, + socketWrapper._buffer.Count, + socketWrapper._socketFlags, + callback, + state); } } #endif diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index ad2c58bf..c1e0b316 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -14,6 +14,7 @@ using System.Runtime.ExceptionServices; using System.Security.Cryptography.X509Certificates; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Internal; namespace MQTTnet.Implementations { @@ -154,7 +155,7 @@ namespace MQTTnet.Implementations public Task DisconnectAsync(CancellationToken cancellationToken) { Dispose(); - return Task.FromResult(0); + return CompletedTask.Instance; } public void Dispose() diff --git a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs index ba1ea40d..ac94261d 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs @@ -13,6 +13,7 @@ using System.Net.Sockets; using System.Security.Cryptography.X509Certificates; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Internal; namespace MQTTnet.Implementations { @@ -56,13 +57,13 @@ namespace MQTTnet.Implementations RegisterListeners(options.TlsEndpointOptions, tlsCertificate, logger, _cancellationTokenSource.Token); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } public Task StopAsync() { Cleanup(); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } public void Dispose() @@ -124,7 +125,7 @@ namespace MQTTnet.Implementations var clientHandler = ClientHandler; if (clientHandler == null) { - return Task.FromResult(0); + return CompletedTask.Instance; } return clientHandler(channelAdapter); diff --git a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs index e39d545e..90e3666e 100644 --- a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs +++ b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs @@ -117,7 +117,7 @@ namespace MQTTnet.Implementations return; } - using (await _sendLock.WaitAsync(cancellationToken).ConfigureAwait(false)) + using (await _sendLock.EnterAsync(cancellationToken).ConfigureAwait(false)) { await _webSocket.SendAsync(buffer, WebSocketMessageType.Binary, isEndOfPacket, cancellationToken).ConfigureAwait(false); } diff --git a/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs b/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs deleted file mode 100644 index d8803349..00000000 --- a/Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs +++ /dev/null @@ -1,43 +0,0 @@ -// Licensed to the .NET Foundation under one or more agreements. -// The .NET Foundation licenses this file to you under the MIT license. -// See the LICENSE file in the project root for more information. - -using System; -using System.Threading; -using System.Threading.Tasks; - -namespace MQTTnet.Implementations -{ - public static class PlatformAbstractionLayer - { -#if NET452 - public static Task CompletedTask { get; } = Task.FromResult(0); - - public static byte[] EmptyByteArray { get; } = new byte[0]; -#else - public static Task CompletedTask { get; } = Task.CompletedTask; - - public static byte[] EmptyByteArray { get; } = Array.Empty(); -#endif - - public static ArraySegment EmptyByteArraySegment { get; } = new ArraySegment(EmptyByteArray); - - public static void Sleep(TimeSpan timeout) - { -#if !NETSTANDARD1_3 && !WINDOWS_UWP - try - { - System.Threading.Thread.Sleep(timeout); - } - catch (ThreadAbortException) - { - // The ThreadAbortException is not actively catched in this project. - // So we use a one which is similar and will be catched properly. - throw new OperationCanceledException(); - } -#else - Task.Delay(timeout).Wait(); -#endif - } - } -} diff --git a/Source/MQTTnet/Internal/AsyncEvent.cs b/Source/MQTTnet/Internal/AsyncEvent.cs index cd67b241..8e988d75 100644 --- a/Source/MQTTnet/Internal/AsyncEvent.cs +++ b/Source/MQTTnet/Internal/AsyncEvent.cs @@ -4,7 +4,6 @@ using System; using System.Collections.Generic; -using System.Linq; using System.Threading.Tasks; using MQTTnet.Diagnostics; @@ -37,7 +36,7 @@ namespace MQTTnet.Internal _handlers.Add(new AsyncEventInvocator(null, handler)); HasHandlers = true; - _handlersForInvoke = new List>(_handlers.ToList()); + _handlersForInvoke = new List>(_handlers); } } @@ -53,7 +52,7 @@ namespace MQTTnet.Internal _handlers.Add(new AsyncEventInvocator(handler, null)); HasHandlers = true; - _handlersForInvoke = new List>(_handlers.ToList()); + _handlersForInvoke = new List>(_handlers); } } @@ -86,7 +85,7 @@ namespace MQTTnet.Internal _handlers.RemoveAll(h => h.WrapsHandler(handler)); HasHandlers = _handlers.Count > 0; - _handlersForInvoke = new List>(_handlers.ToList()); + _handlersForInvoke = new List>(_handlers); } } @@ -102,7 +101,7 @@ namespace MQTTnet.Internal _handlers.RemoveAll(h => h.WrapsHandler(handler)); HasHandlers = _handlers.Count > 0; - _handlersForInvoke = new List>(_handlers.ToList()); + _handlersForInvoke = new List>(_handlers); } } diff --git a/Source/MQTTnet/Internal/AsyncEventInvocator.cs b/Source/MQTTnet/Internal/AsyncEventInvocator.cs index 6258b28b..f344cb1e 100644 --- a/Source/MQTTnet/Internal/AsyncEventInvocator.cs +++ b/Source/MQTTnet/Internal/AsyncEventInvocator.cs @@ -36,7 +36,7 @@ namespace MQTTnet.Internal if (_handler != null) { _handler.Invoke(eventArgs); - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } if (_asyncHandler != null) diff --git a/Source/MQTTnet/Internal/AsyncLock.cs b/Source/MQTTnet/Internal/AsyncLock.cs index 17384ef3..580d570d 100644 --- a/Source/MQTTnet/Internal/AsyncLock.cs +++ b/Source/MQTTnet/Internal/AsyncLock.cs @@ -4,7 +4,7 @@ using System; using System.Collections.Generic; -using System.Diagnostics; +using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -12,173 +12,151 @@ namespace MQTTnet.Internal { public sealed class AsyncLock : IDisposable { - /* - * This async supporting lock does not support reentrancy! - */ - - readonly List _queuedTasks = new List(64); - readonly Task _releaserTaskWithDirectApproval; - - readonly Releaser _releaserWithDirectApproval; - + readonly Task _completedTask; + readonly IDisposable _releaser; readonly object _syncRoot = new object(); + readonly Queue _waiters = new Queue(64); + + volatile bool _isDisposed; + bool _isLocked; - bool _isDisposed; - public AsyncLock() { - _releaserWithDirectApproval = new Releaser(this, null, CancellationToken.None); - _releaserTaskWithDirectApproval = Task.FromResult((IDisposable)_releaserWithDirectApproval); + _releaser = new Releaser(this); + _completedTask = Task.FromResult(_releaser); } public void Dispose() { lock (_syncRoot) { - foreach (var waitingTask in _queuedTasks) - { - waitingTask.Fail(new ObjectDisposedException(nameof(AsyncLock))); - } - - _queuedTasks.Clear(); - _isDisposed = true; + + while (_waiters.Any()) + { + _waiters.Dequeue().Dispose(); + } } } - public Task WaitAsync(CancellationToken cancellationToken) + public Task EnterAsync(CancellationToken cancellationToken = default) { - var hasDirectApproval = false; - Releaser releaser; + cancellationToken.ThrowIfCancellationRequested(); + + if (_isDisposed) + { + throw new ObjectDisposedException(nameof(AsyncLock)); + } lock (_syncRoot) { - if (_isDisposed) + if (!_isLocked) { - throw new ObjectDisposedException(nameof(AsyncLock)); - } - - if (_queuedTasks.Count == 0) - { - // There is no other waiting task apart from the current one. - // So we can approve the current task directly. - releaser = _releaserWithDirectApproval; - hasDirectApproval = true; - Debug.WriteLine("AsyncLock: Task -1 directly approved."); - } - else - { - releaser = new Releaser(this, new TaskCompletionSource(), cancellationToken); + _isLocked = true; + return _completedTask; } - _queuedTasks.Add(releaser); + var waiter = new AsyncLockWaiter(cancellationToken); + _waiters.Enqueue(waiter); + + return waiter.Task; } - - if (!hasDirectApproval) - { - return releaser.Task; - } - - return _releaserTaskWithDirectApproval; } - void Release(Releaser releaser) + void Release() { lock (_syncRoot) { if (_isDisposed) { - // There is no much left to do! + // All waiters have been canceled with a ObjectDisposedException. + // So there is nothing left to do. return; } - var activeTask = _queuedTasks[0]; - if (!ReferenceEquals(activeTask, releaser)) - { - throw new InvalidOperationException("The active task must be the current releaser."); - } + // Assume that there is no waiter left first. + _isLocked = false; - _queuedTasks.RemoveAt(0); - - while (_queuedTasks.Count > 0) + // Try to find the next waiter which can be approved. + // Some of them might be canceled already so it is not + // guaranteed that the very next waiter is the correct one. + while (_waiters.Any()) { - var nextTask = _queuedTasks[0]; - if (!nextTask.IsPending) + var waiter = _waiters.Dequeue(); + var isApproved = waiter.Approve(_releaser); + waiter.Dispose(); + + if (isApproved) { - // Dequeue all canceled or failed tasks. - _queuedTasks.RemoveAt(0); - continue; + _isLocked = true; + return; } - - nextTask.Approve(); - return; } - - Debug.WriteLine("AsyncLock: No Task pending."); } } - sealed class Releaser : IDisposable + sealed class AsyncLockWaiter : IDisposable { - readonly AsyncLock _asyncLock; - readonly CancellationToken _cancellationToken; - readonly int _id; - readonly TaskCompletionSource _promise; + readonly CancellationTokenRegistration _cancellationRegistration; + readonly bool _hasCancellationRegistration; + readonly AsyncTaskCompletionSource _promise = new AsyncTaskCompletionSource(); - // ReSharper disable once FieldCanBeMadeReadOnly.Local - CancellationTokenRegistration _cancellationTokenRegistration; - - internal Releaser(AsyncLock asyncLock, TaskCompletionSource promise, CancellationToken cancellationToken) + public AsyncLockWaiter(CancellationToken cancellationToken) { - _asyncLock = asyncLock ?? throw new ArgumentNullException(nameof(asyncLock)); - _promise = promise; - _cancellationToken = cancellationToken; - + cancellationToken.ThrowIfCancellationRequested(); + if (cancellationToken.CanBeCanceled) { - _cancellationTokenRegistration = cancellationToken.Register(Cancel); + _cancellationRegistration = cancellationToken.Register(Cancel); + _hasCancellationRegistration = true; } - - _id = promise?.Task.Id ?? -1; - - Debug.WriteLine($"AsyncLock: Task {_id} queued."); } - public bool IsPending => _promise != null && !_promise.Task.IsCanceled && !_promise.Task.IsFaulted && !_promise.Task.IsCompleted; + public Task Task => _promise.Task; - public Task Task => _promise?.Task; - - public void Approve() + public bool Approve(IDisposable scope) { - _promise?.TrySetResult(this); + if (scope == null) + { + throw new ArgumentNullException(nameof(scope)); + } - Debug.WriteLine($"AsyncLock: Task {_id} approved."); + if (_promise.Task.IsCompleted) + { + return false; + } + + return _promise.TrySetResult(scope); } public void Dispose() { - if (_cancellationToken.CanBeCanceled) + if (_hasCancellationRegistration) { - _cancellationTokenRegistration.Dispose(); + _cancellationRegistration.Dispose(); } - Debug.WriteLine($"AsyncLock: Task {_id} completed."); - - _asyncLock.Release(this); - } - - public void Fail(Exception exception) - { - _promise?.TrySetException(exception); - - Debug.WriteLine($"AsyncLock: Task {_id} failed ({exception.GetType().Name})."); + _promise.TrySetException(new ObjectDisposedException(nameof(AsyncLockWaiter))); } void Cancel() { - _promise?.TrySetCanceled(); + _promise.TrySetCanceled(); + } + } - Debug.WriteLine($"AsyncLock: Task {_id} canceled."); + readonly struct Releaser : IDisposable + { + readonly AsyncLock _asyncLock; + + public Releaser(AsyncLock asyncLock) + { + _asyncLock = asyncLock ?? throw new ArgumentNullException(nameof(asyncLock)); + } + + public void Dispose() + { + _asyncLock.Release(); } } } diff --git a/Source/MQTTnet/Internal/AsyncTaskCompletionSource.cs b/Source/MQTTnet/Internal/AsyncTaskCompletionSource.cs index a77b56d9..d96c2b5f 100644 --- a/Source/MQTTnet/Internal/AsyncTaskCompletionSource.cs +++ b/Source/MQTTnet/Internal/AsyncTaskCompletionSource.cs @@ -3,6 +3,7 @@ // See the LICENSE file in the project root for more information. using System; +using System.Threading; using System.Threading.Tasks; namespace MQTTnet.Internal @@ -22,7 +23,7 @@ namespace MQTTnet.Internal public Task Task => _taskCompletionSource.Task; - public bool TrySetCanceled() + public void TrySetCanceled() { #if NET452 // To prevent deadlocks it is required to call the _TrySetCanceled_ method @@ -31,19 +32,14 @@ namespace MQTTnet.Internal // do it. But _this_ thread is also reading incoming packets -> deadlock. // NET452 does not support RunContinuationsAsynchronously System.Threading.Tasks.Task.Run(() => _taskCompletionSource.TrySetCanceled()); - return true; + SpinWait.SpinUntil(() => _taskCompletionSource.Task.IsCompleted); #else - return _taskCompletionSource.TrySetCanceled(); + _taskCompletionSource.TrySetCanceled(); #endif } - public bool TrySetException(Exception exception) + public void TrySetException(Exception exception) { - if (exception == null) - { - throw new ArgumentNullException(nameof(exception)); - } - #if NET452 // To prevent deadlocks it is required to call the _TrySetException_ method // from a new thread because the awaiting code will not(!) be executed in @@ -51,9 +47,9 @@ namespace MQTTnet.Internal // do it. But _this_ thread is also reading incoming packets -> deadlock. // NET452 does not support RunContinuationsAsynchronously System.Threading.Tasks.Task.Run(() => _taskCompletionSource.TrySetException(exception)); - return true; + SpinWait.SpinUntil(() => _taskCompletionSource.Task.IsCompleted); #else - return _taskCompletionSource.TrySetException(exception); + _taskCompletionSource.TrySetException(exception); #endif } @@ -65,7 +61,14 @@ namespace MQTTnet.Internal // a new thread automatically (due to await). Furthermore _this_ thread will // do it. But _this_ thread is also reading incoming packets -> deadlock. // NET452 does not support RunContinuationsAsynchronously + if (_taskCompletionSource.Task.IsCompleted) + { + return false; + } + System.Threading.Tasks.Task.Run(() => _taskCompletionSource.TrySetResult(result)); + SpinWait.SpinUntil(() => _taskCompletionSource.Task.IsCompleted); + return true; #else return _taskCompletionSource.TrySetResult(result); diff --git a/Source/MQTTnet/Internal/CompletedTask.cs b/Source/MQTTnet/Internal/CompletedTask.cs new file mode 100644 index 00000000..1f050842 --- /dev/null +++ b/Source/MQTTnet/Internal/CompletedTask.cs @@ -0,0 +1,17 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System.Threading.Tasks; + +namespace MQTTnet.Internal +{ + public static class CompletedTask + { +#if NET452 + public static readonly Task Instance = Task.FromResult(true); +#else + public static readonly Task Instance = Task.CompletedTask; +#endif + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Internal/EmptyBuffer.cs b/Source/MQTTnet/Internal/EmptyBuffer.cs new file mode 100644 index 00000000..5992a741 --- /dev/null +++ b/Source/MQTTnet/Internal/EmptyBuffer.cs @@ -0,0 +1,19 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; + +namespace MQTTnet.Internal +{ + public static class EmptyBuffer + { +#if NET452 + public static readonly byte[] Array = new byte[0]; +#else + public static readonly byte[] Array = System.Array.Empty(); +#endif + + public static readonly ArraySegment ArraySegment = new ArraySegment(Array, 0, 0); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index 35b1536f..c8bf7f0a 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -14,7 +14,6 @@ $(TargetFrameworks);net452;net461 $(TargetFrameworks);uap10.0 7.3 - MQTTnet MQTTnet True diff --git a/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs index eaa916e1..35b70369 100644 --- a/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs @@ -10,6 +10,7 @@ using System.Text; using MQTTnet.Adapter; using MQTTnet.Formatter; using MQTTnet.Implementations; +using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -88,7 +89,7 @@ namespace MQTTnet.Server /// public uint MaximumPacketSize => _connectPacket.MaximumPacketSize; - public string Password => Encoding.UTF8.GetString(RawPassword ?? PlatformAbstractionLayer.EmptyByteArray); + public string Password => Encoding.UTF8.GetString(RawPassword ?? EmptyBuffer.Array); public MqttProtocolVersion ProtocolVersion => ChannelAdapter.PacketFormatterAdapter.ProtocolVersion; diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 18c3804a..4ae2d40d 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -181,7 +181,7 @@ namespace MQTTnet.Server return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } Task HandleIncomingPubAckPacket(MqttPubAckPacket pubAckPacket) @@ -193,7 +193,7 @@ namespace MQTTnet.Server return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubAckPacket); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } Task HandleIncomingPubCompPacket(MqttPubCompPacket pubCompPacket) @@ -205,7 +205,7 @@ namespace MQTTnet.Server return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubCompPacket); } - return PlatformAbstractionLayer.CompletedTask; + return CompletedTask.Instance; } async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken) diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 8a2c2364..308e9416 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -306,6 +306,9 @@ namespace MQTTnet.Server await client.RunAsync().ConfigureAwait(false); } + catch (ObjectDisposedException) + { + } catch (OperationCanceledException) { } @@ -479,7 +482,7 @@ namespace MQTTnet.Server sessionShouldPersist = !connectPacket.CleanSession; } - using (await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _createConnectionSyncRoot.EnterAsync().ConfigureAwait(false)) { MqttSession session; lock (_sessionsManagementLock) @@ -537,6 +540,7 @@ namespace MQTTnet.Server MqttClientDisconnectType.Takeover, existingClient.Endpoint, existingClient.Session.Items); + await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } } diff --git a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs index acc4672d..a9b86a7a 100644 --- a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs @@ -5,10 +5,8 @@ using System; using System.Collections.Generic; using System.Linq; -using System.Threading; using System.Threading.Tasks; using MQTTnet.Diagnostics; -using MQTTnet.Implementations; using MQTTnet.Internal; namespace MQTTnet.Server @@ -85,7 +83,7 @@ namespace MQTTnet.Server } else { - if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? PlatformAbstractionLayer.EmptyByteArray)) + if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? EmptyBuffer.Array)) { _messages[applicationMessage.Topic] = applicationMessage; saveIsRequired = true; @@ -103,7 +101,7 @@ namespace MQTTnet.Server if (saveIsRequired) { - using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _storageAccessLock.EnterAsync().ConfigureAwait(false)) { var eventArgs = new RetainedMessageChangedEventArgs(clientId, applicationMessage, messagesForSave); await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); @@ -132,7 +130,7 @@ namespace MQTTnet.Server _messages.Clear(); } - using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _storageAccessLock.EnterAsync().ConfigureAwait(false)) { await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); } diff --git a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs index de2bd442..3b193446 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs @@ -6,7 +6,6 @@ using System; using System.Threading; using System.Threading.Tasks; using MQTTnet.Diagnostics; -using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.Protocol; @@ -14,16 +13,20 @@ namespace MQTTnet.Server { public sealed class MqttServerKeepAliveMonitor { + readonly MqttNetSourceLogger _logger; readonly MqttServerOptions _options; readonly MqttClientSessionsManager _sessionsManager; - readonly MqttNetSourceLogger _logger; public MqttServerKeepAliveMonitor(MqttServerOptions options, MqttClientSessionsManager sessionsManager, IMqttNetLogger logger) { _options = options ?? throw new ArgumentNullException(nameof(options)); _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager)); - - if (logger == null) throw new ArgumentNullException(nameof(logger)); + + if (logger == null) + { + throw new ArgumentNullException(nameof(logger)); + } + _logger = logger.WithSource(nameof(MqttServerKeepAliveMonitor)); } @@ -45,7 +48,7 @@ namespace MQTTnet.Server while (!cancellationToken.IsCancellationRequested) { TryProcessClients(); - PlatformAbstractionLayer.Sleep(_options.KeepAliveMonitorInterval); + Sleep(_options.KeepAliveMonitorInterval); } } catch (OperationCanceledException) @@ -61,13 +64,22 @@ namespace MQTTnet.Server } } - void TryProcessClients() + static void Sleep(TimeSpan timeout) { - var now = DateTime.UtcNow; - foreach (var client in _sessionsManager.GetClients()) +#if !NETSTANDARD1_3 && !WINDOWS_UWP + try { - TryProcessClient(client, now); + Thread.Sleep(timeout); } + catch (ThreadAbortException) + { + // The ThreadAbortException is not actively catched in this project. + // So we use a one which is similar and will be catched properly. + throw new OperationCanceledException(); + } +#else + Task.Delay(timeout).Wait(); +#endif } void TryProcessClient(MqttClient connection, DateTime now) @@ -119,5 +131,14 @@ namespace MQTTnet.Server _logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts.", connection.Id); } } + + void TryProcessClients() + { + var now = DateTime.UtcNow; + foreach (var client in _sessionsManager.GetClients()) + { + TryProcessClient(client, now); + } + } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 85b73354..2c5969e7 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -11,7 +11,6 @@ using System.Threading; using System.Threading.Tasks; using MQTTnet.Adapter; using MQTTnet.Diagnostics; -using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -172,7 +171,7 @@ namespace MQTTnet.Server { ThrowIfNotStarted(); - return _retainedMessagesManager?.ClearMessages() ?? PlatformAbstractionLayer.CompletedTask; + return _retainedMessagesManager?.ClearMessages() ?? CompletedTask.Instance; } public Task DisconnectClientAsync(string id, MqttDisconnectReasonCode reasonCode) diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index e9319395..6cea318b 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -6,7 +6,6 @@ using System; using System.Collections; using System.Threading.Tasks; using MQTTnet.Formatter; -using MQTTnet.Implementations; using MQTTnet.Internal; namespace MQTTnet.Server @@ -34,8 +33,8 @@ namespace MQTTnet.Server var publishPacketFactory = new MqttPublishPacketFactory(); _session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage))); - - return PlatformAbstractionLayer.CompletedTask; + + return CompletedTask.Instance; } public Task DeliverApplicationMessageAsync(MqttApplicationMessage applicationMessage)