YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-18 10:26:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Improve Server packet management.

Browse Source
This commit is contained in:
Christian
2022-04-21 20:48:52 +02:00
parent 920728e77a
commit fcf276dd1a
8 changed files with 101 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs
+35 -4
View File
@@ -26,6 +26,37 @@ namespace MQTTnet.Tests.Internal
}
}
[TestMethod]
public async Task Fill_From_Different_Task()
{
const int MessageCount = 500;
var delayRandom = new Random();
var bus = new MqttPacketBus();
_ = Task.Run(
() =>
{
for (int i = 0; i < MessageCount; i++)
{
bus.EnqueueItem(new MqttPacketBusItem(MqttPingReqPacket.Instance), MqttPacketBusPartition.Health);
Thread.Sleep(delayRandom.Next(0, 10));
}
});
for (var i = 0; i < MessageCount; i++)
{
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)))
{
await bus.DequeueItemAsync(timeout.Token);
}
}
Assert.AreEqual(0, bus.TotalItemsCount);
}
[TestMethod]
public void Alternate_Priorities()
{
@@ -43,7 +74,7 @@ namespace MQTTnet.Tests.Internal
bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
Assert.AreEqual(9, bus.ItemsCount);
Assert.AreEqual(9, bus.TotalItemsCount);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
@@ -55,7 +86,7 @@ namespace MQTTnet.Tests.Internal
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
Assert.AreEqual(0, bus.ItemsCount);
Assert.AreEqual(0, bus.TotalItemsCount);
}
[TestMethod]
@@ -67,7 +98,7 @@ namespace MQTTnet.Tests.Internal
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
Assert.AreEqual(3, bus.ItemsCount);
Assert.AreEqual(3, bus.TotalItemsCount);
var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control);
Assert.AreEqual(0, exportedPackets.Count);
@@ -78,7 +109,7 @@ namespace MQTTnet.Tests.Internal
exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data);
Assert.AreEqual(3, exportedPackets.Count);
Assert.AreEqual(3, bus.ItemsCount);
Assert.AreEqual(3, bus.TotalItemsCount);
}
[TestMethod]
Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs
-1
View File
@@ -3,7 +3,6 @@ using System.Threading;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Tests.Mockups;
Source/MQTTnet/Internal/MqttPacketBus.cs
+32 -7
View File
@@ -20,12 +20,12 @@ namespace MQTTnet.Internal
new LinkedList<MqttPacketBusItem>()
};
readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0, 1);
readonly object _syncRoot = new object();
int _activePartition = (int)MqttPacketBusPartition.Health;
public int ItemsCount
public int TotalItemsCount
{
get
{
@@ -55,12 +55,26 @@ namespace MQTTnet.Internal
{
for (var i = 0; i < 3; i++)
{
// Iterate through the partitions in order to ensure processing of health packets
// even if lots of data packets are enqueued.
// Partition | Messages (left = oldest).
// DATA | [#]#########################
// CONTROL | [#]#############
// HEALTH | [#]####
// In this sample the 3 oldest messages from the partitions are processed in a row.
// Then the next 3 from all 3 partitions.
MoveActivePartition();
if (_partitions[_activePartition].Count > 0)
var activePartition = _partitions[_activePartition];
if (activePartition.First != null)
{
var item = _partitions[_activePartition].First;
_partitions[_activePartition].RemoveFirst();
var item = activePartition.First;
activePartition.RemoveFirst();
return item.Value;
}
}
@@ -104,9 +118,12 @@ namespace MQTTnet.Internal
lock (_syncRoot)
{
_partitions[(int)partition].AddLast(item);
}
_semaphore.Release();
if (_semaphore.CurrentCount == 0)
{
_semaphore.Release();
}
}
}
public List<MqttPacket> ExportPackets(MqttPacketBusPartition partition)
@@ -117,6 +134,14 @@ namespace MQTTnet.Internal
}
}
public int ItemsCount(MqttPacketBusPartition partition)
{
lock (_syncRoot)
{
return _partitions[(int)partition].Count;
}
}
public int PartitionItemsCount(MqttPacketBusPartition partition)
{
lock (_syncRoot)
Source/MQTTnet/Server/Internal/MqttClient.cs
+9 -11
View File
@@ -189,7 +189,6 @@ namespace MQTTnet.Server
while (!cancellationToken.IsCancellationRequested)
{
var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
if (packet == null)
{
return;
@@ -251,8 +250,7 @@ namespace MQTTnet.Server
else if (packet is MqttPingReqPacket)
{
// See: The Server MUST send a PINGRESP packet in response to a PINGREQ packet [MQTT-3.12.4-1].
//await SendPacketAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false);
Session.EnqueuePacket(new MqttPacketBusItem(MqttPingRespPacket.Instance));
Session.EnqueueHealthPacket(new MqttPacketBusItem(MqttPingRespPacket.Instance));
}
else if (packet is MqttPingRespPacket)
{
@@ -342,7 +340,7 @@ namespace MQTTnet.Server
if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
publishPacket.Dup = true;
Session.EnqueuePacket(new MqttPacketBusItem(publishPacket));
Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
}
@@ -366,13 +364,13 @@ namespace MQTTnet.Server
void HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
{
var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
Session.EnqueuePacket(new MqttPacketBusItem(pubRelPacket));
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
}
void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)
{
var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
Session.EnqueuePacket(new MqttPacketBusItem(pubCompPacket));
Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket));
}
async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
@@ -381,7 +379,7 @@ namespace MQTTnet.Server
var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult);
Session.EnqueuePacket(new MqttPacketBusItem(subAckPacket));
Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket));
if (subscribeResult.CloseConnection)
{
@@ -394,7 +392,7 @@ namespace MQTTnet.Server
foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages)
{
var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
Session.EnqueuePacket(new MqttPacketBusItem(publishPacket));
Session.EnqueueControlPacket(new MqttPacketBusItem(publishPacket));
}
}
}
@@ -405,7 +403,7 @@ namespace MQTTnet.Server
var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult);
Session.EnqueuePacket(new MqttPacketBusItem(unsubAckPacket));
Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket));
if (unsubscribeResult.CloseConnection)
{
@@ -470,13 +468,13 @@ namespace MQTTnet.Server
case MqttQualityOfServiceLevel.AtLeastOnce:
{
var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs);
Session.EnqueuePacket(new MqttPacketBusItem(pubAckPacket));
Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket));
break;
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs);
Session.EnqueuePacket(new MqttPacketBusItem(pubRecPacket));
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket));
break;
}
default:
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+3 -3
View File
@@ -181,7 +181,7 @@ namespace MQTTnet.Server
newPublishPacket.Retain = false;
}
session.EnqueuePacket(new MqttPacketBusItem(newPublishPacket));
session.EnqueueDataPacket(new MqttPacketBusItem(newPublishPacket));
deliveryCount++;
_logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'.", session.Id, applicationMessage.Topic);
@@ -229,7 +229,7 @@ namespace MQTTnet.Server
}
}
public Task<IList<MqttClientStatus>> GetClientStatusAsync()
public Task<IList<MqttClientStatus>> GetClientStatusesAsync()
{
var result = new List<MqttClientStatus>();
@@ -424,7 +424,7 @@ namespace MQTTnet.Server
foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages)
{
var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
clientSession.EnqueuePacket(new MqttPacketBusItem(publishPacket));
clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
}
}
Source/MQTTnet/Server/Internal/MqttSession.cs
+19 -24
View File
@@ -24,7 +24,7 @@ namespace MQTTnet.Server
readonly Dictionary<ushort, MqttPublishPacket> _unacknowledgedPublishPackets = new Dictionary<ushort, MqttPublishPacket>();
// Bookkeeping to know if this is a subscribing client; lazy intialize later.
// Bookkeeping to know if this is a subscribing client; lazy initialize later.
HashSet<string> _subscribedTopics;
public MqttSession(
@@ -100,14 +100,9 @@ namespace MQTTnet.Server
SubscriptionsManager.Dispose();
}
public void EnqueuePacket(MqttPacketBusItem packetBusItem)
public void EnqueueDataPacket(MqttPacketBusItem packetBusItem)
{
if (packetBusItem == null)
{
throw new ArgumentNullException(nameof(packetBusItem));
}
if (_packetBus.ItemsCount >= _serverOptions.MaxPendingMessagesPerClient)
if (_packetBus.ItemsCount(MqttPacketBusPartition.Data) >= _serverOptions.MaxPendingMessagesPerClient)
{
if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
{
@@ -122,25 +117,25 @@ namespace MQTTnet.Server
}
}
if (packetBusItem.Packet is MqttPublishPacket publishPacket)
var publishPacket = (MqttPublishPacket)packetBusItem.Packet;
if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
_unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket;
}
_unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket;
}
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
}
else if (packetBusItem.Packet is MqttPingReqPacket || packetBusItem.Packet is MqttPingRespPacket)
{
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health);
}
else
{
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
}
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
}
public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)
{
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control);
}
public void EnqueueHealthPacket(MqttPacketBusItem packetBusItem)
{
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health);
}
public void Recover()
{
// TODO: Keep the bus and only insert pending items again.
@@ -149,7 +144,7 @@ namespace MQTTnet.Server
foreach (var publishPacket in _unacknowledgedPublishPackets.Values.ToList())
{
EnqueuePacket(new MqttPacketBusItem(publishPacket));
EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
}
Source/MQTTnet/Server/MqttServer.cs
+1 -1
View File
@@ -182,7 +182,7 @@ namespace MQTTnet.Server
{
ThrowIfNotStarted();
return _clientSessionsManager.GetClientStatusAsync();
return _clientSessionsManager.GetClientStatusesAsync();
}
public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
Source/MQTTnet/Server/Status/MqttSessionStatus.cs
+2 -2
View File
@@ -33,7 +33,7 @@ namespace MQTTnet.Server
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
var publishPacketFactory = new MqttPublishPacketFactory();
_session.EnqueuePacket(new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage)));
_session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage)));
return PlatformAbstractionLayer.CompletedTask;
}
@@ -44,7 +44,7 @@ namespace MQTTnet.Server
var publishPacketFactory = new MqttPublishPacketFactory();
var packetBusItem = new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage));
_session.EnqueuePacket(packetBusItem);
_session.EnqueueDataPacket(packetBusItem);
return packetBusItem.WaitForDeliveryAsync();
}