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

Update docs, fix managed client connection issue

Browse Source
This commit is contained in:
Christian Kratky
2017-11-26 22:57:56 +01:00
parent 611fabcd01
commit 69781d822a
8 changed files with 140 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Build/MQTTnet.nuspec
+8 -3
View File
@@ -10,9 +10,14 @@
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<requireLicenseAcceptance>false</requireLicenseAcceptance>
<description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
<releaseNotes>* Merged projects.
* Added a strong name for the assembly.
<releaseNotes>* [Core] Merged projects (BREAKING CHANGE! But only namespace changes).
* [Core] Added a strong name for the assembly.
* [Core] Performance optimizations.
* [Core] Fixed a logging issue when dealing with IOExceptions.
* [Client] Fixed an issue in _ManagedClient_ which can cause the client to stop when publishing subscriptions.
* [Server] The application message interceptor can now delete any received application message.
* [Server] Added a ConnectionValidator context to align with other APIs.
* [Server] Added an interface for the _MqttServerOptions_.
</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2017</copyright>
<tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags>
Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs
+14 -1
View File
@@ -1,6 +1,7 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Net.Sockets;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
@@ -171,13 +172,25 @@ namespace MQTTnet.Adapter
}
catch (COMException comException)
{
if ((uint)comException.HResult == ErrorOperationAborted)
if ((uint) comException.HResult == ErrorOperationAborted)
{
throw new OperationCanceledException();
}
throw new MqttCommunicationException(comException);
}
catch (IOException exception)
{
if (exception.InnerException is SocketException socketException)
{
if (socketException.SocketErrorCode == SocketError.ConnectionAborted)
{
throw new OperationCanceledException();
}
}
throw new MqttCommunicationException(exception);
}
catch (Exception exception)
{
throw new MqttCommunicationException(exception);
Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClient.cs
+46 -27
View File
@@ -142,33 +142,53 @@ namespace MQTTnet.ManagedClient
{
while (!cancellationToken.IsCancellationRequested)
{
var connectionState = await ReconnectIfRequiredAsync().ConfigureAwait(false);
if (connectionState == ReconnectionResult.NotConnected)
{
_publishingCancellationToken?.Cancel(false);
_publishingCancellationToken = null;
await TryMaintainConnectionAsync(cancellationToken);
}
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
_logger.Error<ManagedMqttClient>(exception, "Unhandled exception while maintaining connection.");
}
finally
{
await _mqttClient.DisconnectAsync().ConfigureAwait(false);
_logger.Info<ManagedMqttClient>("Stopped");
}
}
await Task.Delay(_options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
continue;
}
private async Task TryMaintainConnectionAsync(CancellationToken cancellationToken)
{
try
{
var connectionState = await ReconnectIfRequiredAsync().ConfigureAwait(false);
if (connectionState == ReconnectionResult.NotConnected)
{
_publishingCancellationToken?.Cancel(false);
_publishingCancellationToken = null;
if (connectionState == ReconnectionResult.Reconnected || _subscriptionsNotPushed)
{
await PushSubscriptionsAsync();
await Task.Delay(_options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
return;
}
_publishingCancellationToken = new CancellationTokenSource();
if (connectionState == ReconnectionResult.Reconnected || _subscriptionsNotPushed)
{
await PushSubscriptionsAsync().ConfigureAwait(false);
_publishingCancellationToken = new CancellationTokenSource();
#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 PublishQueuedMessagesAsync(_publishingCancellationToken.Token), _publishingCancellationToken.Token).ConfigureAwait(false);
Task.Run(async () => await PublishQueuedMessagesAsync(_publishingCancellationToken.Token).ConfigureAwait(false), _publishingCancellationToken.Token).ConfigureAwait(false);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
continue;
}
return;
}
if (connectionState == ReconnectionResult.StillConnected)
{
await Task.Delay(TimeSpan.FromSeconds(1), _connectionCancellationToken.Token).ConfigureAwait(false);
}
if (connectionState == ReconnectionResult.StillConnected)
{
await Task.Delay(TimeSpan.FromSeconds(1), _connectionCancellationToken.Token).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
@@ -182,11 +202,6 @@ namespace MQTTnet.ManagedClient
{
_logger.Error<ManagedMqttClient>(exception, "Unhandled exception while maintaining connection.");
}
finally
{
await _mqttClient.DisconnectAsync().ConfigureAwait(false);
_logger.Info<ManagedMqttClient>("Stopped");
}
}
private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
@@ -206,16 +221,19 @@ namespace MQTTnet.ManagedClient
continue;
}
await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
await _storageManager.RemoveAsync(message).ConfigureAwait(false);
await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
_logger.Error<ManagedMqttClient>(exception, "Unhandled exception while publishing queued application messages.");
}
finally
{
_logger.Info<ManagedMqttClient>("Stopped publishing messages");
_logger.Trace<ManagedMqttClient>("Stopped publishing messages.");
}
}
@@ -224,6 +242,7 @@ namespace MQTTnet.ManagedClient
try
{
await _mqttClient.PublishAsync(message).ConfigureAwait(false);
await _storageManager.RemoveAsync(message).ConfigureAwait(false);
}
catch (MqttCommunicationException exception)
{
Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs
+6 -1
View File
@@ -1,4 +1,5 @@
using MQTTnet.Serializer;
using System;
using MQTTnet.Serializer;
namespace MQTTnet.Server
{
@@ -7,5 +8,9 @@ namespace MQTTnet.Server
public string ClientId { get; set; }
public MqttProtocolVersion ProtocolVersion { get; set; }
public TimeSpan LastPacketReceivedDuration { get; set; }
public TimeSpan LastNonKeepAlivePacketReceivedDuration{ get; set; }
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs
+14 -5
View File
@@ -120,6 +120,7 @@ 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,
@@ -136,13 +137,21 @@ namespace MQTTnet.Server
{
try
{
var interceptorContext = new MqttApplicationMessageInterceptorContext
if (_options.ApplicationMessageInterceptor != null)
{
ApplicationMessage = applicationMessage
};
var interceptorContext = new MqttApplicationMessageInterceptorContext
{
ApplicationMessage = applicationMessage
};
_options.ApplicationMessageInterceptor?.Invoke(interceptorContext);
applicationMessage = interceptorContext.ApplicationMessage;
_options.ApplicationMessageInterceptor(interceptorContext);
applicationMessage = interceptorContext.ApplicationMessage;
}
if (applicationMessage == null)
{
return;
}
if (applicationMessage.Retain)
{
Tests/MQTTnet.TestApp.NetCore/MqttNetConsoleLogger.cs
+27 -20
View File
@@ -14,6 +14,17 @@ namespace MQTTnet.TestApp.NetCore
MqttNetGlobalLogger.LogMessagePublished += PrintToConsole;
}
public static void PrintToConsole(string message, ConsoleColor color)
{
lock (Lock)
{
var backupColor = Console.ForegroundColor;
Console.ForegroundColor = color;
Console.Write(message);
Console.ForegroundColor = backupColor;
}
}
private static void PrintToConsole(object sender, MqttNetLogMessagePublishedEventArgs e)
{
var output = new StringBuilder();
@@ -23,28 +34,24 @@ namespace MQTTnet.TestApp.NetCore
output.AppendLine(e.TraceMessage.Exception.ToString());
}
lock (Lock)
var color = ConsoleColor.Red;
switch (e.TraceMessage.Level)
{
var backupColor = Console.ForegroundColor;
switch (e.TraceMessage.Level)
{
case MqttNetLogLevel.Error:
Console.ForegroundColor = ConsoleColor.Red;
break;
case MqttNetLogLevel.Warning:
Console.ForegroundColor = ConsoleColor.Yellow;
break;
case MqttNetLogLevel.Info:
Console.ForegroundColor = ConsoleColor.Green;
break;
case MqttNetLogLevel.Verbose:
Console.ForegroundColor = ConsoleColor.Gray;
break;
}
Console.Write(output);
Console.ForegroundColor = backupColor;
case MqttNetLogLevel.Error:
color = ConsoleColor.Red;
break;
case MqttNetLogLevel.Warning:
color = ConsoleColor.Yellow;
break;
case MqttNetLogLevel.Info:
color = ConsoleColor.Green;
break;
case MqttNetLogLevel.Verbose:
color = ConsoleColor.Gray;
break;
}
PrintToConsole(output.ToString(), color);
}
}
}
Tests/MQTTnet.TestApp.NetCore/ServerTest.cs
+20 -1
View File
@@ -1,9 +1,9 @@
using System;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Protocol;
using MQTTnet.Server;
using Newtonsoft.Json.Linq;
namespace MQTTnet.TestApp.NetCore
{
@@ -63,6 +63,25 @@ namespace MQTTnet.TestApp.NetCore
//options.TlsEndpointOptions.IsEnabled = false;
var mqttServer = new MqttFactory().CreateMqttServer();
mqttServer.ApplicationMessageReceived += (s, e) =>
{
MqttNetConsoleLogger.PrintToConsole(
$"'{e.ClientId}' reported '{e.ApplicationMessage.Topic}' > '{Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}'",
ConsoleColor.Magenta);
};
options.ApplicationMessageInterceptor = c =>
{
var content = JObject.Parse(Encoding.UTF8.GetString(c.ApplicationMessage.Payload));
var timestampProperty = content.Property("timestamp");
if (timestampProperty != null && timestampProperty.Value.Type == JTokenType.Null)
{
timestampProperty.Value = DateTime.Now.ToString("O");
c.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(content.ToString());
}
};
mqttServer.ClientDisconnected += (s, e) =>
{
Console.Write("Client disconnected event fired.");
Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs
+5 -3
View File
@@ -407,12 +407,14 @@ namespace MQTTnet.TestApp.UniversalWindows
{
// Configure MQTT server.
var optionsBuilder = new MqttServerOptionsBuilder()
.WithConnectionBacklog(100)
.WithDefaultEndpointPort(1884);
var options = new MqttServerOptions
{
ConnectionBacklog = 100
};
options.DefaultEndpointOptions.Port = 1884;
options.ConnectionValidator = c =>
{
if (c.ClientId != "Highlander")
@@ -425,7 +427,7 @@ namespace MQTTnet.TestApp.UniversalWindows
};
var mqttServer = new MqttFactory().CreateMqttServer();
await mqttServer.StartAsync(options);
await mqttServer.StartAsync(optionsBuilder.Build());
}
{