Fix disposal of SSL streams.

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -283,15 +283,19 @@ namespace MQTTnet.Adapter
 
         private static bool IsWrappedException(Exception exception)
         {
-            return exception is TaskCanceledException ||
+            return exception is OperationCanceledException ||
-                   exception is OperationCanceledException ||
                    exception is MqttCommunicationTimedOutException ||
                    exception is MqttCommunicationException;
         }
 
         private static void WrapException(Exception exception)
         {
-            if (exception is IOException && exception.InnerException is SocketException socketException)
+            if (exception is IOException && exception.InnerException is SocketException innerException)
+            {
+                exception = innerException;
+            }
+
+            if (exception is SocketException socketException)
             {
                 if (socketException.SocketErrorCode == SocketError.ConnectionAborted ||
                     socketException.SocketErrorCode == SocketError.OperationAborted)

Source/MQTTnet/Adapter/MqttServerAdapterClientAcceptedEventArgs.cs

@@ -5,12 +5,12 @@ namespace MQTTnet.Adapter
 {
     public class MqttServerAdapterClientAcceptedEventArgs : EventArgs
     {
-        public MqttServerAdapterClientAcceptedEventArgs(IMqttChannelAdapter client)
+        public MqttServerAdapterClientAcceptedEventArgs(IMqttChannelAdapter channelAdapter)
         {
-            Client = client ?? throw new ArgumentNullException(nameof(client));
+            ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
         }
 
-        public IMqttChannelAdapter Client { get; }
+        public IMqttChannelAdapter ChannelAdapter { get; }
 
         public Task SessionTask { get; set; }
     }

Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Generic;
 using MQTTnet.Adapter;
 using MQTTnet.Packets;
 

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -6,6 +6,7 @@ using System.Security.Cryptography.X509Certificates;
 using System.Threading.Tasks;
 using System.IO;
 using System.Linq;
+using System.Runtime.ExceptionServices;
 using System.Threading;
 using MQTTnet.Channel;
 using MQTTnet.Client.Options;
@@ -17,12 +18,8 @@ namespace MQTTnet.Implementations
         private readonly IMqttClientOptions _clientOptions;
         private readonly MqttClientTcpOptions _options;
 
-        private Socket _socket;
         private Stream _stream;
 
-        /// <summary>
-        /// called on client sockets are created in connect
-        /// </summary>
         public MqttTcpChannel(IMqttClientOptions clientOptions)
         {
             _clientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions));
@@ -31,48 +28,60 @@ namespace MQTTnet.Implementations
             IsSecureConnection = clientOptions.ChannelOptions?.TlsOptions?.UseTls == true;
         }
 
-        /// <summary>
+        public MqttTcpChannel(Stream stream)
-        /// called on server, sockets are passed in
-        /// connect will not be called
-        /// </summary>
-        public MqttTcpChannel(Socket socket, SslStream sslStream)
         {
-            _socket = socket ?? throw new ArgumentNullException(nameof(socket));
+            _stream = stream ?? throw new ArgumentNullException(nameof(stream));
 
-            IsSecureConnection = sslStream != null;
+            IsSecureConnection = stream is SslStream;
-
-            CreateStream(sslStream);
         }
 
-        public string Endpoint => _socket?.RemoteEndPoint?.ToString();
+        public string Endpoint { get; private set; }
 
         public bool IsSecureConnection { get; }
 
         public async Task ConnectAsync(CancellationToken cancellationToken)
         {
-            if (_socket == null)
+            Socket socket;
+
+            if (_options.AddressFamily == AddressFamily.Unspecified)
             {
-                _socket = new Socket(SocketType.Stream, ProtocolType.Tcp) { NoDelay = true };
+                socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+            }
+            else
+            {
+                socket = new Socket(_options.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
             }
 
+            socket.ReceiveBufferSize = _options.BufferSize;
+            socket.SendBufferSize = _options.BufferSize;
+            socket.DualMode = _options.DualMode;
+            socket.NoDelay = _options.NoDelay;
+
             // Workaround for: workaround for https://github.com/dotnet/corefx/issues/24430
-            using (cancellationToken.Register(() => _socket.Dispose()))
+            using (cancellationToken.Register(() => socket.Dispose()))
             {
 #if NET452 || NET461
-                await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false);
+                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);
+                await socket.ConnectAsync(_options.Server, _options.GetPort()).ConfigureAwait(false);
 #endif
             }
 
-            SslStream sslStream = null;
+            var networkStream = new NetworkStream(socket, true);
+
             if (_options.TlsOptions.UseTls)
             {
-                sslStream = new SslStream(new NetworkStream(_socket, true), false, InternalUserCertificateValidationCallback);
+                var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
                 await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
+
+                _stream = sslStream;
+            }
+            else
+            {
+                _stream = networkStream;
             }
 
-            CreateStream(sslStream);
+            Endpoint = socket.RemoteEndPoint?.ToString();
         }
 
         public Task DisconnectAsync(CancellationToken cancellationToken)
@@ -83,37 +92,58 @@ namespace MQTTnet.Implementations
 
         public async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
-            // Workaround for: https://github.com/dotnet/corefx/issues/24430
+            try
-            using (cancellationToken.Register(Dispose))
             {
-                if (cancellationToken.IsCancellationRequested)
+                // Workaround for: https://github.com/dotnet/corefx/issues/24430
+                using (cancellationToken.Register(Dispose))
                 {
-                    return 0;
+                    if (cancellationToken.IsCancellationRequested)
+                    {
+                        return 0;
+                    }
+
+                    return await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+                }
+            }
+            catch (IOException exception)
+            {
+                if (exception.InnerException is SocketException socketException)
+                {
+                    ExceptionDispatchInfo.Capture(socketException).Throw();
                 }
 
-                return await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+                throw;
             }
         }
 
         public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
-            // Workaround for: https://github.com/dotnet/corefx/issues/24430
+            try
-            using (cancellationToken.Register(Dispose))
             {
-                if (cancellationToken.IsCancellationRequested)
+                // Workaround for: https://github.com/dotnet/corefx/issues/24430
+                using (cancellationToken.Register(Dispose))
                 {
-                    return;
+                    if (cancellationToken.IsCancellationRequested)
+                    {
+                        return;
+                    }
+
+                    await _stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+                }
+            }
+            catch (IOException exception)
+            {
+                if (exception.InnerException is SocketException socketException)
+                {
+                    ExceptionDispatchInfo.Capture(socketException).Throw();
                 }
 
-                await _stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+                throw;
-                //await _stream.FlushAsync(cancellationToken);
             }
         }
 
         public void Dispose()
         {
-            _socket = null;
-
             // 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.
             // https://stackoverflow.com/questions/3601521/should-i-manually-dispose-the-socket-after-closing-it
@@ -127,6 +157,8 @@ namespace MQTTnet.Implementations
             catch (NullReferenceException)
             {
             }
+
+            _stream = null;
         }
 
         private bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
@@ -175,18 +207,6 @@ namespace MQTTnet.Implementations
 
             return certificates;
         }
-
-        private void CreateStream(Stream stream)
-        {
-            if (stream != null)
-            {
-                _stream = stream;
-            }
-            else
-            {
-                _stream = new NetworkStream(_socket, true);
-            }
-        }
     }
 }
 #endif

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -1,5 +1,6 @@
 #if !WINDOWS_UWP
 using System;
+using System.IO;
 using System.Net;
 using System.Net.Security;
 using System.Net.Sockets;
@@ -42,7 +43,7 @@ namespace MQTTnet.Implementations
             }
         }
 
-        public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler;
+        public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
 
         public void Start()
         {
@@ -58,53 +59,7 @@ namespace MQTTnet.Implementations
             _logger.Info($"Starting TCP listener for {_socket.LocalEndPoint} TLS={_tlsCertificate != null}.");
 
             _socket.Listen(_options.ConnectionBacklog);
-            Task.Run(AcceptClientConnectionsAsync, _cancellationToken);
+            Task.Run(() => AcceptClientConnectionsAsync(_cancellationToken), _cancellationToken);
-        }
-
-        private async Task AcceptClientConnectionsAsync()
-        {
-            while (!_cancellationToken.IsCancellationRequested)
-            {
-                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
-                    clientSocket.NoDelay = true;
-
-                    SslStream sslStream = null;
-
-                    if (_tlsCertificate != null)
-                    {
-                        sslStream = new SslStream(new NetworkStream(clientSocket), false);
-                        await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, _tlsOptions.SslProtocol, false).ConfigureAwait(false);
-                    }
-
-                    _logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.",
-                        clientSocket.RemoteEndPoint,
-                        _socket.LocalEndPoint,
-                        _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
-
-                    var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketFormatterAdapter(), _logger);
-                    ClientAcceptedHandler?.Invoke(new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
-                }
-                catch (ObjectDisposedException)
-                {
-                    // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
-                }
-                catch (Exception exception)
-                {
-                    if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
-                    {
-                        return;
-                    }
-
-                    _logger.Error(exception, $"Error while accepting connection at TCP listener {_socket.LocalEndPoint} TLS={_tlsCertificate != null}.");
-                    await Task.Delay(TimeSpan.FromSeconds(1), _cancellationToken).ConfigureAwait(false);
-                }
-            }
         }
 
         public void Dispose()
@@ -115,6 +70,78 @@ namespace MQTTnet.Implementations
             _tlsCertificate?.Dispose();
 #endif
         }
+
+        private async Task AcceptClientConnectionsAsync(CancellationToken cancellationToken)
+        {
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                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
+                    await HandleClientConnection(clientSocket).ConfigureAwait(false);
+                }
+                catch (Exception exception)
+                {
+                    _logger.Error(exception, $"Error while accepting connection at TCP listener {_socket.LocalEndPoint} TLS={_tlsCertificate != null}.");
+                    await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
+                }
+            }
+        }
+
+        private async Task HandleClientConnection(Socket clientSocket)
+        {
+            Stream stream = null;
+
+            try
+            {
+                clientSocket.NoDelay = _options.NoDelay;
+
+                stream = new NetworkStream(clientSocket, true);
+
+                if (_tlsCertificate != null)
+                {
+                    var sslStream = new SslStream(stream, false);
+                    await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, _tlsOptions.SslProtocol, false).ConfigureAwait(false);
+                    stream = sslStream;
+                }
+
+                _logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.",
+                    clientSocket.RemoteEndPoint,
+                    _socket.LocalEndPoint,
+                    _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
+
+                var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(stream), new MqttPacketFormatterAdapter(), _logger);
+                ClientAcceptedHandler?.Invoke(new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
+            }
+            catch (Exception exception)
+            {
+                if (exception is ObjectDisposedException)
+                {
+                    // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
+                    return;
+                }
+
+                if (exception is SocketException socketException && socketException.SocketErrorCode == SocketError.OperationAborted)
+                {
+                    return;
+                }
+
+                try
+                {
+                    // Dispose already allocated resources.
+                    stream?.Dispose();
+                    clientSocket?.Dispose();
+                }
+                catch (Exception disposeException)
+                {
+                    throw new AggregateException(exception, disposeException);
+                }
+            }
+        }
     }
 }
 #endif

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -127,7 +127,7 @@ namespace MQTTnet.Server
 
             Session.EnqueueApplicationMessage(applicationMessage, senderClientId, isRetainedApplicationMessage);
 
-            _logger.Verbose("Enqueued application message (ClientId: {0}).", ClientId);
+            _logger.Verbose("Queued application message (ClientId: {0}).", ClientId);
         }
 
         private async Task<MqttClientDisconnectType> RunInternalAsync()
@@ -362,24 +362,6 @@ namespace MQTTnet.Server
             };
 
             return SendAsync(pubRecPacket);
-
-            ////Task.Run(async () =>
-            ////{
-            ////    using (var pubRelPacketAwaiter = _packetDispatcher.AddPacketAwaiter<MqttPubRelPacket>(publishPacket.PacketIdentifier))
-            ////    {
-            ////        await pubRelPacketAwaiter.WaitOneAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
-
-            ////        var pubCompPacket = new MqttPubCompPacket
-            ////        {
-            ////            PacketIdentifier = publishPacket.PacketIdentifier,
-            ////            ReasonCode = MqttPubCompReasonCode.Success
-            ////        };
-
-            ////        await SendAsync(pubCompPacket).ConfigureAwait(false);
-            ////    }
-            ////});
-
-            //await Task.FromResult(0);
         }
 
         private async Task SendPendingPacketsAsync(CancellationToken cancellationToken)

Source/MQTTnet/Server/MqttServer.cs

@@ -186,7 +186,7 @@ namespace MQTTnet.Server
 
         private void OnClientAccepted(MqttServerAdapterClientAcceptedEventArgs eventArgs)
         {
-            eventArgs.SessionTask = _clientSessionsManager.HandleConnectionAsync(eventArgs.Client);
+            eventArgs.SessionTask = _clientSessionsManager.HandleConnectionAsync(eventArgs.ChannelAdapter);
         }
     }
 }

Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs

@@ -21,7 +21,7 @@ namespace MQTTnet.Benchmarks
         {
             var factory = new MqttFactory();
             var tcpServer = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger());
-            tcpServer.ClientAcceptedHandler += args => _serverChannel = (IMqttChannel)args.Client.GetType().GetField("_channel", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance).GetValue(args.Client);
+            tcpServer.ClientAcceptedHandler += args => _serverChannel = (IMqttChannel)args.ChannelAdapter.GetType().GetField("_channel", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance).GetValue(args.ChannelAdapter);
 
             _mqttServer = factory.CreateMqttServer(new[] { tcpServer }, new MqttNetLogger());