Adds events for delivered and dropped messages. (#1866)

kallayj · chkr1011 · web-flow · commit 09ab30fb0f60 · 2023-12-06T18:02:48.000+01:00
* Adds events for delivered and dropped messages.

* Rename new events to match other namings

* Update ReleaseNotes.md

---------

Co-authored-by: Christian &lt;6939810+chkr1011@users.noreply.github.com&gt;
diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md
@@ -1,7 +1,7 @@
 * [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).
diff --git a/Source/MQTTnet/Internal/MqttPacketBus.cs b/Source/MQTTnet/Internal/MqttPacketBus.cs
@@ -104,17 +104,22 @@ public void Dispose()
             _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)
diff --git a/Source/MQTTnet/Server/Events/ApplicationMessageEnqueuedEventArgs.cs b/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; }
+    }
+}
diff --git a/Source/MQTTnet/Server/Events/QueueMessageOverwrittenEventArgs.cs b/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; }
+    }
+}
diff --git a/Source/MQTTnet/Server/Internal/EnqueueDataPacketResult.cs b/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
+    }
+}
diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
@@ -244,7 +244,14 @@ public async Task<DispatchApplicationMessageResult> DispatchApplicationMessage(
 
                         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);
                     }
 
diff --git a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs
@@ -22,6 +22,10 @@ public sealed class MqttServerEventContainer
         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>();
 
diff --git a/Source/MQTTnet/Server/Internal/MqttSession.cs b/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 @@ public MqttSession(
             _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 @@ public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)
             _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 @@ public void EnqueueDataPacket(MqttPacketBusItem packetBusItem)
             }
 
             _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
+            return EnqueueDataPacketResult.Enqueued;
         }
 
         public void EnqueueHealthPacket(MqttPacketBusItem packetBusItem)
diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs
@@ -51,6 +51,12 @@ public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> ada
             _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 @@ public event Func<EventArgs, Task> PreparingSessionAsync
             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);

Original file line number	Diff line number	Diff line change
`@@ -104,17 +104,22 @@ public void Dispose()`
`104`	`104`	`_signal.Dispose();`
`105`	`105`	`}`
`106`	`106`
`107`		`- public void DropFirstItem(MqttPacketBusPartition partition)`
	`107`	`+ public MqttPacketBusItem DropFirstItem(MqttPacketBusPartition partition)`
`108`	`108`	`{`
`109`	`109`	`lock (_syncRoot)`
`110`	`110`	`{`
`111`	`111`	`var partitionInstance = _partitions[(int)partition];`
`112`	`112`
`113`		`- if (partitionInstance.Any())`
	`113`	`+ if (partitionInstance.Count > 0)`
`114`	`114`	`{`
	`115`	`+ var firstItem = partitionInstance.First.Value;`
`115`	`116`	`partitionInstance.RemoveFirst();`
	`117`	`+`
	`118`	`+ return firstItem;`
`116`	`119`	`}`
`117`	`120`	`}`
	`121`	`+`
	`122`	`+ return null;`
`118`	`123`	`}`
`119`	`124`
`120`	`125`	`public void EnqueueItem(MqttPacketBusItem item, MqttPacketBusPartition partition)`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@ namespace MQTTnet.Server`
`18`	`18`	`public sealed class MqttSession : IDisposable`
`19`	`19`	`{`
`20`	`20`	`readonly MqttClientSessionsManager _clientSessionsManager;`
	`21`	`+ readonly MqttServerEventContainer _eventContainer;`
`21`	`22`	`readonly MqttPacketBus _packetBus = new MqttPacketBus();`
`22`	`23`	`readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();`
`23`	`24`
`@@ -44,6 +45,7 @@ public MqttSession(`
`44`	`45`	`_connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));`
`45`	`46`	`_serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));`
`46`	`47`	`_clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));`
	`48`	`+ _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));`
`47`	`49`
`48`	`50`	`_subscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager, clientSessionsManager);`
`49`	`51`	`}`
`@@ -117,20 +119,25 @@ public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)`
`117`	`119`	`_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);`
`118`	`120`	`}`
`119`	`121`
`120`		`- public void EnqueueDataPacket(MqttPacketBusItem packetBusItem)`
	`122`	`+ public EnqueueDataPacketResult EnqueueDataPacket(MqttPacketBusItem packetBusItem)`
`121`	`123`	`{`
`122`	`124`	`if (_packetBus.ItemsCount(MqttPacketBusPartition.Data) >= _serverOptions.MaxPendingMessagesPerClient)`
`123`	`125`	`{`
`124`	`126`	`if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)`
`125`	`127`	`{`
`126`		`- return;`
	`128`	`+ return EnqueueDataPacketResult.Dropped;`
`127`	`129`	`}`
`128`	`130`
`129`	`131`	`if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)`
`130`	`132`	`{`
`131`	`133`	`// Only drop from the data partition. Dropping from control partition might break the connection`
`132`	`134`	`// because the client does not receive PINGREQ packets etc. any longer.`
`133`		`- _packetBus.DropFirstItem(MqttPacketBusPartition.Data);`
	`135`	`+ var firstItem = _packetBus.DropFirstItem(MqttPacketBusPartition.Data);`
	`136`	`+ if (firstItem != null && _eventContainer.QueuedApplicationMessageOverwrittenEvent.HasHandlers)`
	`137`	`+ {`
	`138`	`+ var eventArgs = new QueueMessageOverwrittenEventArgs(Id, firstItem.Packet);`
	`139`	`+ _eventContainer.QueuedApplicationMessageOverwrittenEvent.InvokeAsync(eventArgs).ConfigureAwait(false);`
	`140`	`+ }`
`134`	`141`	`}`
`135`	`142`	`}`
`136`	`143`
`@@ -147,6 +154,7 @@ public void EnqueueDataPacket(MqttPacketBusItem packetBusItem)`
`147`	`154`	`}`
`148`	`155`
`149`	`156`	`_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);`
	`157`	`+ return EnqueueDataPacketResult.Enqueued;`
`150`	`158`	`}`
`151`	`159`
`152`	`160`	`public void EnqueueHealthPacket(MqttPacketBusItem packetBusItem)`
Original file line number	Diff line number	Diff line change
`@@ -51,6 +51,12 @@ public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> ada`
`51`	`51`	`_keepAliveMonitor = new MqttServerKeepAliveMonitor(options, _clientSessionsManager, _rootLogger);`
`52`	`52`	`}`
`53`	`53`
	`54`	`+ public event Func<ApplicationMessageEnqueuedEventArgs, Task> ApplicationMessageEnqueuedOrDroppedAsync`
	`55`	`+ {`
	`56`	`+ add => _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.AddHandler(value);`
	`57`	`+ remove => _eventContainer.ApplicationMessageEnqueuedOrDroppedEvent.RemoveHandler(value);`
	`58`	`+ }`
	`59`	`+`
`54`	`60`	`public event Func<ApplicationMessageNotConsumedEventArgs, Task> ApplicationMessageNotConsumedAsync`
`55`	`61`	`{`
`56`	`62`	`add => _eventContainer.ApplicationMessageNotConsumedEvent.AddHandler(value);`
`@@ -135,6 +141,12 @@ public event Func<EventArgs, Task> PreparingSessionAsync`
`135`	`141`	`remove => _eventContainer.PreparingSessionEvent.RemoveHandler(value);`
`136`	`142`	`}`
`137`	`143`
	`144`	`+ public event Func<QueueMessageOverwrittenEventArgs, Task> QueuedApplicationMessageOverwrittenAsync`
	`145`	`+ {`
	`146`	`+ add => _eventContainer.QueuedApplicationMessageOverwrittenEvent.AddHandler(value);`
	`147`	`+ remove => _eventContainer.QueuedApplicationMessageOverwrittenEvent.RemoveHandler(value);`
	`148`	`+ }`
	`149`	`+`
`138`	`150`	`public event Func<RetainedMessageChangedEventArgs, Task> RetainedMessageChangedAsync`
`139`	`151`	`{`
`140`	`152`	`add => _eventContainer.RetainedMessageChangedEvent.AddHandler(value);`