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

Add session to subscribe interceptor.

Browse Source
This commit is contained in:
Christian
2021-10-14 21:50:16 +02:00
parent 87df9367a7
commit a15fbf532e
16 changed files with 130 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MQTTnet.sln.DotSettings
+1
View File
@@ -3,6 +3,7 @@
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_METHOD_CALL_PARENTHESES/@EntryValue">False</s:Boolean>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_METHOD_PARENTHESES/@EntryValue">False</s:Boolean>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_PARENTHESES/@EntryValue">False</s:Boolean>
<s:Int64 x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/WRAP_LIMIT/@EntryValue">180</s:Int64>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpAttributeForSingleLineMethodUpgrade/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
Source/MQTTnet/Server/Internal/MqttClientConnection.cs
+22 -38
View File
@@ -5,7 +5,6 @@ using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Client;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Diagnostics;
using MQTTnet.Diagnostics.Logger;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
@@ -14,7 +13,6 @@ using MQTTnet.Internal;
using MQTTnet.PacketDispatcher;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server.Status;
namespace MQTTnet.Server.Internal
{
@@ -25,12 +23,10 @@ namespace MQTTnet.Server.Internal
readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
readonly MqttClientSessionsManager _sessionsManager;
readonly IMqttChannelAdapter _channelAdapter;
readonly MqttNetSourceLogger _logger;
readonly IMqttServerOptions _serverOptions;
readonly string _endpoint;
readonly MqttConnectPacket _connectPacket;
CancellationTokenSource _cancellationToken;
@@ -46,8 +42,8 @@ namespace MQTTnet.Server.Internal
_serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
_sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
_channelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
_endpoint = channelAdapter.Endpoint;
ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
Endpoint = channelAdapter.Endpoint;
Session = session ?? throw new ArgumentNullException(nameof(session));
_connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
@@ -58,16 +54,16 @@ namespace MQTTnet.Server.Internal
public string ClientId => _connectPacket.ClientId;
public string Endpoint => _endpoint;
public string Endpoint {get;}
public IMqttChannelAdapter ChannelAdapter { get; }
public MqttClientConnectionStatistics Statistics { get; } = new MqttClientConnectionStatistics();
public bool IsRunning { get; private set; }
public ushort KeepAlivePeriod => _connectPacket.KeepAlivePeriod;
public bool IsReadingPacket => _channelAdapter.IsReadingPacket;
public MqttClientSession Session { get; }
public bool IsTakenOver { get; set; }
@@ -92,21 +88,9 @@ namespace MQTTnet.Server.Internal
public void ResetStatistics()
{
_channelAdapter.ResetStatistics();
ChannelAdapter.ResetStatistics();
}
public void FillClientStatus(MqttClientStatus clientStatus)
{
clientStatus.ClientId = ClientId;
clientStatus.Endpoint = _endpoint;
clientStatus.ProtocolVersion = _channelAdapter.PacketFormatterAdapter.ProtocolVersion;
clientStatus.BytesSent = _channelAdapter.BytesSent;
clientStatus.BytesReceived = _channelAdapter.BytesReceived;
Statistics.FillClientStatus(clientStatus);
}
public void Dispose()
{
_cancellationToken.Dispose();
@@ -154,7 +138,7 @@ namespace MQTTnet.Server.Internal
Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
{
return _channelAdapter.SendPacketAsync(packet, cancellationToken).ContinueWith(task => { Statistics.HandleSentPacket(packet); }, cancellationToken);
return ChannelAdapter.SendPacketAsync(packet, cancellationToken).ContinueWith(task => { Statistics.HandleSentPacket(packet); }, cancellationToken);
}
async Task ReceivePackagesLoop(CancellationToken cancellationToken)
@@ -166,7 +150,7 @@ namespace MQTTnet.Server.Internal
// own exception in the reading loop!
while (!cancellationToken.IsCancellationRequested)
{
var packet = await _channelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
if (packet == null)
{
// The client has closed the connection gracefully.
@@ -278,7 +262,7 @@ namespace MQTTnet.Server.Internal
await SendPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false);
var pubRecPacket = await awaitableRec.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
var pubRelPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
var pubRelPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRelPacket(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
await SendPacketAsync(pubRelPacket, cancellationToken).ConfigureAwait(false);
await awaitableComp.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
@@ -321,10 +305,10 @@ namespace MQTTnet.Server.Internal
async Task<MqttPublishPacket> CreatePublishPacket(MqttQueuedApplicationMessage queuedApplicationMessage)
{
var publishPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage);
var publishPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage);
publishPacket.QualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel;
if (_channelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
{
publishPacket.Properties.SubscriptionIdentifiers = queuedApplicationMessage.SubscriptionIdentifiers;
}
@@ -354,14 +338,14 @@ namespace MQTTnet.Server.Internal
Task HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
{
var pubCompPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
var pubCompPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubCompPacket(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
return SendPacketAsync(pubCompPacket, cancellationToken);
}
async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{
var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket).ConfigureAwait(false);
var subAckPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreateSubAckPacket(subscribePacket, subscribeResult);
var subAckPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreateSubAckPacket(subscribePacket, subscribeResult);
await SendPacketAsync(subAckPacket, cancellationToken).ConfigureAwait(false);
@@ -380,7 +364,7 @@ namespace MQTTnet.Server.Internal
async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
{
var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket).ConfigureAwait(false);
var unsubAckPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreateUnsubAckPacket(unsubscribePacket, unsubscribeResult);
var unsubAckPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreateUnsubAckPacket(unsubscribePacket, unsubscribeResult);
await SendPacketAsync(unsubAckPacket, cancellationToken).ConfigureAwait(false);
@@ -394,7 +378,7 @@ namespace MQTTnet.Server.Internal
{
HandleTopicAlias(publishPacket);
var applicationMessage = _channelAdapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket);
var applicationMessage = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket);
_sessionsManager.DispatchApplicationMessage(applicationMessage, this);
switch (publishPacket.QualityOfServiceLevel)
@@ -405,12 +389,12 @@ namespace MQTTnet.Server.Internal
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
var pubAckPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
var pubAckPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
return SendPacketAsync(pubAckPacket, cancellationToken);
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
var pubRecPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
var pubRecPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
return SendPacketAsync(pubRecPacket, cancellationToken);
}
default:
@@ -458,7 +442,7 @@ namespace MQTTnet.Server.Internal
ReasonString = reason.ToString()
};
var disconnectPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreateDisconnectPacket(disconnectOptions);
var disconnectPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreateDisconnectPacket(disconnectOptions);
using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
{
Source/MQTTnet/Server/Internal/MqttClientConnectionStatistics.cs
+22 -32
View File
@@ -7,12 +7,6 @@ namespace MQTTnet.Server.Internal
{
public sealed class MqttClientConnectionStatistics
{
readonly DateTime _connectedTimestamp;
DateTime _lastNonKeepAlivePacketReceivedTimestamp;
DateTime _lastPacketReceivedTimestamp;
DateTime _lastPacketSentTimestamp;
// Start with 1 because the CONNACK packet is not counted here.
long _receivedPacketsCount = 1;
@@ -24,22 +18,36 @@ namespace MQTTnet.Server.Internal
public MqttClientConnectionStatistics()
{
_connectedTimestamp = DateTime.UtcNow;
ConnectedTimestamp = DateTime.UtcNow;
_lastPacketReceivedTimestamp = _connectedTimestamp;
_lastPacketSentTimestamp = _connectedTimestamp;
LastPacketReceivedTimestamp = ConnectedTimestamp;
LastPacketSentTimestamp = ConnectedTimestamp;
_lastNonKeepAlivePacketReceivedTimestamp = _connectedTimestamp;
LastNonKeepAlivePacketReceivedTimestamp = ConnectedTimestamp;
}
public DateTime LastPacketReceivedTimestamp => _lastPacketReceivedTimestamp;
public DateTime ConnectedTimestamp { get; }
public DateTime LastPacketReceivedTimestamp { get; private set; }
public DateTime LastPacketSentTimestamp { get; private set; }
public DateTime LastNonKeepAlivePacketReceivedTimestamp { get; private set; }
public long SentApplicationMessagesCount => Interlocked.Read(ref _sentApplicationMessagesCount);
public long ReceivedApplicationMessagesCount => Interlocked.Read(ref _receivedApplicationMessagesCount);
public long SentPacketsCount => Interlocked.Read(ref _sentPacketsCount);
public long ReceivedPacketsCount => Interlocked.Read(ref _receivedPacketsCount);
public void HandleReceivedPacket(MqttBasePacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
// This class is tracking all values from Clients perspective!
_lastPacketSentTimestamp = DateTime.UtcNow;
LastPacketSentTimestamp = DateTime.UtcNow;
Interlocked.Increment(ref _sentPacketsCount);
@@ -50,7 +58,7 @@ namespace MQTTnet.Server.Internal
if (!(packet is MqttPingReqPacket || packet is MqttPingRespPacket))
{
_lastNonKeepAlivePacketReceivedTimestamp = _lastPacketReceivedTimestamp;
LastNonKeepAlivePacketReceivedTimestamp = LastPacketReceivedTimestamp;
}
}
@@ -59,7 +67,7 @@ namespace MQTTnet.Server.Internal
if (packet == null) throw new ArgumentNullException(nameof(packet));
// This class is tracking all values from Clients perspective!
_lastPacketReceivedTimestamp = DateTime.UtcNow;
LastPacketReceivedTimestamp = DateTime.UtcNow;
Interlocked.Increment(ref _receivedPacketsCount);
@@ -68,23 +76,5 @@ namespace MQTTnet.Server.Internal
Interlocked.Increment(ref _receivedApplicationMessagesCount);
}
}
public void FillClientStatus(MqttClientStatus clientStatus)
{
if (clientStatus == null) throw new ArgumentNullException(nameof(clientStatus));
clientStatus.ConnectedTimestamp = _connectedTimestamp;
clientStatus.ReceivedPacketsCount = Interlocked.Read(ref _receivedPacketsCount);
clientStatus.SentPacketsCount = Interlocked.Read(ref _sentPacketsCount);
clientStatus.ReceivedApplicationMessagesCount = Interlocked.Read(ref _receivedApplicationMessagesCount);
clientStatus.SentApplicationMessagesCount = Interlocked.Read(ref _sentApplicationMessagesCount);
clientStatus.LastPacketReceivedTimestamp = _lastPacketReceivedTimestamp;
clientStatus.LastPacketSentTimestamp = _lastPacketSentTimestamp;
clientStatus.LastNonKeepAlivePacketReceivedTimestamp = _lastNonKeepAlivePacketReceivedTimestamp;
}
}
}
Source/MQTTnet/Server/Internal/MqttClientSession.cs
+14 -17
View File
@@ -1,20 +1,21 @@
using System;
using System.Collections.Generic;
using MQTTnet.Server.Status;
using System.Threading.Tasks;
namespace MQTTnet.Server.Internal
{
public sealed class MqttClientSession
{
readonly DateTime _createdTimestamp = DateTime.UtcNow;
readonly MqttClientSessionsManager _clientSessionsManager;
public MqttClientSession(
string clientId,
public MqttClientSession(string clientId,
IDictionary<object, object> items,
MqttServerEventDispatcher eventDispatcher,
IMqttServerOptions serverOptions,
IMqttRetainedMessagesManager retainedMessagesManager)
IMqttRetainedMessagesManager retainedMessagesManager,
MqttClientSessionsManager clientSessionsManager)
{
_clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
Items = items ?? throw new ArgumentNullException(nameof(items));
@@ -26,27 +27,23 @@ namespace MQTTnet.Server.Internal
public string ClientId { get; }
public DateTime CreatedTimestamp { get; } = DateTime.UtcNow;
public bool IsCleanSession { get; set; } = true;
public MqttApplicationMessage WillMessage { get; set; }
public MqttClientSubscriptionsManager SubscriptionsManager { get; }
public MqttClientSessionApplicationMessagesQueue ApplicationMessagesQueue { get; }
/// <summary>
/// Gets or sets a key/value collection that can be used to share data within the scope of this session.
/// </summary>
public IDictionary<object, object> Items { get; }
public void FillSessionStatus(MqttSessionStatus status)
{
status.ClientId = ClientId;
status.CreatedTimestamp = _createdTimestamp;
status.PendingApplicationMessagesCount = ApplicationMessagesQueue.Count;
status.Items = Items;
}
public Task DeleteAsync()
{
return _clientSessionsManager.DeleteSessionAsync(ClientId);
}
public void OnDeleted()
{
Deleted?.Invoke(this, EventArgs.Empty);
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+7 -11
View File
@@ -6,7 +6,6 @@ using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Diagnostics;
using MQTTnet.Diagnostics.Logger;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
@@ -190,12 +189,10 @@ namespace MQTTnet.Server.Internal
{
foreach (var connection in _clientConnections.Values)
{
var clientStatus = new MqttClientStatus(connection);
connection.FillClientStatus(clientStatus);
var sessionStatus = new MqttSessionStatus(connection.Session, this);
connection.Session.FillSessionStatus(sessionStatus);
clientStatus.Session = sessionStatus;
var clientStatus = new MqttClientStatus(connection)
{
Session = new MqttSessionStatus(connection.Session)
};
result.Add(clientStatus);
}
@@ -212,9 +209,7 @@ namespace MQTTnet.Server.Internal
{
foreach (var session in _clientSessions.Values)
{
var sessionStatus = new MqttSessionStatus(session, this);
session.FillSessionStatus(sessionStatus);
var sessionStatus = new MqttSessionStatus(session);
result.Add(sessionStatus);
}
}
@@ -571,7 +566,8 @@ namespace MQTTnet.Server.Internal
sessionItems,
_eventDispatcher,
_options,
_retainedMessagesManager);
_retainedMessagesManager,
this);
}
}
}
Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+5 -3
View File
@@ -5,6 +5,7 @@ using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server.Status;
namespace MQTTnet.Server.Internal
{
@@ -15,6 +16,7 @@ namespace MQTTnet.Server.Internal
// completely by swapping references etc.
readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim();
readonly Dictionary<string, Subscription> _subscriptions = new Dictionary<string, Subscription>(4096);
readonly MqttClientSession _clientSession;
readonly IMqttServerOptions _options;
readonly MqttServerEventDispatcher _eventDispatcher;
@@ -29,8 +31,7 @@ namespace MQTTnet.Server.Internal
_clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
_options = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
_eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
_retainedMessagesManager = retainedMessagesManager ??
throw new ArgumentNullException(nameof(retainedMessagesManager));
_retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
}
public async Task<SubscribeResult> Subscribe(MqttSubscribePacket subscribePacket)
@@ -297,7 +298,8 @@ namespace MQTTnet.Server.Internal
{
ClientId = _clientSession.ClientId,
TopicFilter = topicFilter,
SessionItems = _clientSession.Items
SessionItems = _clientSession.Items,
Session = new MqttSessionStatus(_clientSession)
};
if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs
+1 -1
View File
@@ -83,7 +83,7 @@ namespace MQTTnet.Server.Internal
return;
}
if (connection.IsReadingPacket)
if (connection.ChannelAdapter.IsReadingPacket)
{
// The connection is currently reading a (large) packet. So it is obviously
// doing something and thus "connected".
Source/MQTTnet/Server/MqttPendingMessagesOverflowStrategy.cs
+1
View File
@@ -3,6 +3,7 @@
public enum MqttPendingMessagesOverflowStrategy
{
DropOldestQueuedMessage,
DropNewMessage
}
}
Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs
+6
View File
@@ -1,6 +1,7 @@
using System;
using System.Collections.Generic;
using MQTTnet.Protocol;
using MQTTnet.Server.Status;
namespace MQTTnet.Server
{
@@ -68,5 +69,10 @@ namespace MQTTnet.Server
/// Gets or sets whether the broker should close the client connection.
/// </summary>
public bool CloseConnection { get; set; }
/// <summary>
/// Gets the current client session.
/// </summary>
public IMqttSessionStatus Session { get; internal set; }
}
}
Source/MQTTnet/Server/Status/IMqttClientStatus.cs
+2 -2
View File
@@ -16,11 +16,11 @@ namespace MQTTnet.Server.Status
MqttProtocolVersion ProtocolVersion { get; }
DateTime ConnectedTimestamp { get; set; }
DateTime ConnectedTimestamp { get; }
DateTime LastPacketReceivedTimestamp { get; }
DateTime LastPacketSentTimestamp { get; set; }
DateTime LastPacketSentTimestamp { get; }
DateTime LastNonKeepAlivePacketReceivedTimestamp { get; }
Source/MQTTnet/Server/Status/MqttClientStatus.cs
+15 -15
View File
@@ -19,33 +19,33 @@ namespace MQTTnet.Server.Status
/// Gets or sets the client identifier.
/// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
/// </summary>
public string ClientId { get; set; }
public string ClientId => _connection.ClientId;
public string Endpoint { get; set; }
public string Endpoint => _connection.Endpoint;
public MqttProtocolVersion ProtocolVersion { get; set; }
public MqttProtocolVersion ProtocolVersion => _connection.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion;
public DateTime ConnectedTimestamp { get; set; }
public DateTime LastPacketReceivedTimestamp { get; set; }
public DateTime LastPacketSentTimestamp { get; set; }
public DateTime ConnectedTimestamp => _connection.Statistics.ConnectedTimestamp;
public DateTime LastNonKeepAlivePacketReceivedTimestamp { get; set; }
public DateTime LastPacketReceivedTimestamp => _connection.Statistics.LastPacketReceivedTimestamp;
public long ReceivedApplicationMessagesCount { get; set; }
public DateTime LastPacketSentTimestamp => _connection.Statistics.LastPacketSentTimestamp;
public long SentApplicationMessagesCount { get; set; }
public DateTime LastNonKeepAlivePacketReceivedTimestamp => _connection.Statistics.LastNonKeepAlivePacketReceivedTimestamp;
public long ReceivedPacketsCount { get; set; }
public long ReceivedApplicationMessagesCount => _connection.Statistics.ReceivedApplicationMessagesCount;
public long SentPacketsCount { get; set; }
public long SentApplicationMessagesCount => _connection.Statistics.SentApplicationMessagesCount;
public long ReceivedPacketsCount => _connection.Statistics.ReceivedPacketsCount;
public long SentPacketsCount => _connection.Statistics.SentPacketsCount;
public IMqttSessionStatus Session { get; set; }
public long BytesSent { get; set; }
public long BytesSent => _connection.ChannelAdapter.BytesSent;
public long BytesReceived { get; set; }
public long BytesReceived => _connection.ChannelAdapter.BytesReceived;
public Task DisconnectAsync()
{
Source/MQTTnet/Server/Status/MqttSessionStatus.cs
+13 -13
View File
@@ -9,25 +9,25 @@ namespace MQTTnet.Server.Status
public sealed class MqttSessionStatus : IMqttSessionStatus
{
readonly MqttClientSession _session;
readonly MqttClientSessionsManager _sessionsManager;
public MqttSessionStatus(MqttClientSession session, MqttClientSessionsManager sessionsManager)
public MqttSessionStatus(MqttClientSession session)
{
_session = session ?? throw new ArgumentNullException(nameof(session));
_sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
session.Deleted += (_, __) => Deleted?.Invoke(this, EventArgs.Empty);
}
public event EventHandler Deleted;
public string ClientId { get; set; }
public event EventHandler Deleted
{
add => _session.Deleted += value;
remove => _session.Deleted += value;
}
public long PendingApplicationMessagesCount { get; set; }
public string ClientId => _session.ClientId;
public DateTime CreatedTimestamp { get; set; }
public IDictionary<object, object> Items { get; set; }
public long PendingApplicationMessagesCount => _session.ApplicationMessagesQueue.Count;
public DateTime CreatedTimestamp => _session.CreatedTimestamp;
public IDictionary<object, object> Items => _session.Items;
public Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage)
{
@@ -53,7 +53,7 @@ namespace MQTTnet.Server.Status
public Task DeleteAsync()
{
return _sessionsManager.DeleteSessionAsync(ClientId);
return _session.DeleteAsync();
}
public Task ClearPendingApplicationMessagesAsync()
Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs
+8 -9
View File
@@ -10,15 +10,12 @@ using MQTTnet.Exceptions;
using MQTTnet.LowLevelClient;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Tests.Mockups;
namespace MQTTnet.Tests.Client
{
[TestClass]
public class LowLevelMqttClient_Tests
public sealed class LowLevelMqttClient_Tests : BaseTestClass
{
public TestContext TestContext { get; set; }
[TestMethod]
[ExpectedException(typeof(MqttCommunicationException))]
public async Task Connect_To_Not_Existing_Server()
@@ -34,7 +31,7 @@ namespace MQTTnet.Tests.Client
[TestMethod]
public async Task Connect_And_Disconnect()
{
using (var testEnvironment = new TestEnvironment(TestContext))
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
@@ -50,7 +47,7 @@ namespace MQTTnet.Tests.Client
[TestMethod]
public async Task Authenticate()
{
using (var testEnvironment = new TestEnvironment(TestContext))
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
@@ -71,7 +68,7 @@ namespace MQTTnet.Tests.Client
[TestMethod]
public async Task Subscribe()
{
using (var testEnvironment = new TestEnvironment(TestContext))
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
@@ -94,21 +91,23 @@ namespace MQTTnet.Tests.Client
[TestMethod]
public async Task Loose_Connection()
{
using (var testEnvironment = new TestEnvironment(TestContext))
using (var testEnvironment = CreateTestEnvironment())
{
testEnvironment.ServerPort = 8364;
var server = await testEnvironment.StartServer();
var client = await testEnvironment.ConnectLowLevelClient(o => o.WithCommunicationTimeout(TimeSpan.Zero));
await Authenticate(client).ConfigureAwait(false);
await server.StopAsync();
await Task.Delay(1000);
await Task.Delay(2000);
try
{
await client.SendAsync(MqttPingReqPacket.Instance, CancellationToken.None).ConfigureAwait(false);
await Task.Delay(1000);
await client.SendAsync(MqttPingReqPacket.Instance, CancellationToken.None).ConfigureAwait(false);
}
catch (MqttCommunicationException exception)
Tests/MQTTnet.Core.Tests/Server/General.cs
+1 -1
View File
@@ -878,7 +878,7 @@ namespace MQTTnet.Tests.Server
// This should lead to no subscriptions for "n" at all. So also no sending of retained messages.
if (c.TopicFilter.Topic == "n")
{
c.AcceptSubscription = false;
c.ReasonCode = MqttSubscribeReasonCode.UnspecifiedError;
}
}));
Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs
+10 -4
View File
@@ -10,7 +10,7 @@ using MQTTnet.Tests.Mockups;
namespace MQTTnet.Tests.Server
{
[TestClass]
public class MqttSubscriptionsManager_Tests
public class MqttSubscriptionsManager_Tests : BaseTestClass
{
[TestMethod]
public async Task MqttSubscriptionsManager_SubscribeSingleSuccess()
@@ -102,12 +102,18 @@ namespace MQTTnet.Tests.Server
MqttClientSession CreateSession()
{
var logger = new TestLogger();
var options = new MqttServerOptions();
var retainedMessagesManager = new MqttRetainedMessagesManager();
var eventDispatcher = new MqttServerEventDispatcher(logger);
return new MqttClientSession(
"",
new ConcurrentDictionary<object, object>(),
new MqttServerEventDispatcher(new TestLogger()),
new MqttServerOptions(),
new MqttRetainedMessagesManager());
eventDispatcher,
options,
retainedMessagesManager,
new MqttClientSessionsManager(options, retainedMessagesManager, eventDispatcher, logger));
}
}
}
Tests/MQTTnet.TestApp.NetCore/ServerTest.cs
+2 -2
View File
@@ -73,12 +73,12 @@ namespace MQTTnet.TestApp.NetCore
{
if (context.TopicFilter.Topic.StartsWith("admin/foo/bar") && context.ClientId != "theAdmin")
{
context.AcceptSubscription = false;
context.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
}
if (context.TopicFilter.Topic.StartsWith("the/secret/stuff") && context.ClientId != "Imperator")
{
context.AcceptSubscription = false;
context.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
context.CloseConnection = true;
}
})