Merge pull request #833 from JanEggers/bugfixes

Bugfixes

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -16,7 +16,7 @@ using MQTTnet.Server;
 
 namespace MQTTnet.Extensions.ManagedClient
 {
-    public class ManagedMqttClient : IManagedMqttClient
+    public class ManagedMqttClient : Disposable, IManagedMqttClient
     {
         private readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
 
@@ -42,9 +42,7 @@ namespace MQTTnet.Extensions.ManagedClient
         private Task _maintainConnectionTask;
 
         private ManagedMqttClientStorageManager _storageManager;
-
-        private bool _disposed;
-
+        
         public ManagedMqttClient(IMqttClient mqttClient, IMqttNetChildLogger logger)
         {
             _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
@@ -146,6 +144,7 @@ namespace MQTTnet.Extensions.ManagedClient
             ThrowIfDisposed();
 
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+            if (Options == null) throw new InvalidOperationException("call StartAsync before publishing messages");
 
             MqttTopicValidator.ThrowIfInvalid(applicationMessage.ApplicationMessage.Topic);
 
@@ -238,36 +237,25 @@ namespace MQTTnet.Extensions.ManagedClient
             return Task.FromResult(0);
         }
 
-        public void Dispose()
+        protected override void Dispose(bool disposing)
         {
-            if (_disposed)
+            if (disposing)
             {
-                return;
-            }
-
-            _disposed = true;
-
-            StopPublishing();
-            StopMaintainingConnection();
-
-            if (_maintainConnectionTask != null)
-            {
-                Task.WaitAny(_maintainConnectionTask);
-                _maintainConnectionTask = null;
-            }
-
-            _messageQueue.Dispose();
-            _messageQueueLock.Dispose();
-            _mqttClient.Dispose();
-            _subscriptionsQueuedSignal.Dispose();
-        }
-
-        private void ThrowIfDisposed()
-        {
-            if (_disposed)
-            {
-                throw new ObjectDisposedException(nameof(ManagedMqttClient));
+                StopPublishing();
+                StopMaintainingConnection();
+
+                if (_maintainConnectionTask != null)
+                {
+                    _maintainConnectionTask.GetAwaiter().GetResult();
+                    _maintainConnectionTask = null;
+                }
+
+                _messageQueue.Dispose();
+                _messageQueueLock.Dispose();
+                _mqttClient.Dispose();
+                _subscriptionsQueuedSignal.Dispose();
             }
+            base.Dispose(disposing);
         }
 
         private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
@@ -288,7 +276,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
             finally
             {
-                if (!_disposed)
+                if (!IsDisposed)
                 {
                     try
                     {

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -14,7 +14,7 @@ using MQTTnet.Packets;
 
 namespace MQTTnet.Adapter
 {
-    public class MqttChannelAdapter : IMqttChannelAdapter
+    public class MqttChannelAdapter : Disposable, IMqttChannelAdapter
     {
         private const uint ErrorOperationAborted = 0x800703E3;
         private const int ReadBufferSize = 4096;  // TODO: Move buffer size to config
@@ -26,9 +26,7 @@ namespace MQTTnet.Adapter
         private readonly MqttPacketReader _packetReader;
 
         private readonly byte[] _fixedHeaderBuffer = new byte[2];
-
-        private bool _isDisposed;
-
+        
         private long _bytesReceived;
         private long _bytesSent;
 
@@ -269,19 +267,13 @@ namespace MQTTnet.Adapter
             }
         }
 
-        public void Dispose()
+        protected override void Dispose(bool disposing)
         {
-            _isDisposed = true;
-
-            _channel?.Dispose();
-        }
-
-        private void ThrowIfDisposed()
-        {
-            if (_isDisposed)
+            if (disposing)
             {
-                throw new ObjectDisposedException(nameof(MqttChannelAdapter));
+                _channel?.Dispose();
             }
+            base.Dispose(disposing);
         }
 
         private static bool IsWrappedException(Exception exception)

Source/MQTTnet/Client/MqttClient.cs

@@ -20,7 +20,7 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Client
 {
-    public class MqttClient : IMqttClient
+    public class MqttClient : Disposable, IMqttClient
     {
         private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
         private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
@@ -63,6 +63,8 @@ namespace MQTTnet.Client
 
             ThrowIfConnected("It is not allowed to connect with a server after the connection is established.");
 
+            ThrowIfDisposed();
+
             MqttClientAuthenticateResult authenticateResult = null;
 
             try
@@ -79,13 +81,16 @@ namespace MQTTnet.Client
                 var adapter = _adapterFactory.CreateClientAdapter(options, _logger);
                 _adapter = adapter;
 
-                _logger.Verbose($"Trying to connect with server '{options.ChannelOptions}' (Timeout={options.CommunicationTimeout}).");
-                await _adapter.ConnectAsync(options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
-                _logger.Verbose("Connection with server established.");
+                using (var combined = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken))
+                {
+                    _logger.Verbose($"Trying to connect with server '{options.ChannelOptions}' (Timeout={options.CommunicationTimeout}).");
+                    await _adapter.ConnectAsync(options.CommunicationTimeout, combined.Token).ConfigureAwait(false);
+                    _logger.Verbose("Connection with server established.");
 
-                _packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken);
+                    _packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken);
 
-                authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, cancellationToken).ConfigureAwait(false);
+                    authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, combined.Token).ConfigureAwait(false);
+                }
 
                 _sendTracker.Restart();
 
@@ -161,6 +166,7 @@ namespace MQTTnet.Client
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
+            ThrowIfDisposed();
             ThrowIfNotConnected();
 
             var subscribePacket = _adapter.PacketFormatterAdapter.DataConverter.CreateSubscribePacket(options);
@@ -174,6 +180,7 @@ namespace MQTTnet.Client
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
+            ThrowIfDisposed();
             ThrowIfNotConnected();
 
             var unsubscribePacket = _adapter.PacketFormatterAdapter.DataConverter.CreateUnsubscribePacket(options);
@@ -189,6 +196,7 @@ namespace MQTTnet.Client
 
             MqttTopicValidator.ThrowIfInvalid(applicationMessage.Topic);
 
+            ThrowIfDisposed();
             ThrowIfNotConnected();
 
             var publishPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(applicationMessage);
@@ -214,7 +222,7 @@ namespace MQTTnet.Client
             }
         }
 
-        public void Dispose()
+        private void Cleanup()
         {
             _backgroundCancellationTokenSource?.Cancel(false);
             _backgroundCancellationTokenSource?.Dispose();
@@ -224,6 +232,18 @@ namespace MQTTnet.Client
             _adapter = null;
         }
 
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                Cleanup();
+
+                DisconnectedHandler = null;
+            }
+            base.Dispose(disposing);
+        }
+
         private async Task<MqttClientAuthenticateResult> AuthenticateAsync(IMqttChannelAdapter channelAdapter, MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken)
         {
             var connectPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnectPacket(
@@ -288,7 +308,7 @@ namespace MQTTnet.Client
             }
             finally
             {
-                Dispose();
+                Cleanup();
                 _cleanDisconnectInitiated = false;
 
                 _logger.Info("Disconnected.");

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -10,10 +10,11 @@ using System.Runtime.ExceptionServices;
 using System.Threading;
 using MQTTnet.Channel;
 using MQTTnet.Client.Options;
+using MQTTnet.Internal;
 
 namespace MQTTnet.Implementations
 {
-    public class MqttTcpChannel : IMqttChannel
+    public class MqttTcpChannel : Disposable, IMqttChannel
     {
         private readonly IMqttClientOptions _clientOptions;
         private readonly MqttClientTcpOptions _options;
@@ -72,11 +73,7 @@ namespace MQTTnet.Implementations
             // Workaround for: workaround for https://github.com/dotnet/corefx/issues/24430
             using (cancellationToken.Register(() => socket.Dispose()))
             {
-#if NET452 || NET461
-                await Task.Factory.FromAsync(socket.BeginConnect, socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false);
-#else
-                await socket.ConnectAsync(_options.Server, _options.GetPort()).ConfigureAwait(false);
-#endif
+                await PlatformAbstractionLayer.ConnectAsync(socket, _options.Server, _options.GetPort()).ConfigureAwait(false);
             }
 
             var networkStream = new NetworkStream(socket, true);
@@ -98,7 +95,7 @@ namespace MQTTnet.Implementations
 
         public Task DisconnectAsync(CancellationToken cancellationToken)
         {
-            Dispose();
+            Cleanup();
             return Task.FromResult(0);
         }
 
@@ -117,6 +114,10 @@ namespace MQTTnet.Implementations
                     return await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
                 }
             }
+            catch (ObjectDisposedException)
+            {
+                return 0;
+            }
             catch (IOException exception)
             {
                 if (exception.InnerException is SocketException socketException)
@@ -143,6 +144,10 @@ namespace MQTTnet.Implementations
                     await _stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
                 }
             }
+            catch (ObjectDisposedException)
+            {
+                return;
+            }
             catch (IOException exception)
             {
                 if (exception.InnerException is SocketException socketException)
@@ -154,7 +159,7 @@ namespace MQTTnet.Implementations
             }
         }
 
-        public void Dispose()
+        private void Cleanup()
         {
             // When the stream is disposed it will also close the socket and this will also dispose it.
             // So there is no need to dispose the socket again.
@@ -173,6 +178,15 @@ namespace MQTTnet.Implementations
             _stream = null;
         }
 
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                Cleanup();
+            }
+            base.Dispose(disposing);
+        }
+
         private bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
         {
             if (_options.TlsOptions.CertificateValidationCallback != null)

Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs

@@ -8,11 +8,12 @@ using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
+using MQTTnet.Internal;
 using MQTTnet.Server;
 
 namespace MQTTnet.Implementations
 {
-    public class MqttTcpServerAdapter : IMqttServerAdapter
+    public class MqttTcpServerAdapter : Disposable, IMqttServerAdapter
     {
         private readonly List<MqttTcpServerListener> _listeners = new List<MqttTcpServerListener>();
         private readonly IMqttNetChildLogger _logger;
@@ -72,11 +73,11 @@ namespace MQTTnet.Implementations
 
         public Task StopAsync()
         {
-            Dispose();
+            Cleanup();
             return Task.FromResult(0);
         }
 
-        public void Dispose()
+        private void Cleanup()
         {
             _cancellationTokenSource?.Cancel(false);
             _cancellationTokenSource?.Dispose();
@@ -90,6 +91,15 @@ namespace MQTTnet.Implementations
             _listeners.Clear();
         }
 
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                Cleanup();
+            }
+            base.Dispose(disposing);
+        }
+
         private void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate, CancellationToken cancellationToken)
         {
             if (!options.BoundInterNetworkAddress.Equals(IPAddress.None))

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -107,12 +107,7 @@ namespace MQTTnet.Implementations
             {
                 try
                 {
-#if NET452 || NET461
-                    var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false);
-#else
-                    var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
-#endif
-
+                    var clientSocket = await PlatformAbstractionLayer.AcceptAsync(_socket).ConfigureAwait(false);
                     if (clientSocket == null)
                     {
                         continue;

Source/MQTTnet/Implementations/MqttWebSocketChannel.cs

@@ -6,10 +6,11 @@ using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Channel;
 using MQTTnet.Client.Options;
+using MQTTnet.Internal;
 
 namespace MQTTnet.Implementations
 {
-    public class MqttWebSocketChannel : IMqttChannel
+    public class MqttWebSocketChannel : Disposable, IMqttChannel
     {
         private readonly MqttClientWebSocketOptions _options;
 
@@ -111,7 +112,7 @@ namespace MQTTnet.Implementations
                 await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false);
             }
 
-            Dispose();
+            Cleanup();
         }
 
         public async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
@@ -141,7 +142,16 @@ namespace MQTTnet.Implementations
             }
         }
 
-        public void Dispose()
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                Cleanup();
+            }
+            base.Dispose(disposing);
+        }
+
+        private void Cleanup()
         {
             _sendLock?.Dispose();
             _sendLock = null;

Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs

@@ -0,0 +1,92 @@
+using System;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Implementations
+{
+    public static class PlatformAbstractionLayer
+    {
+        public static async Task<Socket> AcceptAsync(Socket socket)
+        {
+#if NET452 || NET461
+            try
+            {
+                return await Task.Factory.FromAsync(socket.BeginAccept, socket.EndAccept, null).ConfigureAwait(false);
+            }
+            catch (ObjectDisposedException)
+            {
+                return null;
+            }
+#else
+            return await socket.AcceptAsync().ConfigureAwait(false);
+#endif
+        }
+
+
+        public static Task ConnectAsync(Socket socket, IPAddress ip, int port)
+        {
+#if NET452 || NET461
+            return Task.Factory.FromAsync(socket.BeginConnect, socket.EndConnect, ip, port, null);
+#else
+            return socket.ConnectAsync(ip, port);
+#endif
+        }
+
+        public static Task ConnectAsync(Socket socket, string host, int port)
+        {
+#if NET452 || NET461
+            return Task.Factory.FromAsync(socket.BeginConnect, socket.EndConnect, host, port, null);
+#else
+            return socket.ConnectAsync(host, port);
+#endif
+        }
+
+#if NET452 || NET461
+        public class SocketWrapper 
+        {
+            private readonly Socket _socket;
+            private readonly ArraySegment<byte> _buffer;
+            private readonly SocketFlags _socketFlags;
+
+            public SocketWrapper(Socket socket, ArraySegment<byte> buffer, SocketFlags socketFlags)
+            {
+                _socket = socket;
+                _buffer = buffer;
+                _socketFlags = socketFlags;
+            }
+
+            public static IAsyncResult BeginSend(AsyncCallback callback, object state)
+            {
+                var real = (SocketWrapper)state;
+                return real._socket.BeginSend(real._buffer.Array, real._buffer.Offset, real._buffer.Count, real._socketFlags, callback, state);
+            }
+
+            public static IAsyncResult BeginReceive(AsyncCallback callback, object state)
+            {
+                var real = (SocketWrapper)state;
+                return real._socket.BeginReceive(real._buffer.Array, real._buffer.Offset, real._buffer.Count, real._socketFlags, callback, state);
+            }
+        }
+#endif
+
+        public static Task SendAsync(Socket socket, ArraySegment<byte> buffer, SocketFlags socketFlags)
+        {
+#if NET452 || NET461            
+            return Task.Factory.FromAsync(SocketWrapper.BeginSend, socket.EndSend, new SocketWrapper(socket, buffer, socketFlags));
+#else
+            return socket.SendAsync(buffer, socketFlags);
+#endif
+        }
+
+        public static Task<int> ReceiveAsync(Socket socket, ArraySegment<byte> buffer, SocketFlags socketFlags)
+        {
+#if NET452 || NET461
+            return Task.Factory.FromAsync(SocketWrapper.BeginReceive, socket.EndReceive, new SocketWrapper(socket, buffer, socketFlags));
+#else
+            return socket.ReceiveAsync(buffer, socketFlags);
+#endif
+        }
+
+    }
+}

Source/MQTTnet/Internal/BlockingQueue.cs

@@ -4,7 +4,7 @@ using System.Threading;
 
 namespace MQTTnet.Internal
 {
-    public class BlockingQueue<TItem> : IDisposable
+    public class BlockingQueue<TItem> : Disposable
     {
         private readonly object _syncRoot = new object();
         private readonly LinkedList<TItem> _items = new LinkedList<TItem>();
@@ -109,9 +109,13 @@ namespace MQTTnet.Internal
             }
         }
 
-        public void Dispose()
+        protected override void Dispose(bool disposing)
         {
-            _gate.Dispose();
+            if (disposing)
+            {
+                _gate.Dispose();
+            }
+            base.Dispose(disposing);
         }
     }
 }

Source/MQTTnet/Internal/Disposable.cs

@@ -0,0 +1,56 @@
+using System;
+
+namespace MQTTnet.Internal
+{
+    public class Disposable : IDisposable
+    {
+        protected bool IsDisposed => _isDisposed;
+
+        protected void ThrowIfDisposed()
+        {
+            if (_isDisposed)
+            {
+                throw new ObjectDisposedException(GetType().Name);
+            }
+        }
+
+
+        #region IDisposable Support
+       
+        private bool _isDisposed = false; // To detect redundant calls
+
+        protected virtual void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                // TODO: dispose managed state (managed objects).
+            }
+
+            // TODO: free unmanaged resources (unmanaged objects) and override a finalizer below.
+            // TODO: set large fields to null.
+        }
+
+        // TODO: override a finalizer only if Dispose(bool disposing) above has code to free unmanaged resources.
+        // ~Disposable()
+        // {
+        //   // Do not change this code. Put cleanup code in Dispose(bool disposing) above.
+        //   Dispose(false);
+        // }
+
+        // This code added to correctly implement the disposable pattern.
+        public void Dispose()
+        {
+            if (_isDisposed)
+            {
+                return;
+            }
+            // Do not change this code. Put cleanup code in Dispose(bool disposing) above.
+            Dispose(true);
+            // TODO: uncomment the following line if the finalizer is overridden above.
+            // GC.SuppressFinalize(this);
+
+            _isDisposed = true;
+        }
+        #endregion
+    }
+}

Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs

@@ -2,11 +2,12 @@
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Exceptions;
+using MQTTnet.Internal;
 using MQTTnet.Packets;
 
 namespace MQTTnet.PacketDispatcher
 {
-    public sealed class MqttPacketAwaiter<TPacket> : IMqttPacketAwaiter where TPacket : MqttBasePacket
+    public sealed class MqttPacketAwaiter<TPacket> : Disposable, IMqttPacketAwaiter where TPacket : MqttBasePacket
     {
         private readonly TaskCompletionSource<MqttBasePacket> _taskCompletionSource = new TaskCompletionSource<MqttBasePacket>();
         private readonly ushort? _packetIdentifier;
@@ -52,9 +53,13 @@ namespace MQTTnet.PacketDispatcher
             Task.Run(() => _taskCompletionSource.TrySetCanceled());
         }
 
-        public void Dispose()
+        protected override void Dispose(bool disposing)
         {
-            _owningPacketDispatcher.RemovePacketAwaiter<TPacket>(_packetIdentifier);
+            if (disposing)
+            {
+                _owningPacketDispatcher.RemovePacketAwaiter<TPacket>(_packetIdentifier);
+            }
+            base.Dispose(disposing);
         }
     }
 }

Source/MQTTnet/Server/MqttClientSessionApplicationMessagesQueue.cs

@@ -6,7 +6,7 @@ using System.Threading.Tasks;
 
 namespace MQTTnet.Server
 {
-    public class MqttClientSessionApplicationMessagesQueue : IDisposable
+    public class MqttClientSessionApplicationMessagesQueue : Disposable
     {
         private readonly AsyncQueue<MqttQueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttQueuedApplicationMessage>();
         
@@ -71,9 +71,14 @@ namespace MQTTnet.Server
             }
         }
 
-        public void Dispose()
+        protected override void Dispose(bool disposing)
         {
-            _messageQueue.Dispose();
+            if (disposing)
+            {
+                _messageQueue.Dispose();
+            }
+
+            base.Dispose(disposing);
         }
     }
 }

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -13,7 +13,7 @@ using MQTTnet.Server.Status;
 
 namespace MQTTnet.Server
 {
-    public class MqttClientSessionsManager : IDisposable
+    public class MqttClientSessionsManager : Disposable
     {
         private readonly AsyncQueue<MqttEnqueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttEnqueuedApplicationMessage>();
 
@@ -145,9 +145,13 @@ namespace MQTTnet.Server
             _logger.Verbose("Session for client '{0}' deleted.", clientId);
         }
 
-        public void Dispose()
+        protected override void Dispose(bool disposing)
         {
-            _messageQueue?.Dispose();
+            if (disposing)
+            {
+                _messageQueue?.Dispose();
+            }
+            base.Dispose(disposing);
         }
 
         private async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken)