Add statistics of last sent packets to the _ConnectedClient_.

Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs

@@ -33,14 +33,14 @@ namespace MQTTnet.Adapter
 
         public Task ConnectAsync(TimeSpan timeout)
         {
-            _logger.Info<MqttChannelAdapter>("Connecting [Timeout={0}]", timeout);
+            _logger.Trace<MqttChannelAdapter>("Connecting [Timeout={0}]", timeout);
 
             return ExecuteAndWrapExceptionAsync(() => _channel.ConnectAsync().TimeoutAfter(timeout));
         }
 
         public Task DisconnectAsync(TimeSpan timeout)
         {
-            _logger.Info<MqttChannelAdapter>("Disconnecting [Timeout={0}]", timeout);
+            _logger.Trace<MqttChannelAdapter>("Disconnecting [Timeout={0}]", timeout);
 
             return ExecuteAndWrapExceptionAsync(() => _channel.DisconnectAsync().TimeoutAfter(timeout));
         }

Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs

@@ -14,8 +14,6 @@ namespace MQTTnet.Client
 {
     public class MqttClient : IMqttClient
     {
-        private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
-
         private readonly IMqttClientAdapterFactory _adapterFactory;
         private readonly MqttPacketDispatcher _packetDispatcher;
         private readonly IMqttNetLogger _logger;
@@ -133,7 +131,7 @@ namespace MQTTnet.Client
                 TopicFilters = topicFilters.ToList()
             };
 
-            await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket);
+            await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket).ConfigureAwait(false);
         }
 
         public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
@@ -141,21 +139,21 @@ namespace MQTTnet.Client
             ThrowIfNotConnected();
 
             var publishPackets = applicationMessages.Select(m => m.ToPublishPacket());
+            var packetGroups = publishPackets.GroupBy(p => p.QualityOfServiceLevel).OrderBy(g => g.Key); 
 
-            foreach (var qosGroup in publishPackets.GroupBy(p => p.QualityOfServiceLevel))
+            foreach (var qosGroup in packetGroups)
             {
-                var qosPackets = qosGroup.ToArray();
                 switch (qosGroup.Key)
                 {
                     case MqttQualityOfServiceLevel.AtMostOnce:
                         {
                             // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
-                            await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosPackets).ConfigureAwait(false);
+                            await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosGroup).ConfigureAwait(false);
                             break;
                         }
                     case MqttQualityOfServiceLevel.AtLeastOnce:
                         {
-                            foreach (var publishPacket in qosPackets)
+                            foreach (var publishPacket in qosGroup)
                             {
                                 publishPacket.PacketIdentifier = GetNewPacketIdentifier();
                                 await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket).ConfigureAwait(false);
@@ -165,7 +163,7 @@ namespace MQTTnet.Client
                         }
                     case MqttQualityOfServiceLevel.ExactlyOnce:
                         {
-                            foreach (var publishPacket in qosPackets)
+                            foreach (var publishPacket in qosGroup)
                             {
                                 publishPacket.PacketIdentifier = GetNewPacketIdentifier();
                                 var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
@@ -249,28 +247,28 @@ namespace MQTTnet.Client
             try
             {
                 _logger.Info<MqttClient>("Received <<< {0}", packet);
+                
+                if (packet is MqttPublishPacket publishPacket)
+                {
+                    await ProcessReceivedPublishPacketAsync(publishPacket).ConfigureAwait(false);
+                    return;
+                }
 
                 if (packet is MqttPingReqPacket)
                 {
-                    await SendAsync(new MqttPingRespPacket());
+                    await SendAsync(new MqttPingRespPacket()).ConfigureAwait(false);
                     return;
                 }
 
                 if (packet is MqttDisconnectPacket)
                 {
-                    await DisconnectAsync();
-                    return;
-                }
-
-                if (packet is MqttPublishPacket publishPacket)
-                {
-                    await ProcessReceivedPublishPacket(publishPacket);
+                    await DisconnectAsync().ConfigureAwait(false);
                     return;
                 }
 
                 if (packet is MqttPubRelPacket pubRelPacket)
                 {
-                    await ProcessReceivedPubRelPacket(pubRelPacket);
+                    await ProcessReceivedPubRelPacket(pubRelPacket).ConfigureAwait(false);
                     return;
                 }
 
@@ -295,32 +293,25 @@ namespace MQTTnet.Client
             }
         }
 
-        private async Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket)
+        private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket)
         {
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
                 FireApplicationMessageReceivedEvent(publishPacket);
-                return;
+                return Task.FromResult(0);
             }
 
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
             {
                 FireApplicationMessageReceivedEvent(publishPacket);
-                await SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
-                return;
+                return SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
             }
 
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
             {
                 // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
-                lock (_unacknowledgedPublishPackets)
-                {
-                    _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
-                }
-
                 FireApplicationMessageReceivedEvent(publishPacket);
-                await SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
-                return;
+                return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
             }
 
             throw new MqttCommunicationException("Received a not supported QoS level.");
@@ -328,11 +319,6 @@ namespace MQTTnet.Client
 
         private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket)
         {
-            lock (_unacknowledgedPublishPackets)
-            {
-                _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
-            }
-
             return SendAsync(pubRelPacket.CreateResponse<MqttPubCompPacket>());
         }
 
@@ -454,7 +440,7 @@ namespace MQTTnet.Client
         {
 #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
             Task.Run(
-                () => ProcessReceivedPacketAsync(packet),
+                async () => await ProcessReceivedPacketAsync(packet).ConfigureAwait(false),
                 cancellationToken).ConfigureAwait(false);
 #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
         }
@@ -465,7 +451,7 @@ namespace MQTTnet.Client
 
 #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
             Task.Run(
-                async () => await ReceivePacketsAsync(cancellationToken),
+                async () => await ReceivePacketsAsync(cancellationToken).ConfigureAwait(false),
                 cancellationToken).ConfigureAwait(false);
 #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
 
@@ -479,7 +465,7 @@ namespace MQTTnet.Client
         {
 #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
             Task.Run(
-                async () => await SendKeepAliveMessagesAsync(cancellationToken), 
+                async () => await SendKeepAliveMessagesAsync(cancellationToken).ConfigureAwait(false), 
                 cancellationToken).ConfigureAwait(false);
 #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
         }

Frameworks/MQTTnet.NetStandard/Diagnostics/MqttNetLogger.cs

@@ -10,7 +10,7 @@ namespace MQTTnet.Diagnostics
         {
             _logId = logId;
         }
-        
+
         public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
 
         public void Trace<TSource>(string message, params object[] parameters)

Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs

@@ -42,7 +42,7 @@ namespace MQTTnet.Implementations
                 _defaultEndpointSocket.Bind(new IPEndPoint(IPAddress.Any, options.GetDefaultEndpointPort()));
                 _defaultEndpointSocket.Listen(options.ConnectionBacklog);
 
-                Task.Run(() => AcceptDefaultEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
+                Task.Run(async () => await AcceptDefaultEndpointConnectionsAsync(_cancellationTokenSource.Token).ConfigureAwait(false), _cancellationTokenSource.Token).ConfigureAwait(false);
             }
 
             if (options.TlsEndpointOptions.IsEnabled)
@@ -62,7 +62,7 @@ namespace MQTTnet.Implementations
                 _tlsEndpointSocket.Bind(new IPEndPoint(IPAddress.Any, options.GetTlsEndpointPort()));
                 _tlsEndpointSocket.Listen(options.ConnectionBacklog);
 
-                Task.Run(() => AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
+                Task.Run(async () => await AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token).ConfigureAwait(false), _cancellationTokenSource.Token).ConfigureAwait(false);
             }
 
             return Task.FromResult(0);
@@ -102,13 +102,29 @@ namespace MQTTnet.Implementations
 #else
                     var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false);
 #endif
+
                     var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer(), _logger);
                     ClientAccepted?.Invoke(this, 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 (SocketException exception)
+                {
+                    if (exception.SocketErrorCode == SocketError.OperationAborted)
+                    {
+                        return;
+                    }
+
+                    _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
+                }
                 catch (Exception exception)
                 {
                     _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
-
+                }
+                finally
+                {
                     //excessive CPU consumed if in endless loop of socket errors
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
                 }
@@ -133,10 +149,25 @@ namespace MQTTnet.Implementations
                     var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _logger);
                     ClientAccepted?.Invoke(this, 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 (SocketException exception)
+                {
+                    if (exception.SocketErrorCode == SocketError.OperationAborted)
+                    {
+                        return;
+                    }
+
+                    _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
+                }
                 catch (Exception exception)
                 {
                     _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at TLS endpoint.");
-
+                }
+                finally
+                {
                     //excessive CPU consumed if in endless loop of socket errors
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
                 }

Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs

@@ -9,8 +9,8 @@ namespace MQTTnet.Server
 
         public MqttProtocolVersion ProtocolVersion { get; set; }
 
-        public TimeSpan LastPacketReceivedDuration { get; set; }
+        public TimeSpan LastPacketReceived { get; set; }
 
-        public TimeSpan LastNonKeepAlivePacketReceivedDuration{ get; set; }
+        public TimeSpan LastNonKeepAlivePacketReceived { get; set; }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
@@ -13,6 +14,9 @@ namespace MQTTnet.Server
 {
     public sealed class MqttClientSession
     {
+        private readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch();
+        private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = new Stopwatch();
+
         private readonly MqttClientSubscriptionsManager _subscriptionsManager;
         private readonly MqttClientSessionsManager _sessionsManager;
         private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
@@ -44,6 +48,10 @@ namespace MQTTnet.Server
 
         public MqttProtocolVersion? ProtocolVersion => _adapter?.PacketSerializer.ProtocolVersion;
 
+        public TimeSpan LastPacketReceived => _lastPacketReceivedTracker.Elapsed;
+
+        public TimeSpan LastNonKeepAlivePacketReceived => _lastNonKeepAlivePacketReceivedTracker.Elapsed;
+
         public bool IsConnected => _adapter != null;
 
         public async Task RunAsync(MqttApplicationMessage willMessage, IMqttChannelAdapter adapter)
@@ -52,6 +60,9 @@ namespace MQTTnet.Server
 
             try
             {
+                _lastPacketReceivedTracker.Restart();
+                _lastNonKeepAlivePacketReceivedTracker.Restart();
+
                 var cancellationTokenSource = new CancellationTokenSource();
 
                 _willMessage = willMessage;
@@ -78,6 +89,11 @@ namespace MQTTnet.Server
         {
             try
             {
+                if (_cancellationTokenSource == null)
+                {
+                    return;
+                }
+
                 _cancellationTokenSource?.Cancel(false);
                 _cancellationTokenSource?.Dispose();
                 _cancellationTokenSource = null;
@@ -143,6 +159,13 @@ namespace MQTTnet.Server
 
         private Task ProcessReceivedPacketAsync(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken)
         {
+            _lastPacketReceivedTracker.Restart();
+
+            if (!(packet is MqttPingReqPacket))
+            {
+                _lastNonKeepAlivePacketReceivedTracker.Restart();
+            }
+
             if (packet is MqttPublishPacket publishPacket)
             {
                 return HandleIncomingPublishPacketAsync(adapter, publishPacket, cancellationToken);
@@ -185,7 +208,7 @@ namespace MQTTnet.Server
             }
 
             _logger.Warning<MqttClientSession>("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
-            return StopAsync();            
+            return StopAsync();
         }
 
         private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
@@ -211,7 +234,7 @@ namespace MQTTnet.Server
             }
         }
 
-        private async Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        private Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
             var applicationMessage = publishPacket.ToApplicationMessage();
 
@@ -219,36 +242,44 @@ namespace MQTTnet.Server
             {
                 case MqttQualityOfServiceLevel.AtMostOnce:
                     {
-                        await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
-                        return;
+                        return _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
                     }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
                     {
-                        await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
-
-                        await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken,
-                            new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false);
-
-                        return;
+                        return HandleIncomingPublishPacketWithQoS1(adapter, applicationMessage, publishPacket, cancellationToken);
                     }
                 case MqttQualityOfServiceLevel.ExactlyOnce:
                     {
-                        // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
-                        await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
-
-                        await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken,
-                            new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false);
-
-                        return;
+                        return HandleIncomingPublishPacketWithQoS2(adapter, applicationMessage, publishPacket, cancellationToken);
                     }
                 default:
-                    throw new MqttCommunicationException("Received a not supported QoS level.");
+                    {
+                        throw new MqttCommunicationException("Received a not supported QoS level.");
+                    }
             }
         }
 
+        private async Task HandleIncomingPublishPacketWithQoS1(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        {
+            await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
+
+            var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier };
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false);
+        }
+
+        private async Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        {
+            // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
+            await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
+
+            var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier };
+            await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false);
+        }
+
         private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
         {
-            return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier });
+            var response = new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier };
+            return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response);
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs

@@ -120,11 +120,12 @@ namespace MQTTnet.Server
             await _sessionsSemaphore.WaitAsync().ConfigureAwait(false);
             try
             {
-                var now = DateTime.UtcNow;
                 return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
                 {
                     ClientId = s.Value.ClientId,
-                    ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311
+                    ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311,
+                    LastPacketReceived = s.Value.LastPacketReceived,
+                    LastNonKeepAlivePacketReceived = s.Value.LastNonKeepAlivePacketReceived
                 }).ToList();
             }
             finally

Tests/MQTTnet.TestApp.NetCore/Program.cs

@@ -40,7 +40,7 @@ namespace MQTTnet.TestApp.NetCore
 
             Thread.Sleep(Timeout.Infinite);
         }
-        
+
         // This code is used at the Wiki on GitHub!
         // ReSharper disable once UnusedMember.Local
         private static async void WikiCode()
@@ -71,6 +71,26 @@ namespace MQTTnet.TestApp.NetCore
                 var factory = new MqttFactory();
                 var client = factory.CreateMqttClient();
             }
+
+            {
+                // Write all trace messages to the console window.
+                MqttNetGlobalLogger.LogMessagePublished += (s, e) =>
+                {
+                    var trace = $">> [{e.TraceMessage.Timestamp:O}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Source}] [{e.TraceMessage.Level}]: {e.TraceMessage.Message}";
+                    if (e.TraceMessage.Exception != null)
+                    {
+                        trace += Environment.NewLine + e.TraceMessage.Exception.ToString();
+                    }
+
+                    Console.WriteLine(trace);
+                };
+            }
+
+            {
+                // Use a custom log ID for the logger.
+                var factory = new MqttFactory();
+                var mqttClient = factory.CreateMqttClient(new MqttNetLogger("MyCustomId"));
+            }
         }
     }
 
@@ -106,4 +126,4 @@ namespace MQTTnet.TestApp.NetCore
             return Task.FromResult(retainedMessages);
         }
     }
-}
+}