1889 no service for type mqttnetservermqttserver has been registered (#1892)

* Add missing service to service collection

* Update ReleaseNotes.md

* Restore service registrations

.github/workflows/ReleaseNotes.md

@@ -1,7 +1,3 @@
-* [Client] Added support for custom CA chain validation (#1851, thanks to @rido-min).
-* [Client] Added support for custom CA chain validation (#1851, thanks to @rido-min).
-* [Client] Fixed handling of unobserved tasks exceptions (#1871).
-* [Client] Fixed not specified ReasonCode when using _SendExtendedAuthenticationExchangeDataAsync_ (#1882, thanks to @rido-min).
-* [Server] Fixed not working _UpdateRetainedMessageAsync_ public api (#1858, thanks to @kimdiego2098).
-* [Server] Added support for custom DISCONNECT packets when stopping the server or disconnect a client (BREAKING CHANGE!, #1846).
-* [Server] Added new property to stop the server from accepting new connections even if it is running (#1846).
+* [Server] Added new events for delivered and dropped messages (#1866, thanks to @kallayj).
+* [Server] The server will no longer treat a client which is receiving a large payload as alive. The packet must be received completely within the keep alive boundaries (BREAKING CHANGE!, #1883).
+* [Server] Fixed "service not registered" exception in ASP.NET integration (#1889).

.github/workflows/ci.yml

@@ -3,7 +3,7 @@ name: CI
 on: [push, pull_request]
 
 env:
-  VERSION: "4.3.2.${{github.run_number}}"
+  VERSION: "4.3.3.${{github.run_number}}"
 
 jobs:
   build:

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -83,8 +83,6 @@ namespace MQTTnet.AspNetCore
             }
         }
 
-        public bool IsReadingPacket { get; private set; }
-
         public bool IsSecureConnection
         {
             get
@@ -164,11 +162,6 @@ namespace MQTTnet.AspNetCore
                                 BytesReceived += received;
                                 return packet;
                             }
-                            else
-                            {
-                                // we did receive something but the message is not yet complete
-                                IsReadingPacket = true;
-                            }
                         }
                         else if (readResult.IsCompleted)
                         {
@@ -189,12 +182,9 @@ namespace MQTTnet.AspNetCore
                 // completing the channel makes sure that there is no more data read after a protocol error
                 _input?.Complete(exception);
                 _output?.Complete(exception);
+
                 throw;
             }
-            finally
-            {
-                IsReadingPacket = false;
-            }
 
             cancellationToken.ThrowIfCancellationRequested();
             return null;

Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs

@@ -5,6 +5,7 @@
 using System;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Hosting;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
 using MQTTnet.Implementations;
@@ -58,7 +59,8 @@ namespace MQTTnet.AspNetCore
             services.TryAddSingleton(new MqttFactory());
 
             services.AddSingleton<MqttHostedServer>();
-            services.AddHostedService<MqttHostedServer>();
+            services.AddSingleton<IHostedService>(s => s.GetService<MqttHostedServer>());
+            services.AddSingleton<MqttServer>(s => s.GetService<MqttHostedServer>());
         }
 
         public static IServiceCollection AddHostedMqttServerWithServices(this IServiceCollection services, Action<AspNetMqttServerOptionsBuilder> configure)

Source/MQTTnet/Adapter/IMqttChannelAdapter.cs

@@ -25,8 +25,6 @@ namespace MQTTnet.Adapter
 
         long BytesReceived { get; }
 
-        bool IsReadingPacket { get; }
-
         Task ConnectAsync(CancellationToken cancellationToken);
 
         Task DisconnectAsync(CancellationToken cancellationToken);

Source/MQTTnet/Adapter/IMqttServerAdapter.cs

@@ -14,6 +14,7 @@ namespace MQTTnet.Adapter
         Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
 
         Task StartAsync(MqttServerOptions options, IMqttNetLogger logger);
+
         Task StopAsync();
     }
 }

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -55,8 +55,6 @@ namespace MQTTnet.Adapter
 
         public string Endpoint => _channel.Endpoint;
 
-        public bool IsReadingPacket { get; private set; }
-
         public bool IsSecureConnection => _channel.IsSecureConnection;
 
         public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
@@ -389,53 +387,45 @@ namespace MQTTnet.Adapter
                 return ReceivedMqttPacket.Empty;
             }
 
-            IsReadingPacket = true;
-            try
+            var fixedHeader = readFixedHeaderResult.FixedHeader;
+            if (fixedHeader.RemainingLength == 0)
             {
-                var fixedHeader = readFixedHeaderResult.FixedHeader;
-                if (fixedHeader.RemainingLength == 0)
+                return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2);
+            }
+
+            var bodyLength = fixedHeader.RemainingLength;
+            var body = new byte[bodyLength];
+
+            var bodyOffset = 0;
+            var chunkSize = Math.Min(ReadBufferSize, bodyLength);
+
+            do
+            {
+                var bytesLeft = body.Length - bodyOffset;
+                if (chunkSize > bytesLeft)
                 {
-                    return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2);
+                    chunkSize = bytesLeft;
                 }
 
-                var bodyLength = fixedHeader.RemainingLength;
-                var body = new byte[bodyLength];
+                var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
 
-                var bodyOffset = 0;
-                var chunkSize = Math.Min(ReadBufferSize, bodyLength);
-
-                do
+                if (cancellationToken.IsCancellationRequested)
                 {
-                    var bytesLeft = body.Length - bodyOffset;
-                    if (chunkSize > bytesLeft)
-                    {
-                        chunkSize = bytesLeft;
-                    }
+                    return ReceivedMqttPacket.Empty;
+                }
 
-                    var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
+                if (readBytes == 0)
+                {
+                    return ReceivedMqttPacket.Empty;
+                }
 
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        return ReceivedMqttPacket.Empty;
-                    }
+                bodyOffset += readBytes;
+            } while (bodyOffset < bodyLength);
 
-                    if (readBytes == 0)
-                    {
-                        return ReceivedMqttPacket.Empty;
-                    }
+            PacketInspector?.FillReceiveBuffer(body);
 
-                    bodyOffset += readBytes;
-                } while (bodyOffset < bodyLength);
-
-                PacketInspector?.FillReceiveBuffer(body);
-
-                var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
-                return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
-            }
-            finally
-            {
-                IsReadingPacket = false;
-            }
+            var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
+            return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
         }
 
         static bool WrapAndThrowException(Exception exception)

Source/MQTTnet/Internal/MqttPacketBus.cs

@@ -104,17 +104,22 @@ namespace MQTTnet.Internal
             _signal.Dispose();
         }
 
-        public void DropFirstItem(MqttPacketBusPartition partition)
+        public MqttPacketBusItem DropFirstItem(MqttPacketBusPartition partition)
         {
             lock (_syncRoot)
             {
                 var partitionInstance = _partitions[(int)partition];
 
-                if (partitionInstance.Any())
+                if (partitionInstance.Count > 0)
                 {
+                    var firstItem = partitionInstance.First.Value;
                     partitionInstance.RemoveFirst();
+
+                    return firstItem;
                 }
             }
+
+            return null;
         }
 
         public void EnqueueItem(MqttPacketBusItem item, MqttPacketBusPartition partition)

Source/MQTTnet/Server/Events/ApplicationMessageEnqueuedEventArgs.cs

@@ -0,0 +1,27 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+
+namespace MQTTnet.Server
+{
+    public sealed class ApplicationMessageEnqueuedEventArgs : EventArgs
+    {
+        public ApplicationMessageEnqueuedEventArgs(string senderClientId, string receiverClientId, MqttApplicationMessage applicationMessage, bool isDropped)
+        {
+            SenderClientId = senderClientId ?? throw new ArgumentNullException( nameof(senderClientId));
+            ReceiverClientId = receiverClientId ?? throw new ArgumentNullException(nameof(receiverClientId));
+            ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+            IsDropped = isDropped;
+        }
+
+        public string SenderClientId { get; }
+
+        public string ReceiverClientId { get; }
+
+        public bool IsDropped { get; }
+
+        public MqttApplicationMessage ApplicationMessage { get; }
+    }
+}

Source/MQTTnet/Server/Events/QueueMessageOverwrittenEventArgs.cs

@@ -0,0 +1,22 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Server
+{
+    public sealed class QueueMessageOverwrittenEventArgs : EventArgs
+    {
+        public QueueMessageOverwrittenEventArgs(string receiverClientId, MqttPacket packet)
+        {
+            ReceiverClientId = receiverClientId ?? throw new ArgumentNullException(nameof(receiverClientId));
+            Packet = packet ?? throw new ArgumentNullException(nameof(packet));
+        }
+
+        public MqttPacket Packet { get; }
+
+        public string ReceiverClientId { get; }
+    }
+}

Source/MQTTnet/Server/Internal/EnqueueDataPacketResult.cs

@@ -0,0 +1,12 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace MQTTnet.Server
+{
+    public enum EnqueueDataPacketResult
+    {
+        Enqueued,
+        Dropped
+    }
+}

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -244,7 +244,14 @@ namespace MQTTnet.Server
 
                         matchingSubscribersCount++;
 
-                        session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketCopy));
+                        var result = session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketCopy));
+
+                        if (_eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.HasHandlers)
+                        {
+                            var eventArgs = new ApplicationMessageEnqueuedEventArgs(senderId, session.Id, applicationMessage, result == EnqueueDataPacketResult.Dropped);
+                            await _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+                        }
+
                         _logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'", session.Id, applicationMessage.Topic);
                     }
 

Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs

@@ -22,6 +22,10 @@ namespace MQTTnet.Server
         public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<ClientUnsubscribedTopicEventArgs>();
 
         public AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs> InterceptingClientEnqueueEvent { get; } = new AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs>();
+        
+        public AsyncEvent<ApplicationMessageEnqueuedEventArgs> ApplicationMessageEnqueuedOrDroppedEvent { get; } = new AsyncEvent<ApplicationMessageEnqueuedEventArgs>();
+
+        public AsyncEvent<QueueMessageOverwrittenEventArgs> QueuedApplicationMessageOverwrittenEvent { get; } = new AsyncEvent<QueueMessageOverwrittenEventArgs>();
 
         public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
 

Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs

@@ -107,13 +107,6 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                if (connection.ChannelAdapter.IsReadingPacket)
-                {
-                    // The connection is currently reading a (large) packet. So it is obviously 
-                    // doing something and thus "connected".
-                    return;
-                }
-
                 // Values described here: [MQTT-3.1.2-24].
                 // If the client sends 5 sec. the server will allow up to 7.5 seconds.
                 // If the client sends 1 sec. the server will allow up to 1.5 seconds.

Source/MQTTnet/Server/Internal/MqttSession.cs

@@ -18,6 +18,7 @@ namespace MQTTnet.Server
     public sealed class MqttSession : IDisposable
     {
         readonly MqttClientSessionsManager _clientSessionsManager;
+        readonly MqttServerEventContainer _eventContainer;
         readonly MqttPacketBus _packetBus = new MqttPacketBus();
         readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
 
@@ -44,6 +45,7 @@ namespace MQTTnet.Server
             _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));
+            _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
 
             _subscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager, clientSessionsManager);
         }
@@ -117,20 +119,25 @@ namespace MQTTnet.Server
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
         }
 
-        public void EnqueueDataPacket(MqttPacketBusItem packetBusItem)
+        public EnqueueDataPacketResult EnqueueDataPacket(MqttPacketBusItem packetBusItem)
         {
             if (_packetBus.ItemsCount(MqttPacketBusPartition.Data) >= _serverOptions.MaxPendingMessagesPerClient)
             {
                 if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
                 {
-                    return;
+                    return EnqueueDataPacketResult.Dropped;
                 }
 
                 if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
                 {
                     // Only drop from the data partition. Dropping from control partition might break the connection
                     // because the client does not receive PINGREQ packets etc. any longer.
-                    _packetBus.DropFirstItem(MqttPacketBusPartition.Data);
+                    var firstItem = _packetBus.DropFirstItem(MqttPacketBusPartition.Data);
+                    if (firstItem != null && _eventContainer.QueuedApplicationMessageOverwrittenEvent.HasHandlers)
+                    {
+                        var eventArgs = new QueueMessageOverwrittenEventArgs(Id, firstItem.Packet);
+                        _eventContainer.QueuedApplicationMessageOverwrittenEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+                    }
                 }
             }
 
@@ -147,6 +154,7 @@ namespace MQTTnet.Server
             }
 
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
+            return EnqueueDataPacketResult.Enqueued;
         }
 
         public void EnqueueHealthPacket(MqttPacketBusItem packetBusItem)

Source/MQTTnet/Server/MqttServer.cs

@@ -51,6 +51,12 @@ namespace MQTTnet.Server
             _keepAliveMonitor = new MqttServerKeepAliveMonitor(options, _clientSessionsManager, _rootLogger);
         }
 
+        public event Func<ApplicationMessageEnqueuedEventArgs, Task> ApplicationMessageEnqueuedOrDroppedAsync
+        {
+            add => _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.AddHandler(value);
+            remove => _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.RemoveHandler(value);
+        }
+
         public event Func<ApplicationMessageNotConsumedEventArgs, Task> ApplicationMessageNotConsumedAsync
         {
             add => _eventContainer.ApplicationMessageNotConsumedEvent.AddHandler(value);
@@ -135,6 +141,12 @@ namespace MQTTnet.Server
             remove => _eventContainer.PreparingSessionEvent.RemoveHandler(value);
         }
 
+        public event Func<QueueMessageOverwrittenEventArgs, Task> QueuedApplicationMessageOverwrittenAsync
+        {
+            add => _eventContainer.QueuedApplicationMessageOverwrittenEvent.AddHandler(value);
+            remove => _eventContainer.QueuedApplicationMessageOverwrittenEvent.RemoveHandler(value);
+        }
+
         public event Func<RetainedMessageChangedEventArgs, Task> RetainedMessageChangedAsync
         {
             add => _eventContainer.RetainedMessageChangedEvent.AddHandler(value);

Source/MQTTnet/Server/Options/MqttServerKeepAliveOptions.cs

@@ -0,0 +1,21 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttServerKeepAliveOptions
+    {
+        /// <summary>
+        ///     When this mode is enabled the MQTT server will not close a connection when the
+        ///     client is currently sending a (large) payload. This may lead to "dead" connections
+        ///     When this mode is disabled the MQTT server will disconnect a client when the keep
+        ///     alive timeout is reached even if the client is currently sending a (large) payload.
+        /// </summary>
+        public bool DisconnectClientWhenReadingPayload { get; set; }
+
+        public TimeSpan MonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500);
+    }
+}

Source/MQTTnet/Server/Options/MqttServerOptions.cs

@@ -9,12 +9,19 @@ namespace MQTTnet.Server
     public sealed class MqttServerOptions
     {
         public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(100);
-        
+
         public MqttServerTcpEndpointOptions DefaultEndpointOptions { get; } = new MqttServerTcpEndpointOptions();
 
         public bool EnablePersistentSessions { get; set; }
 
-        public TimeSpan KeepAliveMonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500);
+        [Obsolete("Use KeepAliveOptions instead.")]
+        public TimeSpan KeepAliveMonitorInterval
+        {
+            get => KeepAliveOptions.MonitorInterval;
+            set => KeepAliveOptions.MonitorInterval = value;
+        }
+
+        public MqttServerKeepAliveOptions KeepAliveOptions { get; } = new MqttServerKeepAliveOptions();
 
         public int MaxPendingMessagesPerClient { get; set; } = 250;