Merge pull request #1 from chkr1011/master

Merging with mainline.

Build/MQTTnet.nuspec

@@ -12,10 +12,7 @@
     <requireLicenseAcceptance>true</requireLicenseAcceptance>
     <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</description>
     <releaseNotes>
-* [Client] Fixed wrong value for "ClientWasConnected" in "MqttClientDisconnectedEventArgs" #976 (thanks to @dbeinder).
+* [Server] ..
-* [Client] Added direct support for Amazon AWS connections (requires .NET Core 3.1) (thanks to @henning-krause).
-* [ManagedClient] Exposed the internal MQTT client.
-* [Server] Added client message queue interceptor for QoS level (tahnks to @msallin).
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2020</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor</tags> 

Source/MQTTnet/Client/MqttClient.cs

@@ -245,16 +245,22 @@ namespace MQTTnet.Client
 
         void Cleanup()
         {
-            _backgroundCancellationTokenSource?.Cancel(false);
+            try
-            _backgroundCancellationTokenSource?.Dispose();
+            {
-            _backgroundCancellationTokenSource = null;
+                _backgroundCancellationTokenSource?.Cancel(false);
+            }
+            finally
+            {
+                _backgroundCancellationTokenSource?.Dispose();
+                _backgroundCancellationTokenSource = null;
 
-            _publishPacketReceiverQueue?.Dispose();
+                _publishPacketReceiverQueue?.Dispose();
-            _publishPacketReceiverQueue = null;
+                _publishPacketReceiverQueue = null;
 
-            _adapter?.Dispose();
+                _adapter?.Dispose();
+            }
         }
-        
+
         protected override void Dispose(bool disposing)
         {
             if (disposing)
@@ -387,7 +393,7 @@ namespace MQTTnet.Client
         async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket
         {
             cancellationToken.ThrowIfCancellationRequested();
-            
+
             ushort identifier = 0;
             if (requestPacket is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier.HasValue)
             {
@@ -436,7 +442,7 @@ namespace MQTTnet.Client
                 {
                     // Values described here: [MQTT-3.1.2-24].
                     var waitTime = keepAlivePeriod - _sendTracker.Elapsed;
-                    
+
                     if (waitTime <= TimeSpan.Zero)
                     {
                         await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false);
@@ -561,12 +567,15 @@ namespace MQTTnet.Client
                 {
                     await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false);
                 }
-                else if (packet is MqttDisconnectPacket)
+                else if (packet is MqttDisconnectPacket disconnectPacket)
                 {
                     // Also dispatch disconnect to waiting threads to generate a proper exception.
                     _packetDispatcher.Dispatch(packet);
 
-                    await DisconnectAsync(null, cancellationToken).ConfigureAwait(false);
+                    await DisconnectAsync(new MqttClientDisconnectOptions()
+                    {
+                        ReasonCode = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.UnspecifiedError)
+                    }, cancellationToken).ConfigureAwait(false);
                 }
                 else if (packet is MqttAuthPacket authPacket)
                 {
@@ -628,7 +637,6 @@ namespace MQTTnet.Client
                 try
                 {
                     var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken);
-
                     if (!publishPacketDequeueResult.IsSuccess)
                     {
                         return;

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -87,7 +87,7 @@ namespace MQTTnet.Implementations
                             ApplicationProtocols = _options.TlsOptions.ApplicationProtocols,
                             ClientCertificates = LoadCertificates(),
                             EnabledSslProtocols = _options.TlsOptions.SslProtocol,
-                            CertificateRevocationCheckMode = _options.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.Online : X509RevocationMode.NoCheck,
+                            CertificateRevocationCheckMode = _options.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : X509RevocationMode.Online,
                             TargetHost = _options.Server
                         };
 

Source/MQTTnet/Internal/AsyncQueue.cs

@@ -7,16 +7,15 @@ namespace MQTTnet.Internal
 {
     public sealed class AsyncQueue<TItem> : IDisposable
     {
-        private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
+        SemaphoreSlim _semaphore = new SemaphoreSlim(0);
-
+        ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
-        private ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
+        
-
         public int Count => _queue.Count;
 
         public void Enqueue(TItem item)
         {
             _queue.Enqueue(item);
-            _semaphore.Release();
+            _semaphore?.Release();
         }
 
         public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
@@ -25,12 +24,17 @@ namespace MQTTnet.Internal
             {
                 try
                 {
+                    if (_semaphore == null)
+                    {
+                        return new AsyncQueueDequeueResult<TItem>(false, default);
+                    }
+
                     await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
                     cancellationToken.ThrowIfCancellationRequested();
                 }
                 catch (OperationCanceledException)
                 {
-                    return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
+                    return new AsyncQueueDequeueResult<TItem>(false, default);
                 }
 
                 if (_queue.TryDequeue(out var item))
@@ -39,7 +43,7 @@ namespace MQTTnet.Internal
                 }
             }
 
-            return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
+            return new AsyncQueueDequeueResult<TItem>(false, default);
         }
 
         public AsyncQueueDequeueResult<TItem> TryDequeue()
@@ -49,7 +53,7 @@ namespace MQTTnet.Internal
                 return new AsyncQueueDequeueResult<TItem>(true, item);
             }
 
-            return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
+            return new AsyncQueueDequeueResult<TItem>(false, default);
         }
 
         public void Clear()
@@ -60,6 +64,7 @@ namespace MQTTnet.Internal
         public void Dispose()
         {
             _semaphore?.Dispose();
+            _semaphore = null;
         }
     }
 }

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -23,8 +23,6 @@ namespace MQTTnet.Server
         readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource();
 
         readonly IMqttRetainedMessagesManager _retainedMessagesManager;
-        readonly Func<Task> _onStart;
-        readonly Func<MqttClientDisconnectType, Task> _onStop;
         readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
         readonly MqttClientSessionsManager _sessionsManager;
 
@@ -32,6 +30,7 @@ namespace MQTTnet.Server
         readonly IMqttServerOptions _serverOptions;
 
         readonly IMqttChannelAdapter _channelAdapter;
+        readonly MqttConnectionValidatorContext _connectionValidatorContext;
         readonly IMqttDataConverter _dataConverter;
         readonly string _endpoint;
         readonly DateTime _connectedTimestamp;
@@ -47,25 +46,22 @@ namespace MQTTnet.Server
 
         volatile bool _isTakeover;
 
-        public MqttClientConnection(
+        public MqttClientConnection(MqttConnectPacket connectPacket,
-            MqttConnectPacket connectPacket,
             IMqttChannelAdapter channelAdapter,
             MqttClientSession session,
+            MqttConnectionValidatorContext connectionValidatorContext,
             IMqttServerOptions serverOptions,
             MqttClientSessionsManager sessionsManager,
             IMqttRetainedMessagesManager retainedMessagesManager,
-            Func<Task> onStart,
-            Func<MqttClientDisconnectType, Task> onStop,
             IMqttNetLogger logger)
         {
             Session = session ?? throw new ArgumentNullException(nameof(session));
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
-            _onStart = onStart ?? throw new ArgumentNullException(nameof(onStart));
-            _onStop = onStop ?? throw new ArgumentNullException(nameof(onStop));
 
             _channelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
+            _connectionValidatorContext = connectionValidatorContext ?? throw new ArgumentNullException(nameof(connectionValidatorContext));
             _dataConverter = _channelAdapter.PacketFormatterAdapter.DataConverter;
             _endpoint = _channelAdapter.Endpoint;
             ConnectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
@@ -131,18 +127,17 @@ namespace MQTTnet.Server
             _cancellationToken.Dispose();
         }
 
-        public Task RunAsync(MqttConnectionValidatorContext connectionValidatorContext)
+        public Task RunAsync()
         {
-            _packageReceiverTask = RunInternalAsync(connectionValidatorContext);
+            _packageReceiverTask = RunInternalAsync();
             return _packageReceiverTask;
         }
 
-        async Task RunInternalAsync(MqttConnectionValidatorContext connectionValidatorContext)
+        async Task RunInternalAsync()
         {
             var disconnectType = MqttClientDisconnectType.NotClean;
             try
             {
-                await _onStart();
                 _logger.Info("Client '{0}': Session started.", ClientId);
 
                 _channelAdapter.ReadingPacketStartedCallback = OnAdapterReadingPacketStarted;
@@ -155,9 +150,7 @@ namespace MQTTnet.Server
                 // TODO: Change to single thread in SessionManager. Or use SessionManager and stats from KeepAliveMonitor.
                 _keepAliveMonitor.Start(ConnectPacket.KeepAlivePeriod, _cancellationToken.Token);
 
-                await SendAsync(
+                await SendAsync(_channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(_connectionValidatorContext)).ConfigureAwait(false);
-                    _channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext)
-                    ).ConfigureAwait(false);
 
                 Session.IsCleanSession = false;
 
@@ -262,17 +255,33 @@ namespace MQTTnet.Server
                 _channelAdapter.ReadingPacketStartedCallback = null;
                 _channelAdapter.ReadingPacketCompletedCallback = null;
 
-                _logger.Info("Client '{0}': Connection stopped.", ClientId);
-
                 _packageReceiverTask = null;
 
+                if (_isTakeover)
+                {
+                    try
+                    {
+                        // Don't use SendAsync here _cancellationToken is already cancelled.
+                        await _channelAdapter.SendPacketAsync(new MqttDisconnectPacket
+                        {
+                            ReasonCode = MqttDisconnectReasonCode.SessionTakenOver
+                        }, TimeSpan.Zero, CancellationToken.None).ConfigureAwait(false);
+                    }
+                    catch (Exception exception)
+                    {
+                        _logger.Error(exception, "Client '{0}': Error while sending DISCONNECT packet after takeover.", ClientId);
+                    }
+                }
+
+                _logger.Info("Client '{0}': Connection stopped.", ClientId);
+                
                 try
                 {
-                    await _onStop(disconnectType);
+                    await _sessionsManager.CleanUpClient(ClientId, _channelAdapter, disconnectType);
                 }
                 catch (Exception e)
                 {
-                    _logger.Error(e, "client '{0}': Error while cleaning up", ClientId);
+                    _logger.Error(e, "Client '{0}': Error while cleaning up", ClientId);
                 }
             }
         }

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -21,6 +21,7 @@ namespace MQTTnet.Server
         readonly AsyncLock _createConnectionGate = new AsyncLock();
         readonly ConcurrentDictionary<string, MqttClientConnection> _connections = new ConcurrentDictionary<string, MqttClientConnection>();
         readonly ConcurrentDictionary<string, MqttClientSession> _sessions = new ConcurrentDictionary<string, MqttClientSession>();
+
         readonly IDictionary<object, object> _serverSessionItems = new ConcurrentDictionary<object, object>();
 
         readonly CancellationToken _cancellationToken;
@@ -183,7 +184,6 @@ namespace MQTTnet.Server
                 }
 
                 var dequeueResult = await _messageQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
-
                 if (!dequeueResult.IsSuccess)
                 {
                     return;
@@ -258,8 +258,6 @@ namespace MQTTnet.Server
 
         async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
         {
-            string clientId = null;
-
             try
             {
                 MqttConnectPacket connectPacket;
@@ -291,17 +289,9 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                clientId = connectPacket.ClientId;
+                var connection = await CreateClientConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false);
-
+                await _eventDispatcher.SafeNotifyClientConnectedAsync(connectPacket.ClientId).ConfigureAwait(false);
-                var connection = await CreateClientConnectionAsync(
+                await connection.RunAsync().ConfigureAwait(false);
-                    connectPacket,
-                    connectionValidatorContext,
-                    channelAdapter,
-                    async () => await _eventDispatcher.SafeNotifyClientConnectedAsync(clientId).ConfigureAwait(false),
-                    async disconnectType => await CleanUpClient(clientId, channelAdapter, disconnectType)
-                ).ConfigureAwait(false);
-
-                await connection.RunAsync(connectionValidatorContext).ConfigureAwait(false);
             }
             catch (OperationCanceledException)
             {
@@ -312,15 +302,19 @@ namespace MQTTnet.Server
             }
         }
 
-        private async Task CleanUpClient(string clientId, IMqttChannelAdapter channelAdapter, MqttClientDisconnectType disconnectType)
+        public async Task CleanUpClient(string clientId, IMqttChannelAdapter channelAdapter, MqttClientDisconnectType disconnectType)
         {
             if (clientId != null)
             {
-                _connections.TryRemove(clientId, out _);
+                // in case it is a takeover _connections already contains the new connection
-
+                if (disconnectType != MqttClientDisconnectType.Takeover)
-                if (!_options.EnablePersistentSessions)
                 {
-                    await DeleteSessionAsync(clientId).ConfigureAwait(false);
+                    _connections.TryRemove(clientId, out _);
+
+                    if (!_options.EnablePersistentSessions)
+                    {
+                        await DeleteSessionAsync(clientId).ConfigureAwait(false);
+                    }
                 }
             }
 
@@ -360,45 +354,52 @@ namespace MQTTnet.Server
             return context;
         }
 
-        async Task<MqttClientConnection> CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter, Func<Task> onStart, Func<MqttClientDisconnectType, Task> onStop)
+        async Task<MqttClientConnection> CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
         {
+            MqttClientConnection connection;
+            MqttClientConnection existingConnection = null;
+
             using (await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false))
             {
-                var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var session);
+                var session = _sessions.AddOrUpdate(
-
+                    connectPacket.ClientId,
-                var isConnectionPresent = _connections.TryGetValue(connectPacket.ClientId, out var existingConnection);
+                    key =>
-                if (isConnectionPresent)
-                {
-                    await existingConnection.StopAsync(true).ConfigureAwait(false);
-                }
-
-                if (isSessionPresent)
-                {
-                    if (connectPacket.CleanSession)
                     {
-                        session = null;
+                        _logger.Verbose("Created a new session for client '{0}'.", key);
-
+                        return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
-                        _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
+                    },
-                    }
+                    (key, existingSession) =>
-                    else
                     {
+                        if (connectPacket.CleanSession)
+                        {
+                            _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
+                            return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
+                        }
+
                         _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
-                    }
+                        return existingSession;
-                }
+                    });
 
-                if (session == null)
+                connection = new MqttClientConnection(connectPacket, channelAdapter, session, connectionValidatorContext, _options, this, _retainedMessagesManager, _rootLogger);
-                {
-                    session = new MqttClientSession(connectPacket.ClientId, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
-                    _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
-                }
 
-                var connection = new MqttClientConnection(connectPacket, channelAdapter, session, _options, this, _retainedMessagesManager, onStart, onStop, _rootLogger);
+                _connections.AddOrUpdate(
-
+                    connectPacket.ClientId,
-                _connections[connection.ClientId] = connection;
+                    key => connection,
-                _sessions[session.ClientId] = session;
+                    (key, tempExistingConnection) =>
-
+                    {
-                return connection;
+                        existingConnection = tempExistingConnection;
+                        return connection;
+                    });
             }
+
+            // Disconnect the client outside of the lock so that new clients can still connect while a single
+            // one is being disconnected.
+            if (existingConnection != null)
+            {
+                await existingConnection.StopAsync(true).ConfigureAwait(false);
+            }
+
+            return connection;
         }
 
         async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)

Tests/MQTTnet.Core.Tests/Session_Tests.cs

@@ -1,5 +1,6 @@
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Client;
+using MQTTnet.Client.Options;
 using MQTTnet.Client.Subscribing;
 using MQTTnet.Server;
 using MQTTnet.Tests.Mockups;
@@ -84,5 +85,40 @@ namespace MQTTnet.Tests
                 Assert.AreEqual(true, session.Items["can_subscribe_x"]);
             }
         }
+
+
+        [TestMethod]
+        public async Task Manage_Session_MaxParallel()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+                var serverOptions = new MqttServerOptionsBuilder();
+                await testEnvironment.StartServerAsync(serverOptions);
+
+                var options = new MqttClientOptionsBuilder()
+                    .WithClientId("1")
+                    ;
+
+                var clients = await Task.WhenAll(Enumerable.Range(0, 10)
+                    .Select(i => TryConnect(testEnvironment, options)));
+
+                var connectedClients = clients.Where(c => c?.IsConnected ?? false).ToList();
+
+                Assert.AreEqual(1, connectedClients.Count);
+            }
+        }
+
+        private async Task<IMqttClient> TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options)
+        {
+            try
+            {
+                return await testEnvironment.ConnectClientAsync(options);
+            }
+            catch (System.Exception)
+            {
+                return null;
+            }
+        }
     }
 }