Make internal message of server async.

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -34,8 +34,10 @@ namespace MQTTnet.Server
         private readonly MqttConnectPacket _connectPacket;
 
         private DateTime _lastPacketReceivedTimestamp;
+        private DateTime _lastNonKeepAlivePacketReceivedTimestamp;
+
         private long _receivedPacketsCount;
-        private long _sentPacketsCount;
+        private long _sentPacketsCount = 1; // Start with 1 because the CONNECT packet is not counted anywhere.
         private long _receivedApplicationMessagesCount;
         private long _sentApplicationMessagesCount;
 
@@ -64,6 +66,7 @@ namespace MQTTnet.Server
             _keepAliveMonitor = new MqttClientKeepAliveMonitor(this, _logger);
 
             _lastPacketReceivedTimestamp = DateTime.UtcNow;
+            _lastNonKeepAlivePacketReceivedTimestamp = _lastPacketReceivedTimestamp;
         }
 
         public string ClientId => _connectPacket.ClientId;
@@ -85,7 +88,7 @@ namespace MQTTnet.Server
         {
             status.ClientId = ClientId;
             status.Endpoint = _endpoint;
-            status.ProtocolVersion = _channelAdapter.PacketFormatterAdapter.ProtocolVersion.Value;
+            status.ProtocolVersion = _channelAdapter.PacketFormatterAdapter.ProtocolVersion;
 
             status.ReceivedApplicationMessagesCount = Interlocked.Read(ref _receivedApplicationMessagesCount);
             status.SentApplicationMessagesCount = Interlocked.Read(ref _sentApplicationMessagesCount);
@@ -94,8 +97,10 @@ namespace MQTTnet.Server
             status.SentPacketsCount = Interlocked.Read(ref _sentPacketsCount);
 
             status.LastPacketReceivedTimestamp = _lastPacketReceivedTimestamp;
+            status.LastNonKeepAlivePacketReceivedTimestamp = _lastNonKeepAlivePacketReceivedTimestamp;
 
-            //status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived;
+            status.BytesSent = _channelAdapter.BytesSent;
+            status.BytesReceived = _channelAdapter.BytesReceived;
         }
         
         //public void ClearPendingApplicationMessages()
@@ -131,10 +136,9 @@ namespace MQTTnet.Server
             try
             {
                 _logger.Info("Client '{0}': Session started.", ClientId);
-                //_eventDispatcher.OnClientConnected(ClientId);
-
-                _channelAdapter.ReadingPacketStarted += OnAdapterReadingPacketStarted;
-                _channelAdapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted;
+                
+                _channelAdapter.ReadingPacketStartedCallback = OnAdapterReadingPacketStarted;
+                _channelAdapter.ReadingPacketCompletedCallback = OnAdapterReadingPacketCompleted;
 
                 Session.WillMessage = _connectPacket.WillMessage;
 
@@ -166,7 +170,12 @@ namespace MQTTnet.Server
 
                     Interlocked.Increment(ref _sentPacketsCount);
                     _lastPacketReceivedTimestamp = DateTime.UtcNow;
-                    
+
+                    if (!(packet is MqttPingReqPacket || packet is MqttPingRespPacket))
+                    {
+                        _lastNonKeepAlivePacketReceivedTimestamp = _lastPacketReceivedTimestamp;
+                    }
+
                     _keepAliveMonitor.PacketReceived();
 
                     if (packet is MqttPublishPacket publishPacket)
@@ -243,12 +252,11 @@ namespace MQTTnet.Server
 
                 _packetDispatcher.Reset();
 
-                _channelAdapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
-                _channelAdapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
+                _channelAdapter.ReadingPacketStartedCallback = null;
+                _channelAdapter.ReadingPacketCompletedCallback = null;
 
                 _logger.Info("Client '{0}': Session stopped.", ClientId);
-                //_eventDispatcher.OnClientDisconnected(ClientId);
-
+                
                 _packageReceiverTask = null;
             }
 
@@ -376,7 +384,7 @@ namespace MQTTnet.Server
 
         private async Task SendPendingPacketsAsync(CancellationToken cancellationToken)
         {
-            MqttPendingApplicationMessage queuedApplicationMessage = null;
+            MqttQueuedApplicationMessage queuedApplicationMessage = null;
             MqttPublishPacket publishPacket = null;
 
             try
@@ -501,12 +509,12 @@ namespace MQTTnet.Server
             }
         }
 
-        private void OnAdapterReadingPacketCompleted(object sender, EventArgs e)
+        private void OnAdapterReadingPacketCompleted()
         {
             _keepAliveMonitor?.Resume();
         }
 
-        private void OnAdapterReadingPacketStarted(object sender, EventArgs e)
+        private void OnAdapterReadingPacketStarted()
         {
             _keepAliveMonitor?.Pause();
         }

Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs

@@ -3,7 +3,6 @@ using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
-using MQTTnet.Packets;
 
 namespace MQTTnet.Server
 {

Source/MQTTnet/Server/MqttClientSessionApplicationMessagesQueue.cs

@@ -1,54 +1,29 @@
 using MQTTnet.Internal;
 using MQTTnet.Protocol;
 using System;
-using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 
 namespace MQTTnet.Server
 {
-    public class MqttPendingApplicationMessage
-    {
-        public MqttApplicationMessage ApplicationMessage { get; set; }
-
-        public string SenderClientId { get; set; }
-
-        public bool IsRetainedMessage { get; set; }
-
-        public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
-
-        public bool IsDuplicate { get; set; }
-    }
-
     public class MqttClientSessionApplicationMessagesQueue : IDisposable
     {
-        private readonly Queue<MqttPendingApplicationMessage> _messageQueue = new Queue<MqttPendingApplicationMessage>();
-        private readonly AsyncAutoResetEvent _messageQueueLock = new AsyncAutoResetEvent();
-
+        private readonly AsyncQueue<MqttQueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttQueuedApplicationMessage>();
+        
         private readonly IMqttServerOptions _options;
 
         public MqttClientSessionApplicationMessagesQueue(IMqttServerOptions options)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
-
         }
 
-        public int Count
-        {
-            get
-            {
-                lock (_messageQueue)
-                {
-                    return _messageQueue.Count;
-                }
-            }
-        }
+        public int Count => _messageQueue.Count;
 
         public void Enqueue(MqttApplicationMessage applicationMessage, string senderClientId, MqttQualityOfServiceLevel qualityOfServiceLevel, bool isRetainedMessage)
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
-            Enqueue(new MqttPendingApplicationMessage
+            Enqueue(new MqttQueuedApplicationMessage
             {
                 ApplicationMessage = applicationMessage,
                 SenderClientId = senderClientId,
@@ -59,39 +34,23 @@ namespace MQTTnet.Server
 
         public void Clear()
         {
-            lock (_messageQueue)
-            {
-                _messageQueue.Clear();
-            }
+            _messageQueue.Clear();
         }
 
-        public void Dispose()
+        public async Task<MqttQueuedApplicationMessage> TakeAsync(CancellationToken cancellationToken)
         {
-        }
-
-        public async Task<MqttPendingApplicationMessage> TakeAsync(CancellationToken cancellationToken)
-        {
-            // TODO: Create a blocking queue from this.
-
-            while (!cancellationToken.IsCancellationRequested)
+            var dequeueResult = await _messageQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
+            if (!dequeueResult.IsSuccess)
             {
-                lock (_messageQueue)
-                {
-                    if (_messageQueue.Count > 0)
-                    {
-                        return _messageQueue.Dequeue();
-                    }
-                }
-
-                await _messageQueueLock.WaitOneAsync(cancellationToken).ConfigureAwait(false);
+                return null;
             }
 
-            return null;
+            return dequeueResult.Item;
         }
 
-        public void Enqueue(MqttPendingApplicationMessage enqueuedApplicationMessage)
+        public void Enqueue(MqttQueuedApplicationMessage queuedApplicationMessage)
         {
-            if (enqueuedApplicationMessage == null) throw new ArgumentNullException(nameof(enqueuedApplicationMessage));
+            if (queuedApplicationMessage == null) throw new ArgumentNullException(nameof(queuedApplicationMessage));
 
             lock (_messageQueue)
             {
@@ -104,14 +63,17 @@ namespace MQTTnet.Server
 
                     if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
                     {
-                        _messageQueue.Dequeue();
+                        _messageQueue.TryDequeue();
                     }
                 }
 
-                _messageQueue.Enqueue(enqueuedApplicationMessage);
+                _messageQueue.Enqueue(queuedApplicationMessage);
             }
+        }
 
-            _messageQueueLock.Set();
+        public void Dispose()
+        {
+            _messageQueue.Dispose();
         }
     }
 }

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -68,7 +68,7 @@ namespace MQTTnet.Server
 
             foreach (var connection in _connections.Values)
             {
-                var clientStatus = new MqttClientStatus(connection, this);
+                var clientStatus = new MqttClientStatus(connection);
                 connection.FillStatus(clientStatus);
                 
                 var sessionStatus = new MqttSessionStatus(connection.Session, this);
@@ -175,7 +175,8 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                var queuedApplicationMessage = await _messageQueue.DequeueAsync(cancellationToken).ConfigureAwait(false);
+                var dequeueResult = await _messageQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
+                var queuedApplicationMessage = dequeueResult.Item;
 
                 var sender = queuedApplicationMessage.Sender;
                 var applicationMessage = queuedApplicationMessage.ApplicationMessage;

Source/MQTTnet/Server/MqttQueuedApplicationMessage.cs

@@ -0,0 +1,17 @@
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server
+{
+    public class MqttQueuedApplicationMessage
+    {
+        public MqttApplicationMessage ApplicationMessage { get; set; }
+
+        public string SenderClientId { get; set; }
+
+        public bool IsRetainedMessage { get; set; }
+
+        public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
+
+        public bool IsDuplicate { get; set; }
+    }
+}

Source/MQTTnet/Server/Status/IMqttClientStatus.cs

@@ -26,6 +26,10 @@ namespace MQTTnet.Server.Status
 
         IMqttSessionStatus Session { get; }
 
+        long BytesSent { get; }
+
+        long BytesReceived { get; }
+
         Task DisconnectAsync();
     }
 }

Source/MQTTnet/Server/Status/MqttClientStatus.cs

@@ -6,13 +6,11 @@ namespace MQTTnet.Server.Status
 {
     public class MqttClientStatus : IMqttClientStatus
     {
-        private readonly MqttClientSessionsManager _sessionsManager;
         private readonly MqttClientConnection _connection;
 
-        public MqttClientStatus(MqttClientConnection connection, MqttClientSessionsManager sessionsManager)
+        public MqttClientStatus(MqttClientConnection connection)
         {
             _connection = connection;
-            _sessionsManager = sessionsManager;
         }
 
         public string ClientId { get; set; }
@@ -35,6 +33,10 @@ namespace MQTTnet.Server.Status
 
         public IMqttSessionStatus Session { get; set; }
 
+        public long BytesSent { get; set; }
+
+        public long BytesReceived { get; set; }
+
         public Task DisconnectAsync()
         {
             return _connection.StopAsync();