From 0ee3c8e135261567ac5755aa262c586039e4a3e9 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Tue, 28 Nov 2017 21:19:30 +0100 Subject: [PATCH] Add statistics of last sent packets to the _ConnectedClient_. --- .../Adapter/MqttChannelAdapter.cs | 4 +- .../MQTTnet.NetStandard/Client/MqttClient.cs | 58 ++++++---------- .../Diagnostics/MqttNetLogger.cs | 2 +- .../Implementations/MqttServerAdapter.cs | 39 +++++++++-- .../Server/ConnectedMqttClient.cs | 4 +- .../Server/MqttClientSession.cs | 69 ++++++++++++++----- .../Server/MqttClientSessionsManager.cs | 5 +- Tests/MQTTnet.TestApp.NetCore/Program.cs | 24 ++++++- 8 files changed, 137 insertions(+), 68 deletions(-) diff --git a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs index 2a9815d4..af399583 100644 --- a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs @@ -33,14 +33,14 @@ namespace MQTTnet.Adapter public Task ConnectAsync(TimeSpan timeout) { - _logger.Info("Connecting [Timeout={0}]", timeout); + _logger.Trace("Connecting [Timeout={0}]", timeout); return ExecuteAndWrapExceptionAsync(() => _channel.ConnectAsync().TimeoutAfter(timeout)); } public Task DisconnectAsync(TimeSpan timeout) { - _logger.Info("Disconnecting [Timeout={0}]", timeout); + _logger.Trace("Disconnecting [Timeout={0}]", timeout); return ExecuteAndWrapExceptionAsync(() => _channel.DisconnectAsync().TimeoutAfter(timeout)); } diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs index a24fbff9..b7658705 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs @@ -14,8 +14,6 @@ namespace MQTTnet.Client { public class MqttClient : IMqttClient { - private readonly HashSet _unacknowledgedPublishPackets = new HashSet(); - private readonly IMqttClientAdapterFactory _adapterFactory; private readonly MqttPacketDispatcher _packetDispatcher; private readonly IMqttNetLogger _logger; @@ -133,7 +131,7 @@ namespace MQTTnet.Client TopicFilters = topicFilters.ToList() }; - await SendAndReceiveAsync(unsubscribePacket); + await SendAndReceiveAsync(unsubscribePacket).ConfigureAwait(false); } public async Task PublishAsync(IEnumerable applicationMessages) @@ -141,21 +139,21 @@ namespace MQTTnet.Client ThrowIfNotConnected(); var publishPackets = applicationMessages.Select(m => m.ToPublishPacket()); + var packetGroups = publishPackets.GroupBy(p => p.QualityOfServiceLevel).OrderBy(g => g.Key); - foreach (var qosGroup in publishPackets.GroupBy(p => p.QualityOfServiceLevel)) + foreach (var qosGroup in packetGroups) { - var qosPackets = qosGroup.ToArray(); switch (qosGroup.Key) { case MqttQualityOfServiceLevel.AtMostOnce: { // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] - await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosPackets).ConfigureAwait(false); + await _adapter.SendPacketsAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token, qosGroup).ConfigureAwait(false); break; } case MqttQualityOfServiceLevel.AtLeastOnce: { - foreach (var publishPacket in qosPackets) + foreach (var publishPacket in qosGroup) { publishPacket.PacketIdentifier = GetNewPacketIdentifier(); await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); @@ -165,7 +163,7 @@ namespace MQTTnet.Client } case MqttQualityOfServiceLevel.ExactlyOnce: { - foreach (var publishPacket in qosPackets) + foreach (var publishPacket in qosGroup) { publishPacket.PacketIdentifier = GetNewPacketIdentifier(); var pubRecPacket = await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); @@ -249,28 +247,28 @@ namespace MQTTnet.Client try { _logger.Info("Received <<< {0}", packet); + + if (packet is MqttPublishPacket publishPacket) + { + await ProcessReceivedPublishPacketAsync(publishPacket).ConfigureAwait(false); + return; + } if (packet is MqttPingReqPacket) { - await SendAsync(new MqttPingRespPacket()); + await SendAsync(new MqttPingRespPacket()).ConfigureAwait(false); return; } if (packet is MqttDisconnectPacket) { - await DisconnectAsync(); - return; - } - - if (packet is MqttPublishPacket publishPacket) - { - await ProcessReceivedPublishPacket(publishPacket); + await DisconnectAsync().ConfigureAwait(false); return; } if (packet is MqttPubRelPacket pubRelPacket) { - await ProcessReceivedPubRelPacket(pubRelPacket); + await ProcessReceivedPubRelPacket(pubRelPacket).ConfigureAwait(false); return; } @@ -295,32 +293,25 @@ namespace MQTTnet.Client } } - private async Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket) + private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket) { if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) { FireApplicationMessageReceivedEvent(publishPacket); - return; + return Task.FromResult(0); } if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) { FireApplicationMessageReceivedEvent(publishPacket); - await SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }); - return; + return SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }); } if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) { // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery] - lock (_unacknowledgedPublishPackets) - { - _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier); - } - FireApplicationMessageReceivedEvent(publishPacket); - await SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }); - return; + return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }); } throw new MqttCommunicationException("Received a not supported QoS level."); @@ -328,11 +319,6 @@ namespace MQTTnet.Client private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket) { - lock (_unacknowledgedPublishPackets) - { - _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier); - } - return SendAsync(pubRelPacket.CreateResponse()); } @@ -454,7 +440,7 @@ namespace MQTTnet.Client { #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed Task.Run( - () => ProcessReceivedPacketAsync(packet), + async () => await ProcessReceivedPacketAsync(packet).ConfigureAwait(false), cancellationToken).ConfigureAwait(false); #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed } @@ -465,7 +451,7 @@ namespace MQTTnet.Client #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed Task.Run( - async () => await ReceivePacketsAsync(cancellationToken), + async () => await ReceivePacketsAsync(cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false); #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed @@ -479,7 +465,7 @@ namespace MQTTnet.Client { #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed Task.Run( - async () => await SendKeepAliveMessagesAsync(cancellationToken), + async () => await SendKeepAliveMessagesAsync(cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false); #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed } diff --git a/Frameworks/MQTTnet.NetStandard/Diagnostics/MqttNetLogger.cs b/Frameworks/MQTTnet.NetStandard/Diagnostics/MqttNetLogger.cs index 7d9b2907..8ad1d4c4 100644 --- a/Frameworks/MQTTnet.NetStandard/Diagnostics/MqttNetLogger.cs +++ b/Frameworks/MQTTnet.NetStandard/Diagnostics/MqttNetLogger.cs @@ -10,7 +10,7 @@ namespace MQTTnet.Diagnostics { _logId = logId; } - + public event EventHandler LogMessagePublished; public void Trace(string message, params object[] parameters) diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs index d4ab0950..a0441bcc 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs @@ -42,7 +42,7 @@ namespace MQTTnet.Implementations _defaultEndpointSocket.Bind(new IPEndPoint(IPAddress.Any, options.GetDefaultEndpointPort())); _defaultEndpointSocket.Listen(options.ConnectionBacklog); - Task.Run(() => AcceptDefaultEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token); + Task.Run(async () => await AcceptDefaultEndpointConnectionsAsync(_cancellationTokenSource.Token).ConfigureAwait(false), _cancellationTokenSource.Token).ConfigureAwait(false); } if (options.TlsEndpointOptions.IsEnabled) @@ -62,7 +62,7 @@ namespace MQTTnet.Implementations _tlsEndpointSocket.Bind(new IPEndPoint(IPAddress.Any, options.GetTlsEndpointPort())); _tlsEndpointSocket.Listen(options.ConnectionBacklog); - Task.Run(() => AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token); + Task.Run(async () => await AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token).ConfigureAwait(false), _cancellationTokenSource.Token).ConfigureAwait(false); } return Task.FromResult(0); @@ -102,13 +102,29 @@ namespace MQTTnet.Implementations #else var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false); #endif + var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer(), _logger); ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); } + catch (ObjectDisposedException) + { + // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed. + } + catch (SocketException exception) + { + if (exception.SocketErrorCode == SocketError.OperationAborted) + { + return; + } + + _logger.Error(exception, "Error while accepting connection at default endpoint."); + } catch (Exception exception) { _logger.Error(exception, "Error while accepting connection at default endpoint."); - + } + finally + { //excessive CPU consumed if in endless loop of socket errors await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false); } @@ -133,10 +149,25 @@ namespace MQTTnet.Implementations var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _logger); ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); } + catch (ObjectDisposedException) + { + // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed. + } + catch (SocketException exception) + { + if (exception.SocketErrorCode == SocketError.OperationAborted) + { + return; + } + + _logger.Error(exception, "Error while accepting connection at default endpoint."); + } catch (Exception exception) { _logger.Error(exception, "Error while accepting connection at TLS endpoint."); - + } + finally + { //excessive CPU consumed if in endless loop of socket errors await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false); } diff --git a/Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs b/Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs index ee3a55fb..cd941930 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs @@ -9,8 +9,8 @@ namespace MQTTnet.Server public MqttProtocolVersion ProtocolVersion { get; set; } - public TimeSpan LastPacketReceivedDuration { get; set; } + public TimeSpan LastPacketReceived { get; set; } - public TimeSpan LastNonKeepAlivePacketReceivedDuration{ get; set; } + public TimeSpan LastNonKeepAlivePacketReceived { get; set; } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs index c5f16a9e..12ef891a 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs @@ -1,4 +1,5 @@ using System; +using System.Diagnostics; using System.Threading; using System.Threading.Tasks; using MQTTnet.Adapter; @@ -13,6 +14,9 @@ namespace MQTTnet.Server { public sealed class MqttClientSession { + private readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch(); + private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = new Stopwatch(); + private readonly MqttClientSubscriptionsManager _subscriptionsManager; private readonly MqttClientSessionsManager _sessionsManager; private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue; @@ -44,6 +48,10 @@ namespace MQTTnet.Server public MqttProtocolVersion? ProtocolVersion => _adapter?.PacketSerializer.ProtocolVersion; + public TimeSpan LastPacketReceived => _lastPacketReceivedTracker.Elapsed; + + public TimeSpan LastNonKeepAlivePacketReceived => _lastNonKeepAlivePacketReceivedTracker.Elapsed; + public bool IsConnected => _adapter != null; public async Task RunAsync(MqttApplicationMessage willMessage, IMqttChannelAdapter adapter) @@ -52,6 +60,9 @@ namespace MQTTnet.Server try { + _lastPacketReceivedTracker.Restart(); + _lastNonKeepAlivePacketReceivedTracker.Restart(); + var cancellationTokenSource = new CancellationTokenSource(); _willMessage = willMessage; @@ -78,6 +89,11 @@ namespace MQTTnet.Server { try { + if (_cancellationTokenSource == null) + { + return; + } + _cancellationTokenSource?.Cancel(false); _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; @@ -143,6 +159,13 @@ namespace MQTTnet.Server private Task ProcessReceivedPacketAsync(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken) { + _lastPacketReceivedTracker.Restart(); + + if (!(packet is MqttPingReqPacket)) + { + _lastNonKeepAlivePacketReceivedTracker.Restart(); + } + if (packet is MqttPublishPacket publishPacket) { return HandleIncomingPublishPacketAsync(adapter, publishPacket, cancellationToken); @@ -185,7 +208,7 @@ namespace MQTTnet.Server } _logger.Warning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); - return StopAsync(); + return StopAsync(); } private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) @@ -211,7 +234,7 @@ namespace MQTTnet.Server } } - private async Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken) + private Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken) { var applicationMessage = publishPacket.ToApplicationMessage(); @@ -219,36 +242,44 @@ namespace MQTTnet.Server { case MqttQualityOfServiceLevel.AtMostOnce: { - await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false); - return; + return _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage); } case MqttQualityOfServiceLevel.AtLeastOnce: { - await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false); - - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, - new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false); - - return; + return HandleIncomingPublishPacketWithQoS1(adapter, applicationMessage, publishPacket, cancellationToken); } case MqttQualityOfServiceLevel.ExactlyOnce: { - // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery] - await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false); - - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, - new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false); - - return; + return HandleIncomingPublishPacketWithQoS2(adapter, applicationMessage, publishPacket, cancellationToken); } default: - throw new MqttCommunicationException("Received a not supported QoS level."); + { + throw new MqttCommunicationException("Received a not supported QoS level."); + } } } + private async Task HandleIncomingPublishPacketWithQoS1(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken) + { + await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false); + + var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }; + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false); + } + + private async Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken) + { + // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery] + await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false); + + var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }; + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false); + } + private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }); + var response = new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }; + return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response); } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index f65ed2ba..5bb65984 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -120,11 +120,12 @@ namespace MQTTnet.Server await _sessionsSemaphore.WaitAsync().ConfigureAwait(false); try { - var now = DateTime.UtcNow; return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient { ClientId = s.Value.ClientId, - ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311 + ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311, + LastPacketReceived = s.Value.LastPacketReceived, + LastNonKeepAlivePacketReceived = s.Value.LastNonKeepAlivePacketReceived }).ToList(); } finally diff --git a/Tests/MQTTnet.TestApp.NetCore/Program.cs b/Tests/MQTTnet.TestApp.NetCore/Program.cs index 0e4868af..77b9ea6e 100644 --- a/Tests/MQTTnet.TestApp.NetCore/Program.cs +++ b/Tests/MQTTnet.TestApp.NetCore/Program.cs @@ -40,7 +40,7 @@ namespace MQTTnet.TestApp.NetCore Thread.Sleep(Timeout.Infinite); } - + // This code is used at the Wiki on GitHub! // ReSharper disable once UnusedMember.Local private static async void WikiCode() @@ -71,6 +71,26 @@ namespace MQTTnet.TestApp.NetCore var factory = new MqttFactory(); var client = factory.CreateMqttClient(); } + + { + // Write all trace messages to the console window. + MqttNetGlobalLogger.LogMessagePublished += (s, e) => + { + var trace = $">> [{e.TraceMessage.Timestamp:O}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Source}] [{e.TraceMessage.Level}]: {e.TraceMessage.Message}"; + if (e.TraceMessage.Exception != null) + { + trace += Environment.NewLine + e.TraceMessage.Exception.ToString(); + } + + Console.WriteLine(trace); + }; + } + + { + // Use a custom log ID for the logger. + var factory = new MqttFactory(); + var mqttClient = factory.CreateMqttClient(new MqttNetLogger("MyCustomId")); + } } } @@ -106,4 +126,4 @@ namespace MQTTnet.TestApp.NetCore return Task.FromResult(retainedMessages); } } -} +} \ No newline at end of file