Compare commits

...
24 changed files with 305 additions and 175 deletions
+10 -1
View File
@@ -1,4 +1,13 @@
* [Core] Add validation of maximum string lengths (#1718). * [Core] Add validation of maximum string lengths (#1718).
* [Core] Added .NET 4.8 builds (#1729).
* [Core] Exposed more details of DISCONNECT packet in log (#1729).
* [Core] Fixed _ArgumentNullException_ in _ConvertPayloadToString_ (#1745, thanks to @ramonsmits).
* [Client] Added overloads for setting packet payload and will payload (#1720). * [Client] Added overloads for setting packet payload and will payload (#1720).
* [Client] The proper connect result is now exposed in the _Disconnected_ event when authentication fails (#1139). * [Client] The proper connect result is now exposed in the _Disconnected_ event when authentication fails (#1139).
* [Client] Exposed _UseDefaultCredentials_ and more for Web Socket options and proxy options (#1734, thanks to @impworks).
* [Client] Exposed more TLS options (#1729).
* [Client] Fixed wrong return code conversion (#1729).
* [Client] Added an option to avoid throwing an exception when the server returned a proper non success (but valid) response (#1681).
* [Server] Improved performance by changing internal locking strategy for subscriptions (#1716, thanks to @zeheng). * [Server] Improved performance by changing internal locking strategy for subscriptions (#1716, thanks to @zeheng).
* [Server] Fixed exceptions when clients are connecting and disconnecting very fast while accessing the client status for connection validation (#1742).
* [Server] Exposed more properties in _ClientConnectedEventArgs_ (#1738).
@@ -2,7 +2,7 @@
<PropertyGroup> <PropertyGroup>
<TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks> <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks> <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461;net48</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks> <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
<AssemblyName>MQTTnet.Extensions.ManagedClient</AssemblyName> <AssemblyName>MQTTnet.Extensions.ManagedClient</AssemblyName>
@@ -2,7 +2,7 @@
<PropertyGroup> <PropertyGroup>
<TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks> <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks> <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461;net48</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks> <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
<AssemblyName>MQTTnet.Extensions.Rpc</AssemblyName> <AssemblyName>MQTTnet.Extensions.Rpc</AssemblyName>
@@ -2,7 +2,7 @@
<PropertyGroup> <PropertyGroup>
<TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks> <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks> <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461;net48</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks> <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
<AssemblyName>MQTTnet.Extensions.WebSocket4Net</AssemblyName> <AssemblyName>MQTTnet.Extensions.WebSocket4Net</AssemblyName>
@@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System; using System;
using System.Collections.Generic;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting; using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -10,6 +11,7 @@ using MQTTnet.Client;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server; using MQTTnet.Server;
@@ -128,7 +130,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
await client.DisconnectAsync(disconnectOptions); await client.DisconnectAsync(disconnectOptions);
await LongTestDelay(); await LongTestDelay();
Assert.IsNotNull(eventArgs); Assert.IsNotNull(eventArgs);
Assert.AreEqual(MqttDisconnectReasonCode.MessageRateTooHigh, eventArgs.ReasonCode); Assert.AreEqual(MqttDisconnectReasonCode.MessageRateTooHigh, eventArgs.ReasonCode);
} }
@@ -156,7 +158,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
await client.DisconnectAsync(disconnectOptions); await client.DisconnectAsync(disconnectOptions);
await LongTestDelay(); await LongTestDelay();
Assert.IsNotNull(eventArgs); Assert.IsNotNull(eventArgs);
Assert.IsNotNull(eventArgs.UserProperties); Assert.IsNotNull(eventArgs.UserProperties);
Assert.AreEqual(1, eventArgs.UserProperties.Count); Assert.AreEqual(1, eventArgs.UserProperties.Count);
@@ -164,5 +166,35 @@ namespace MQTTnet.Tests.Clients.MqttClient
Assert.AreEqual("test_value", eventArgs.UserProperties[0].Value); Assert.AreEqual("test_value", eventArgs.UserProperties[0].Value);
} }
} }
[TestMethod]
public async Task Return_Non_Success()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
var server = await testEnvironment.StartServer();
server.ValidatingConnectionAsync += args =>
{
args.ResponseUserProperties = new List<MqttUserProperty>
{
new MqttUserProperty("Property", "Value")
};
args.ReasonCode = MqttConnectReasonCode.QuotaExceeded;
return CompletedTask.Instance;
};
var client = testEnvironment.CreateClient();
var response = await client.ConnectAsync(testEnvironment.CreateDefaultClientOptionsBuilder().WithoutThrowOnNonSuccessfulConnectResponse().Build());
Assert.IsNotNull(response);
Assert.AreEqual(MqttClientConnectResultCode.QuotaExceeded, response.ResultCode);
Assert.AreEqual(response.UserProperties[0].Name, "Property");
Assert.AreEqual(response.UserProperties[0].Value, "Value");
}
}
} }
} }
@@ -78,7 +78,7 @@ namespace MQTTnet.Tests.Formatter
Assert.IsNull(deserialized.UserProperties); // Not supported in v3.1.1 Assert.IsNull(deserialized.UserProperties); // Not supported in v3.1.1
} }
[TestMethod] [TestMethod]
public void Serialize_Full_MqttConnAckPacket_V310() public void Serialize_Full_MqttConnAckPacket_V310()
{ {
var connAckPacket = new MqttConnAckPacket var connAckPacket = new MqttConnAckPacket
@@ -6,6 +6,7 @@ namespace MQTTnet.Client.Internal
{ {
public static class MqttClientResultFactory public static class MqttClientResultFactory
{ {
public static readonly MqttClientConnectResultFactory ConnectResult = new MqttClientConnectResultFactory();
public static readonly MqttClientPublishResultFactory PublishResult = new MqttClientPublishResultFactory(); public static readonly MqttClientPublishResultFactory PublishResult = new MqttClientPublishResultFactory();
public static readonly MqttClientSubscribeResultFactory SubscribeResult = new MqttClientSubscribeResultFactory(); public static readonly MqttClientSubscribeResultFactory SubscribeResult = new MqttClientSubscribeResultFactory();
public static readonly MqttClientUnsubscribeResultFactory UnsubscribeResult = new MqttClientUnsubscribeResultFactory(); public static readonly MqttClientUnsubscribeResultFactory UnsubscribeResult = new MqttClientUnsubscribeResultFactory();
+88 -72
View File
@@ -121,11 +121,11 @@ namespace MQTTnet.Client
var mqttClientAliveToken = _mqttClientAlive.Token; var mqttClientAliveToken = _mqttClientAlive.Token;
var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_events.InspectPacketEvent, _rootLogger), _rootLogger); var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_events.InspectPacketEvent, _rootLogger), _rootLogger);
_adapter = adapter; _adapter = adapter ?? throw new InvalidOperationException("The adapter factory did not provide an adapter.");
if (cancellationToken.CanBeCanceled) if (cancellationToken.CanBeCanceled)
{ {
connectResult = await ConnectInternal(cancellationToken).ConfigureAwait(false); connectResult = await ConnectInternal(adapter, cancellationToken).ConfigureAwait(false);
} }
else else
{ {
@@ -133,10 +133,16 @@ namespace MQTTnet.Client
// CancellationToken.None or similar. // CancellationToken.None or similar.
using (var timeout = new CancellationTokenSource(Options.Timeout)) using (var timeout = new CancellationTokenSource(Options.Timeout))
{ {
connectResult = await ConnectInternal(timeout.Token).ConfigureAwait(false); connectResult = await ConnectInternal(adapter, timeout.Token).ConfigureAwait(false);
} }
} }
if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
{
_logger.Warning("Connecting failed: {0}", connectResult.ResultCode);
return connectResult;
}
_lastPacketSentTimestamp = DateTime.UtcNow; _lastPacketSentTimestamp = DateTime.UtcNow;
var keepAliveInterval = Options.KeepAlivePeriod; var keepAliveInterval = Options.KeepAlivePeriod;
@@ -421,7 +427,7 @@ namespace MQTTnet.Client
return CompletedTask.Instance; return CompletedTask.Instance;
} }
async Task<MqttClientConnectResult> Authenticate(MqttClientOptions options, CancellationToken cancellationToken) async Task<MqttClientConnectResult> Authenticate(IMqttChannelAdapter channelAdapter, MqttClientOptions options, CancellationToken cancellationToken)
{ {
MqttClientConnectResult result; MqttClientConnectResult result;
@@ -434,8 +440,7 @@ namespace MQTTnet.Client
if (receivedPacket is MqttConnAckPacket connAckPacket) if (receivedPacket is MqttConnAckPacket connAckPacket)
{ {
var clientConnectResultFactory = new MqttClientConnectResultFactory(); result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
result = clientConnectResultFactory.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
} }
else else
{ {
@@ -447,9 +452,18 @@ namespace MQTTnet.Client
throw new MqttConnectingFailedException($"Error while authenticating. {exception.Message}", exception, null); throw new MqttConnectingFailedException($"Error while authenticating. {exception.Message}", exception, null);
} }
if (result.ResultCode != MqttClientConnectResultCode.Success) // This is no feature. It is basically a backward compatibility option and should be removed in the future.
// The client should not throw any exception if the transport layer connection was successful and the server
// did send a proper ACK packet with a non success response.
if (options.ThrowOnNonSuccessfulConnectResponse)
{ {
throw new MqttConnectingFailedException($"Connecting with MQTT server failed ({result.ResultCode}).", null, result); _logger.Warning(
"Client will now throw an _MqttConnectingFailedException_. This is obsolete and will be removed in the future. Consider setting _ThrowOnNonSuccessfulResponseFromServer=False_ in client options.");
if (result.ResultCode != MqttClientConnectResultCode.Success)
{
throw new MqttConnectingFailedException($"Connecting with MQTT server failed ({result.ResultCode}).", null, result);
}
} }
_logger.Verbose("Authenticated MQTT connection with server established."); _logger.Verbose("Authenticated MQTT connection with server established.");
@@ -484,7 +498,7 @@ namespace MQTTnet.Client
return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand); return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand);
} }
async Task<MqttClientConnectResult> ConnectInternal(CancellationToken cancellationToken) async Task<MqttClientConnectResult> ConnectInternal(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
{ {
var backgroundCancellationToken = _mqttClientAlive.Token; var backgroundCancellationToken = _mqttClientAlive.Token;
@@ -497,10 +511,12 @@ namespace MQTTnet.Client
_publishPacketReceiverQueue?.Dispose(); _publishPacketReceiverQueue?.Dispose();
_publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>(); _publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>();
var connectResult = await Authenticate(Options, effectiveCancellationToken.Token).ConfigureAwait(false); var connectResult = await Authenticate(channelAdapter, Options, effectiveCancellationToken.Token).ConfigureAwait(false);
if (connectResult.ResultCode == MqttClientConnectResultCode.Success)
_publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken); {
_packetReceiverTask = Task.Run(() => ReceivePacketsLoop(backgroundCancellationToken), backgroundCancellationToken); _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken);
_packetReceiverTask = Task.Run(() => ReceivePacketsLoop(backgroundCancellationToken), backgroundCancellationToken);
}
return connectResult; return connectResult;
} }
@@ -754,6 +770,65 @@ namespace MQTTnet.Client
return packet; return packet;
} }
async Task ReceivePacketsLoop(CancellationToken cancellationToken)
{
try
{
_logger.Verbose("Start receiving packets.");
while (!cancellationToken.IsCancellationRequested)
{
var packet = await Receive(cancellationToken).ConfigureAwait(false);
if (cancellationToken.IsCancellationRequested)
{
return;
}
if (packet == null)
{
await DisconnectInternal(_packetReceiverTask, null, null).ConfigureAwait(false);
return;
}
await TryProcessReceivedPacket(packet, cancellationToken).ConfigureAwait(false);
}
}
catch (Exception exception)
{
if (_cleanDisconnectInitiated)
{
return;
}
if (exception is AggregateException aggregateException)
{
exception = aggregateException.GetBaseException();
}
if (exception is OperationCanceledException)
{
}
else if (exception is MqttCommunicationException)
{
_logger.Warning(exception, "Communication error while receiving packets.");
}
else
{
_logger.Error(exception, "Error while receiving packets.");
}
_packetDispatcher.FailAll(exception);
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
}
finally
{
_logger.Verbose("Stopped receiving packets.");
}
}
async Task<TResponsePacket> Request<TResponsePacket>(MqttPacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttPacket async Task<TResponsePacket> Request<TResponsePacket>(MqttPacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttPacket
{ {
cancellationToken.ThrowIfCancellationRequested(); cancellationToken.ThrowIfCancellationRequested();
@@ -920,65 +995,6 @@ namespace MQTTnet.Client
} }
} }
async Task ReceivePacketsLoop(CancellationToken cancellationToken)
{
try
{
_logger.Verbose("Start receiving packets.");
while (!cancellationToken.IsCancellationRequested)
{
var packet = await Receive(cancellationToken).ConfigureAwait(false);
if (cancellationToken.IsCancellationRequested)
{
return;
}
if (packet == null)
{
await DisconnectInternal(_packetReceiverTask, null, null).ConfigureAwait(false);
return;
}
await TryProcessReceivedPacket(packet, cancellationToken).ConfigureAwait(false);
}
}
catch (Exception exception)
{
if (_cleanDisconnectInitiated)
{
return;
}
if (exception is AggregateException aggregateException)
{
exception = aggregateException.GetBaseException();
}
if (exception is OperationCanceledException)
{
}
else if (exception is MqttCommunicationException)
{
_logger.Warning(exception, "Communication error while receiving packets.");
}
else
{
_logger.Error(exception, "Error while receiving packets.");
}
_packetDispatcher.FailAll(exception);
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
}
finally
{
_logger.Verbose("Stopped receiving packets.");
}
}
async Task TrySendKeepAliveMessages(CancellationToken cancellationToken) async Task TrySendKeepAliveMessages(CancellationToken cancellationToken)
{ {
try try
@@ -12,6 +12,14 @@ namespace MQTTnet.Client
{ {
public sealed class MqttClientOptions public sealed class MqttClientOptions
{ {
/// <summary>
/// Usually the MQTT packets can be send partially. This is done by using multiple TCP packets
/// or WebSocket frames etc. Unfortunately not all brokers (like Amazon Web Services (AWS)) do support this feature and
/// will close the connection when receiving such packets. If such a service is used this flag must
/// be set to _false_.
/// </summary>
public bool AllowPacketFragmentation { get; set; } = true;
/// <summary> /// <summary>
/// Gets or sets the authentication data. /// Gets or sets the authentication data.
/// <remarks>MQTT 5.0.0+ feature.</remarks> /// <remarks>MQTT 5.0.0+ feature.</remarks>
@@ -24,14 +32,6 @@ namespace MQTTnet.Client
/// </summary> /// </summary>
public string AuthenticationMethod { get; set; } public string AuthenticationMethod { get; set; }
/// <summary>
/// Usually the MQTT packets can be send partially. This is done by using multiple TCP packets
/// or WebSocket frames etc. Unfortunately not all brokers (like Amazon Web Services (AWS)) do support this feature and
/// will close the connection when receiving such packets. If such a service is used this flag must
/// be set to _false_.
/// </summary>
public bool AllowPacketFragmentation { get; set; } = true;
public IMqttClientChannelOptions ChannelOptions { get; set; } public IMqttClientChannelOptions ChannelOptions { get; set; }
/// <summary> /// <summary>
@@ -100,6 +100,11 @@ namespace MQTTnet.Client
/// </summary> /// </summary>
public uint SessionExpiryInterval { get; set; } public uint SessionExpiryInterval { get; set; }
/// <summary>
/// Gets or sets whether an exception should be thrown when the server has sent a non success ACK packet.
/// </summary>
public bool ThrowOnNonSuccessfulConnectResponse { get; set; } = true;
/// <summary> /// <summary>
/// Gets or sets the timeout which will be applied at socket level and internal operations. /// Gets or sets the timeout which will be applied at socket level and internal operations.
/// The default value is the same as for sockets in .NET in general. /// The default value is the same as for sockets in .NET in general.
@@ -222,4 +227,4 @@ namespace MQTTnet.Client
/// </summary> /// </summary>
public int WriterBufferSizeMax { get; set; } = 65535; public int WriterBufferSizeMax { get; set; } = 65535;
} }
} }
@@ -78,7 +78,17 @@ namespace MQTTnet.Client
return _options; return _options;
} }
/// <summary>
/// The client will not throw an exception when the MQTT server responses with a non success ACK packet.
/// This will become the default behavior in future versions of the library.
/// </summary>
public MqttClientOptionsBuilder WithoutThrowOnNonSuccessfulConnectResponse()
{
_options.ThrowOnNonSuccessfulConnectResponse = false;
return this;
}
public MqttClientOptionsBuilder WithAuthentication(string method, byte[] data) public MqttClientOptionsBuilder WithAuthentication(string method, byte[] data)
{ {
_options.AuthenticationMethod = method; _options.AuthenticationMethod = method;
@@ -4,6 +4,7 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Net.Security;
using System.Security.Authentication; using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates; using System.Security.Cryptography.X509Certificates;
@@ -30,11 +31,21 @@ namespace MQTTnet.Client
#endif #endif
#if NETCOREAPP3_1_OR_GREATER #if NETCOREAPP3_1_OR_GREATER
public List<System.Net.Security.SslApplicationProtocol> ApplicationProtocols { get; set; } public List<SslApplicationProtocol> ApplicationProtocols { get; set; }
public System.Net.Security.CipherSuitesPolicy CipherSuitesPolicy { get; set; } public CipherSuitesPolicy CipherSuitesPolicy { get; set; }
public EncryptionPolicy EncryptionPolicy { get; set; } = EncryptionPolicy.RequireEncryption;
public bool AllowRenegotiation { get; set; } = true;
#endif #endif
/// <summary>
/// Gets or sets the target host.
/// If the value is null or empty the same host as the TCP socket host will be used.
/// </summary>
public string TargetHost { get; set; }
#if NET48 || NETCOREAPP3_1_OR_GREATER #if NET48 || NETCOREAPP3_1_OR_GREATER
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12 | SslProtocols.Tls13; public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12 | SslProtocols.Tls13;
#else #else
@@ -2,15 +2,19 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Net; using System.Net;
using System.Net.WebSockets;
namespace MQTTnet.Client namespace MQTTnet.Client
{ {
public sealed class MqttClientWebSocketOptions : IMqttClientChannelOptions public sealed class MqttClientWebSocketOptions : IMqttClientChannelOptions
{ {
public CookieContainer CookieContainer { get; set; } public CookieContainer CookieContainer { get; set; }
public ICredentials Credentials { get; set; }
public MqttClientWebSocketProxyOptions ProxyOptions { get; set; } public MqttClientWebSocketProxyOptions ProxyOptions { get; set; }
public IDictionary<string, string> RequestHeaders { get; set; } public IDictionary<string, string> RequestHeaders { get; set; }
@@ -18,12 +22,34 @@ namespace MQTTnet.Client
public ICollection<string> SubProtocols { get; set; } = new List<string> { "mqtt" }; public ICollection<string> SubProtocols { get; set; } = new List<string> { "mqtt" };
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
public string Uri { get; set; } public string Uri { get; set; }
public override string ToString() public override string ToString()
{ {
return Uri; return Uri;
} }
#if !NETSTANDARD1_3
#if !WINDOWS_UWP
/// <summary>
/// Gets or sets the keep alive interval for the Web Socket connection.
/// This is not related to the keep alive interval for the MQTT protocol.
/// </summary>
public TimeSpan KeepAliveInterval { get; set; } = WebSocket.DefaultKeepAliveInterval;
/// <summary>
/// Gets or sets whether the default (system) credentials should be used when connecting via Web Socket connection.
/// This is not related to the credentials which are used for the MQTT protocol.
/// </summary>
public bool UseDefaultCredentials { get; set; }
#else
/// <summary>
/// Gets or sets the keep alive interval for the Web Socket connection.
/// This is not related to the keep alive interval for the MQTT protocol.
/// </summary>
public TimeSpan KeepAliveInterval { get; set; } = TimeSpan.FromSeconds(30);
#endif
#endif
} }
} }
@@ -16,6 +16,8 @@ namespace MQTTnet.Client
public bool BypassOnLocal { get; set; } public bool BypassOnLocal { get; set; }
public bool UseDefaultCredentials { get; set; }
public string[] BypassList { get; set; } public string[] BypassList { get; set; }
} }
} }
@@ -16,6 +16,8 @@ namespace MQTTnet.Diagnostics
return "net461"; return "net461";
#elif NET472 #elif NET472
return "net472"; return "net472";
#elif NET48
return "net48";
#elif NETSTANDARD1_3 #elif NETSTANDARD1_3
return "netstandard1.3"; return "netstandard1.3";
#elif NETSTANDARD2_0 #elif NETSTANDARD2_0
@@ -2,54 +2,12 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using MQTTnet.Exceptions;
using MQTTnet.Protocol; using MQTTnet.Protocol;
namespace MQTTnet.Formatter namespace MQTTnet.Formatter
{ {
public static class MqttConnectReasonCodeConverter public static class MqttConnectReasonCodeConverter
{ {
public static MqttConnectReasonCode ToConnectReasonCode(MqttConnectReturnCode returnCode)
{
switch (returnCode)
{
case MqttConnectReturnCode.ConnectionAccepted:
{
return MqttConnectReasonCode.Success;
}
case MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion:
{
return MqttConnectReasonCode.UnsupportedProtocolVersion;
}
case MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword:
{
return MqttConnectReasonCode.BadUserNameOrPassword;
}
case MqttConnectReturnCode.ConnectionRefusedIdentifierRejected:
{
return MqttConnectReasonCode.ClientIdentifierNotValid;
}
case MqttConnectReturnCode.ConnectionRefusedServerUnavailable:
{
return MqttConnectReasonCode.ServerUnavailable;
}
case MqttConnectReturnCode.ConnectionRefusedNotAuthorized:
{
return MqttConnectReasonCode.NotAuthorized;
}
default:
{
throw new MqttProtocolViolationException("Unable to convert connect reason code (MQTTv5) to return code (MQTTv3).");
}
}
}
public static MqttConnectReturnCode ToConnectReturnCode(MqttConnectReasonCode reasonCode) public static MqttConnectReturnCode ToConnectReturnCode(MqttConnectReasonCode reasonCode)
{ {
switch (reasonCode) switch (reasonCode)
@@ -59,11 +17,13 @@ namespace MQTTnet.Formatter
return MqttConnectReturnCode.ConnectionAccepted; return MqttConnectReturnCode.ConnectionAccepted;
} }
case MqttConnectReasonCode.Banned:
case MqttConnectReasonCode.NotAuthorized: case MqttConnectReasonCode.NotAuthorized:
{ {
return MqttConnectReturnCode.ConnectionRefusedNotAuthorized; return MqttConnectReturnCode.ConnectionRefusedNotAuthorized;
} }
case MqttConnectReasonCode.BadAuthenticationMethod:
case MqttConnectReasonCode.BadUserNameOrPassword: case MqttConnectReasonCode.BadUserNameOrPassword:
{ {
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
@@ -79,6 +39,7 @@ namespace MQTTnet.Formatter
return MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion; return MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion;
} }
case MqttConnectReasonCode.UseAnotherServer:
case MqttConnectReasonCode.ServerUnavailable: case MqttConnectReasonCode.ServerUnavailable:
case MqttConnectReasonCode.ServerBusy: case MqttConnectReasonCode.ServerBusy:
case MqttConnectReasonCode.ServerMoved: case MqttConnectReasonCode.ServerMoved:
@@ -87,9 +48,7 @@ namespace MQTTnet.Formatter
} }
default: default:
{ return MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion;
throw new MqttProtocolViolationException("Unable to convert connect reason code (MQTTv5) to return code (MQTTv3).");
}
} }
} }
} }
@@ -130,6 +130,9 @@ namespace MQTTnet.Formatter.V5
MaximumQoS = MqttQualityOfServiceLevel.ExactlyOnce MaximumQoS = MqttQualityOfServiceLevel.ExactlyOnce
}; };
// Also set the return code of MQTT 3.1.1 for backward compatibility and debugging purposes.
packet.ReturnCode = MqttConnectReasonCodeConverter.ToConnectReturnCode(packet.ReasonCode);
var propertiesReader = new MqttV5PropertiesReader(_bufferReader); var propertiesReader = new MqttV5PropertiesReader(_bufferReader);
while (propertiesReader.MoveNext()) while (propertiesReader.MoveNext())
{ {
@@ -100,6 +100,12 @@ namespace MQTTnet.Implementations
if (_tcpOptions.TlsOptions?.UseTls == true) if (_tcpOptions.TlsOptions?.UseTls == true)
{ {
var targetHost = _tcpOptions.TlsOptions.TargetHost;
if (string.IsNullOrEmpty(targetHost))
{
targetHost = _tcpOptions.Server;
}
var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback); var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
try try
{ {
@@ -111,14 +117,16 @@ namespace MQTTnet.Implementations
EnabledSslProtocols = _tcpOptions.TlsOptions.SslProtocol, EnabledSslProtocols = _tcpOptions.TlsOptions.SslProtocol,
CertificateRevocationCheckMode = CertificateRevocationCheckMode =
_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode, _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode,
TargetHost = _tcpOptions.Server, TargetHost = targetHost,
CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy,
EncryptionPolicy = _tcpOptions.TlsOptions.EncryptionPolicy,
AllowRenegotiation = _tcpOptions.TlsOptions.AllowRenegotiation
}; };
await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false); await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
#else #else
await sslStream.AuthenticateAsClientAsync( await sslStream.AuthenticateAsClientAsync(
_tcpOptions.Server, targetHost,
LoadCertificates(), LoadCertificates(),
_tcpOptions.TlsOptions.SslProtocol, _tcpOptions.TlsOptions.SslProtocol,
!_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors) !_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors)
@@ -290,12 +298,12 @@ namespace MQTTnet.Implementations
X509CertificateCollection LoadCertificates() X509CertificateCollection LoadCertificates()
{ {
var certificates = new X509CertificateCollection();
if (_tcpOptions.TlsOptions.Certificates == null) if (_tcpOptions.TlsOptions.Certificates == null)
{ {
return certificates; return null;
} }
var certificates = new X509CertificateCollection();
foreach (var certificate in _tcpOptions.TlsOptions.Certificates) foreach (var certificate in _tcpOptions.TlsOptions.Certificates)
{ {
certificates.Add(certificate); certificates.Add(certificate);
@@ -155,14 +155,26 @@ namespace MQTTnet.Implementations
throw new NotSupportedException("Proxies are not supported when using 'netstandard 1.3'."); throw new NotSupportedException("Proxies are not supported when using 'netstandard 1.3'.");
#else #else
var proxyUri = new Uri(_options.ProxyOptions.Address); var proxyUri = new Uri(_options.ProxyOptions.Address);
WebProxy webProxy;
if (!string.IsNullOrEmpty(_options.ProxyOptions.Username) && !string.IsNullOrEmpty(_options.ProxyOptions.Password)) if (!string.IsNullOrEmpty(_options.ProxyOptions.Username) && !string.IsNullOrEmpty(_options.ProxyOptions.Password))
{ {
var credentials = new NetworkCredential(_options.ProxyOptions.Username, _options.ProxyOptions.Password, _options.ProxyOptions.Domain); var credentials = new NetworkCredential(_options.ProxyOptions.Username, _options.ProxyOptions.Password, _options.ProxyOptions.Domain);
return new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList, credentials); webProxy = new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList, credentials);
}
else
{
webProxy = new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList);
}
if (_options.ProxyOptions.UseDefaultCredentials)
{
// Only update the property if required because setting it to false will alter
// the used credentials internally!
webProxy.UseDefaultCredentials = true;
} }
return new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList); return webProxy;
#endif #endif
} }
@@ -207,6 +219,14 @@ namespace MQTTnet.Implementations
} }
} }
#if !NETSTANDARD1_3
#if !WINDOWS_UWP
clientWebSocket.Options.UseDefaultCredentials = _options.UseDefaultCredentials;
#endif
clientWebSocket.Options.KeepAliveInterval = _options.KeepAliveInterval;
#endif
clientWebSocket.Options.Credentials = _options.Credentials;
var certificateValidationHandler = _options.TlsOptions?.CertificateValidationHandler; var certificateValidationHandler = _options.TlsOptions?.CertificateValidationHandler;
if (certificateValidationHandler != null) if (certificateValidationHandler != null)
{ {
@@ -220,6 +240,8 @@ namespace MQTTnet.Implementations
throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net452'."); throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net452'.");
#elif NET461 #elif NET461
throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net461'."); throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net461'.");
#elif NET48
throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net48'.");
#else #else
clientWebSocket.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) => clientWebSocket.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) =>
{ {
+1 -1
View File
@@ -11,7 +11,7 @@
<PropertyGroup> <PropertyGroup>
<TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks> <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks> <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461;net48</TargetFrameworks>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks> <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
<LangVersion>7.3</LangVersion> <LangVersion>7.3</LangVersion>
<AssemblyName>MQTTnet</AssemblyName> <AssemblyName>MQTTnet</AssemblyName>
@@ -4,6 +4,7 @@
using System; using System;
using System.Text; using System.Text;
using MQTTnet.Internal;
namespace MQTTnet namespace MQTTnet
{ {
@@ -16,6 +17,16 @@ namespace MQTTnet
throw new ArgumentNullException(nameof(applicationMessage)); throw new ArgumentNullException(nameof(applicationMessage));
} }
if(applicationMessage.PayloadSegment == EmptyBuffer.ArraySegment)
{
return null;
}
if (applicationMessage.PayloadSegment.Array == null)
{
return null;
}
var payloadSegment = applicationMessage.PayloadSegment; var payloadSegment = applicationMessage.PayloadSegment;
return Encoding.UTF8.GetString(payloadSegment.Array, payloadSegment.Offset, payloadSegment.Count); return Encoding.UTF8.GetString(payloadSegment.Array, payloadSegment.Offset, payloadSegment.Count);
} }
@@ -36,7 +36,7 @@ namespace MQTTnet.Packets
public override string ToString() public override string ToString()
{ {
return $"Disconnect: [ReasonCode={ReasonCode}]"; return $"Disconnect: [ReasonCode={ReasonCode}] [ReasonString={ReasonString}] [ServerReference={ServerReference}] [SessionExpiryInterval={SessionExpiryInterval}]";
} }
} }
} }
@@ -4,26 +4,33 @@
using System; using System;
using System.Collections; using System.Collections;
using System.Collections.Generic;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Packets;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
public sealed class ClientConnectedEventArgs : EventArgs public sealed class ClientConnectedEventArgs : EventArgs
{ {
public ClientConnectedEventArgs(string clientId, string userName, MqttProtocolVersion protocolVersion, string endpoint, IDictionary sessionItems) readonly MqttConnectPacket _connectPacket;
public ClientConnectedEventArgs(MqttConnectPacket connectPacket, MqttProtocolVersion protocolVersion, string endpoint, IDictionary sessionItems)
{ {
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
UserName = userName;
ProtocolVersion = protocolVersion; ProtocolVersion = protocolVersion;
Endpoint = endpoint; Endpoint = endpoint;
SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
} }
public byte[] AuthenticationData => _connectPacket.AuthenticationData;
public string AuthenticationMethod => _connectPacket.AuthenticationMethod;
/// <summary> /// <summary>
/// Gets the client identifier of the connected client. /// Gets the client identifier of the connected client.
/// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
/// </summary> /// </summary>
public string ClientId { get; } public string ClientId => _connectPacket.ClientId;
/// <summary> /// <summary>
/// Gets the endpoint of the connected client. /// Gets the endpoint of the connected client.
@@ -43,6 +50,12 @@ namespace MQTTnet.Server
/// <summary> /// <summary>
/// Gets the user name of the connected client. /// Gets the user name of the connected client.
/// </summary> /// </summary>
public string UserName { get; } public string UserName => _connectPacket.Username;
/// <summary>
/// Gets the user properties sent by the client.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public List<MqttUserProperty> UserProperties => _connectPacket?.UserProperties;
} }
} }
@@ -58,7 +58,7 @@ namespace MQTTnet.Server
_eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer)); _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
} }
public async Task CloseAllConnectionsAsync() public async Task CloseAllConnections()
{ {
List<MqttClient> connections; List<MqttClient> connections;
lock (_clients) lock (_clients)
@@ -306,17 +306,17 @@ namespace MQTTnet.Server
} }
} }
public Task<IList<MqttClientStatus>> GetClientStatusesAsync() public Task<IList<MqttClientStatus>> GetClientsStatus()
{ {
var result = new List<MqttClientStatus>(); var result = new List<MqttClientStatus>();
lock (_clients) lock (_clients)
{ {
foreach (var connection in _clients.Values) foreach (var client in _clients.Values)
{ {
var clientStatus = new MqttClientStatus(connection) var clientStatus = new MqttClientStatus(client)
{ {
Session = new MqttSessionStatus(connection.Session) Session = new MqttSessionStatus(client.Session)
}; };
result.Add(clientStatus); result.Add(clientStatus);
@@ -326,7 +326,7 @@ namespace MQTTnet.Server
return Task.FromResult((IList<MqttClientStatus>)result); return Task.FromResult((IList<MqttClientStatus>)result);
} }
public Task<IList<MqttSessionStatus>> GetSessionStatusAsync() public Task<IList<MqttSessionStatus>> GetSessionsStatus()
{ {
var result = new List<MqttSessionStatus>(); var result = new List<MqttSessionStatus>();
@@ -377,9 +377,7 @@ namespace MQTTnet.Server
if (_eventContainer.ClientConnectedEvent.HasHandlers) if (_eventContainer.ClientConnectedEvent.HasHandlers)
{ {
var eventArgs = new ClientConnectedEventArgs( var eventArgs = new ClientConnectedEventArgs(connectPacket,
connectPacket.ClientId,
connectPacket.Username,
channelAdapter.PacketFormatterAdapter.ProtocolVersion, channelAdapter.PacketFormatterAdapter.ProtocolVersion,
channelAdapter.Endpoint, channelAdapter.Endpoint,
client.Session.Items); client.Session.Items);
@@ -614,17 +612,19 @@ namespace MQTTnet.Server
_sessions[connectPacket.ClientId] = session; _sessions[connectPacket.ClientId] = session;
// Create a new client (always required). // Create a new client (always required).
_clients.TryGetValue(connectPacket.ClientId, out oldClient); lock (_clients)
if (oldClient != null)
{ {
// This will stop the current client from sending and receiving but remains the connection active _clients.TryGetValue(connectPacket.ClientId, out oldClient);
// for a later DISCONNECT packet. if (oldClient != null)
oldClient.IsTakenOver = true; {
// This will stop the current client from sending and receiving but remains the connection active
// for a later DISCONNECT packet.
oldClient.IsTakenOver = true;
}
client = CreateClient(connectPacket, channelAdapter, session);
_clients[connectPacket.ClientId] = client;
} }
client = CreateClient(connectPacket, channelAdapter, session);
_clients[connectPacket.ClientId] = client;
} }
finally finally
{ {
+3 -3
View File
@@ -200,7 +200,7 @@ namespace MQTTnet.Server
{ {
ThrowIfNotStarted(); ThrowIfNotStarted();
return _clientSessionsManager.GetClientStatusesAsync(); return _clientSessionsManager.GetClientsStatus();
} }
public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync() public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
@@ -226,7 +226,7 @@ namespace MQTTnet.Server
{ {
ThrowIfNotStarted(); ThrowIfNotStarted();
return _clientSessionsManager.GetSessionStatusAsync(); return _clientSessionsManager.GetSessionsStatus();
} }
public Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage, CancellationToken cancellationToken = default) public Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage, CancellationToken cancellationToken = default)
@@ -292,7 +292,7 @@ namespace MQTTnet.Server
_cancellationTokenSource.Cancel(false); _cancellationTokenSource.Cancel(false);
await _clientSessionsManager.CloseAllConnectionsAsync().ConfigureAwait(false); await _clientSessionsManager.CloseAllConnections().ConfigureAwait(false);
foreach (var adapter in _adapters) foreach (var adapter in _adapters)
{ {