From a15fbf532e03a51040aeb3566873a974752c8aba Mon Sep 17 00:00:00 2001 From: Christian Date: Thu, 14 Oct 2021 21:50:16 +0200 Subject: [PATCH] Add session to subscribe interceptor. --- MQTTnet.sln.DotSettings | 1 + .../Server/Internal/MqttClientConnection.cs | 60 +++++++------------ .../MqttClientConnectionStatistics.cs | 54 +++++++---------- .../Server/Internal/MqttClientSession.cs | 31 +++++----- .../Internal/MqttClientSessionsManager.cs | 18 +++--- .../MqttClientSubscriptionsManager.cs | 8 ++- .../Internal/MqttServerKeepAliveMonitor.cs | 2 +- .../MqttPendingMessagesOverflowStrategy.cs | 1 + .../MqttSubscriptionInterceptorContext.cs | 6 ++ .../Server/Status/IMqttClientStatus.cs | 4 +- .../MQTTnet/Server/Status/MqttClientStatus.cs | 30 +++++----- .../Server/Status/MqttSessionStatus.cs | 26 ++++---- .../Client/LowLevelMqttClient_Tests.cs | 17 +++--- Tests/MQTTnet.Core.Tests/Server/General.cs | 2 +- .../Server/MqttSubscriptionsManager_Tests.cs | 14 +++-- Tests/MQTTnet.TestApp.NetCore/ServerTest.cs | 4 +- 16 files changed, 130 insertions(+), 148 deletions(-) diff --git a/MQTTnet.sln.DotSettings b/MQTTnet.sln.DotSettings index e81ffe21..bcb734e5 100644 --- a/MQTTnet.sln.DotSettings +++ b/MQTTnet.sln.DotSettings @@ -3,6 +3,7 @@ False False False + 180 True True True diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs index 9fe11a66..3362f66f 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs @@ -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 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)) { diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnectionStatistics.cs b/Source/MQTTnet/Server/Internal/MqttClientConnectionStatistics.cs index 14923f1b..c23c6f06 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnectionStatistics.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientConnectionStatistics.cs @@ -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; - } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientSession.cs b/Source/MQTTnet/Server/Internal/MqttClientSession.cs index e03d4908..b37099bb 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSession.cs @@ -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 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; } - /// - /// Gets or sets a key/value collection that can be used to share data within the scope of this session. - /// public IDictionary 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); diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index d47ad9e1..b296a385 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -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); } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 3179abde..abf8d725 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -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 _subscriptions = new Dictionary(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 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) diff --git a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs index abcc1201..72f50056 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs @@ -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". diff --git a/Source/MQTTnet/Server/MqttPendingMessagesOverflowStrategy.cs b/Source/MQTTnet/Server/MqttPendingMessagesOverflowStrategy.cs index 1601487e..05ca9ad5 100644 --- a/Source/MQTTnet/Server/MqttPendingMessagesOverflowStrategy.cs +++ b/Source/MQTTnet/Server/MqttPendingMessagesOverflowStrategy.cs @@ -3,6 +3,7 @@ public enum MqttPendingMessagesOverflowStrategy { DropOldestQueuedMessage, + DropNewMessage } } diff --git a/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs b/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs index 5b41285b..3ce23b1a 100644 --- a/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs +++ b/Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs @@ -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. /// public bool CloseConnection { get; set; } + + /// + /// Gets the current client session. + /// + public IMqttSessionStatus Session { get; internal set; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Status/IMqttClientStatus.cs b/Source/MQTTnet/Server/Status/IMqttClientStatus.cs index 29fb2d99..268d675d 100644 --- a/Source/MQTTnet/Server/Status/IMqttClientStatus.cs +++ b/Source/MQTTnet/Server/Status/IMqttClientStatus.cs @@ -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; } diff --git a/Source/MQTTnet/Server/Status/MqttClientStatus.cs b/Source/MQTTnet/Server/Status/MqttClientStatus.cs index 1770cde9..ca729de0 100644 --- a/Source/MQTTnet/Server/Status/MqttClientStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttClientStatus.cs @@ -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. /// - 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() { diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index 99be2101..9fdeee05 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -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 Items { get; set; } + public long PendingApplicationMessagesCount => _session.ApplicationMessagesQueue.Count; + + public DateTime CreatedTimestamp => _session.CreatedTimestamp; + + public IDictionary 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() diff --git a/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs index 814adcaa..79d1855a 100644 --- a/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs @@ -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) diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index cda0340f..28e8e1a0 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -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; } })); diff --git a/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs index 4ef18b4d..b2272770 100644 --- a/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs @@ -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(), - new MqttServerEventDispatcher(new TestLogger()), - new MqttServerOptions(), - new MqttRetainedMessagesManager()); + eventDispatcher, + options, + retainedMessagesManager, + new MqttClientSessionsManager(options, retainedMessagesManager, eventDispatcher, logger)); } } } diff --git a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs index 96841724..00951789 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs @@ -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; } })