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] 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] 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] 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>
<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>
<AssemblyName>MQTTnet.Extensions.ManagedClient</AssemblyName>
@@ -2,7 +2,7 @@
<PropertyGroup>
<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>
<AssemblyName>MQTTnet.Extensions.Rpc</AssemblyName>
@@ -2,7 +2,7 @@
<PropertyGroup>
<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>
<AssemblyName>MQTTnet.Extensions.WebSocket4Net</AssemblyName>
@@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information.
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -10,6 +11,7 @@ using MQTTnet.Client;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
@@ -128,7 +130,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
await client.DisconnectAsync(disconnectOptions);
await LongTestDelay();
Assert.IsNotNull(eventArgs);
Assert.AreEqual(MqttDisconnectReasonCode.MessageRateTooHigh, eventArgs.ReasonCode);
}
@@ -156,7 +158,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
await client.DisconnectAsync(disconnectOptions);
await LongTestDelay();
Assert.IsNotNull(eventArgs);
Assert.IsNotNull(eventArgs.UserProperties);
Assert.AreEqual(1, eventArgs.UserProperties.Count);
@@ -164,5 +166,35 @@ namespace MQTTnet.Tests.Clients.MqttClient
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
}
[TestMethod]
[TestMethod]
public void Serialize_Full_MqttConnAckPacket_V310()
{
var connAckPacket = new MqttConnAckPacket
@@ -6,6 +6,7 @@ namespace MQTTnet.Client.Internal
{
public static class MqttClientResultFactory
{
public static readonly MqttClientConnectResultFactory ConnectResult = new MqttClientConnectResultFactory();
public static readonly MqttClientPublishResultFactory PublishResult = new MqttClientPublishResultFactory();
public static readonly MqttClientSubscribeResultFactory SubscribeResult = new MqttClientSubscribeResultFactory();
public static readonly MqttClientUnsubscribeResultFactory UnsubscribeResult = new MqttClientUnsubscribeResultFactory();
+88 -72
View File
@@ -121,11 +121,11 @@ namespace MQTTnet.Client
var mqttClientAliveToken = _mqttClientAlive.Token;
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)
{
connectResult = await ConnectInternal(cancellationToken).ConfigureAwait(false);
connectResult = await ConnectInternal(adapter, cancellationToken).ConfigureAwait(false);
}
else
{
@@ -133,10 +133,16 @@ namespace MQTTnet.Client
// CancellationToken.None or similar.
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;
var keepAliveInterval = Options.KeepAlivePeriod;
@@ -421,7 +427,7 @@ namespace MQTTnet.Client
return CompletedTask.Instance;
}
async Task<MqttClientConnectResult> Authenticate(MqttClientOptions options, CancellationToken cancellationToken)
async Task<MqttClientConnectResult> Authenticate(IMqttChannelAdapter channelAdapter, MqttClientOptions options, CancellationToken cancellationToken)
{
MqttClientConnectResult result;
@@ -434,8 +440,7 @@ namespace MQTTnet.Client
if (receivedPacket is MqttConnAckPacket connAckPacket)
{
var clientConnectResultFactory = new MqttClientConnectResultFactory();
result = clientConnectResultFactory.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
}
else
{
@@ -447,9 +452,18 @@ namespace MQTTnet.Client
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.");
@@ -484,7 +498,7 @@ namespace MQTTnet.Client
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;
@@ -497,10 +511,12 @@ namespace MQTTnet.Client
_publishPacketReceiverQueue?.Dispose();
_publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>();
var connectResult = await Authenticate(Options, effectiveCancellationToken.Token).ConfigureAwait(false);
_publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken);
_packetReceiverTask = Task.Run(() => ReceivePacketsLoop(backgroundCancellationToken), backgroundCancellationToken);
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);
}
return connectResult;
}
@@ -754,6 +770,65 @@ namespace MQTTnet.Client
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
{
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)
{
try
@@ -12,6 +12,14 @@ namespace MQTTnet.Client
{
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>
/// Gets or sets the authentication data.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
@@ -24,14 +32,6 @@ namespace MQTTnet.Client
/// </summary>
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; }
/// <summary>
@@ -100,6 +100,11 @@ namespace MQTTnet.Client
/// </summary>
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>
/// 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.
@@ -222,4 +227,4 @@ namespace MQTTnet.Client
/// </summary>
public int WriterBufferSizeMax { get; set; } = 65535;
}
}
}
@@ -78,7 +78,17 @@ namespace MQTTnet.Client
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)
{
_options.AuthenticationMethod = method;
@@ -4,6 +4,7 @@
using System;
using System.Collections.Generic;
using System.Net.Security;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
@@ -30,11 +31,21 @@ namespace MQTTnet.Client
#endif
#if NETCOREAPP3_1_OR_GREATER
public List<System.Net.Security.SslApplicationProtocol> ApplicationProtocols { get; set; }
public System.Net.Security.CipherSuitesPolicy CipherSuitesPolicy { get; set; }
public List<SslApplicationProtocol> ApplicationProtocols { get; set; }
public CipherSuitesPolicy CipherSuitesPolicy { get; set; }
public EncryptionPolicy EncryptionPolicy { get; set; } = EncryptionPolicy.RequireEncryption;
public bool AllowRenegotiation { get; set; } = true;
#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
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12 | SslProtocols.Tls13;
#else
@@ -2,15 +2,19 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Collections.Generic;
using System.Net;
using System.Net.WebSockets;
namespace MQTTnet.Client
{
public sealed class MqttClientWebSocketOptions : IMqttClientChannelOptions
{
public CookieContainer CookieContainer { get; set; }
public ICredentials Credentials { get; set; }
public MqttClientWebSocketProxyOptions ProxyOptions { 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 MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
public string Uri { get; set; }
public override string ToString()
{
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 UseDefaultCredentials { get; set; }
public string[] BypassList { get; set; }
}
}
@@ -16,6 +16,8 @@ namespace MQTTnet.Diagnostics
return "net461";
#elif NET472
return "net472";
#elif NET48
return "net48";
#elif NETSTANDARD1_3
return "netstandard1.3";
#elif NETSTANDARD2_0
@@ -2,54 +2,12 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using MQTTnet.Exceptions;
using MQTTnet.Protocol;
namespace MQTTnet.Formatter
{
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)
{
switch (reasonCode)
@@ -59,11 +17,13 @@ namespace MQTTnet.Formatter
return MqttConnectReturnCode.ConnectionAccepted;
}
case MqttConnectReasonCode.Banned:
case MqttConnectReasonCode.NotAuthorized:
{
return MqttConnectReturnCode.ConnectionRefusedNotAuthorized;
}
case MqttConnectReasonCode.BadAuthenticationMethod:
case MqttConnectReasonCode.BadUserNameOrPassword:
{
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
@@ -79,6 +39,7 @@ namespace MQTTnet.Formatter
return MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion;
}
case MqttConnectReasonCode.UseAnotherServer:
case MqttConnectReasonCode.ServerUnavailable:
case MqttConnectReasonCode.ServerBusy:
case MqttConnectReasonCode.ServerMoved:
@@ -87,9 +48,7 @@ namespace MQTTnet.Formatter
}
default:
{
throw new MqttProtocolViolationException("Unable to convert connect reason code (MQTTv5) to return code (MQTTv3).");
}
return MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion;
}
}
}
@@ -130,6 +130,9 @@ namespace MQTTnet.Formatter.V5
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);
while (propertiesReader.MoveNext())
{
@@ -100,6 +100,12 @@ namespace MQTTnet.Implementations
if (_tcpOptions.TlsOptions?.UseTls == true)
{
var targetHost = _tcpOptions.TlsOptions.TargetHost;
if (string.IsNullOrEmpty(targetHost))
{
targetHost = _tcpOptions.Server;
}
var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
try
{
@@ -111,14 +117,16 @@ namespace MQTTnet.Implementations
EnabledSslProtocols = _tcpOptions.TlsOptions.SslProtocol,
CertificateRevocationCheckMode =
_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode,
TargetHost = _tcpOptions.Server,
CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy
TargetHost = targetHost,
CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy,
EncryptionPolicy = _tcpOptions.TlsOptions.EncryptionPolicy,
AllowRenegotiation = _tcpOptions.TlsOptions.AllowRenegotiation
};
await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
#else
await sslStream.AuthenticateAsClientAsync(
_tcpOptions.Server,
targetHost,
LoadCertificates(),
_tcpOptions.TlsOptions.SslProtocol,
!_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors)
@@ -290,12 +298,12 @@ namespace MQTTnet.Implementations
X509CertificateCollection LoadCertificates()
{
var certificates = new X509CertificateCollection();
if (_tcpOptions.TlsOptions.Certificates == null)
{
return certificates;
return null;
}
var certificates = new X509CertificateCollection();
foreach (var certificate in _tcpOptions.TlsOptions.Certificates)
{
certificates.Add(certificate);
@@ -155,14 +155,26 @@ namespace MQTTnet.Implementations
throw new NotSupportedException("Proxies are not supported when using 'netstandard 1.3'.");
#else
var proxyUri = new Uri(_options.ProxyOptions.Address);
WebProxy webProxy;
if (!string.IsNullOrEmpty(_options.ProxyOptions.Username) && !string.IsNullOrEmpty(_options.ProxyOptions.Password))
{
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
}
@@ -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;
if (certificateValidationHandler != null)
{
@@ -220,6 +240,8 @@ namespace MQTTnet.Implementations
throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net452'.");
#elif 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
clientWebSocket.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) =>
{
+1 -1
View File
@@ -11,7 +11,7 @@
<PropertyGroup>
<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>
<LangVersion>7.3</LangVersion>
<AssemblyName>MQTTnet</AssemblyName>
@@ -4,6 +4,7 @@
using System;
using System.Text;
using MQTTnet.Internal;
namespace MQTTnet
{
@@ -16,6 +17,16 @@ namespace MQTTnet
throw new ArgumentNullException(nameof(applicationMessage));
}
if(applicationMessage.PayloadSegment == EmptyBuffer.ArraySegment)
{
return null;
}
if (applicationMessage.PayloadSegment.Array == null)
{
return null;
}
var payloadSegment = applicationMessage.PayloadSegment;
return Encoding.UTF8.GetString(payloadSegment.Array, payloadSegment.Offset, payloadSegment.Count);
}
@@ -36,7 +36,7 @@ namespace MQTTnet.Packets
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.Collections;
using System.Collections.Generic;
using MQTTnet.Formatter;
using MQTTnet.Packets;
namespace MQTTnet.Server
{
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));
UserName = userName;
_connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
ProtocolVersion = protocolVersion;
Endpoint = endpoint;
SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
}
public byte[] AuthenticationData => _connectPacket.AuthenticationData;
public string AuthenticationMethod => _connectPacket.AuthenticationMethod;
/// <summary>
/// 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.
/// </summary>
public string ClientId { get; }
public string ClientId => _connectPacket.ClientId;
/// <summary>
/// Gets the endpoint of the connected client.
@@ -43,6 +50,12 @@ namespace MQTTnet.Server
/// <summary>
/// Gets the user name of the connected client.
/// </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));
}
public async Task CloseAllConnectionsAsync()
public async Task CloseAllConnections()
{
List<MqttClient> connections;
lock (_clients)
@@ -306,17 +306,17 @@ namespace MQTTnet.Server
}
}
public Task<IList<MqttClientStatus>> GetClientStatusesAsync()
public Task<IList<MqttClientStatus>> GetClientsStatus()
{
var result = new List<MqttClientStatus>();
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);
@@ -326,7 +326,7 @@ namespace MQTTnet.Server
return Task.FromResult((IList<MqttClientStatus>)result);
}
public Task<IList<MqttSessionStatus>> GetSessionStatusAsync()
public Task<IList<MqttSessionStatus>> GetSessionsStatus()
{
var result = new List<MqttSessionStatus>();
@@ -377,9 +377,7 @@ namespace MQTTnet.Server
if (_eventContainer.ClientConnectedEvent.HasHandlers)
{
var eventArgs = new ClientConnectedEventArgs(
connectPacket.ClientId,
connectPacket.Username,
var eventArgs = new ClientConnectedEventArgs(connectPacket,
channelAdapter.PacketFormatterAdapter.ProtocolVersion,
channelAdapter.Endpoint,
client.Session.Items);
@@ -614,17 +612,19 @@ namespace MQTTnet.Server
_sessions[connectPacket.ClientId] = session;
// Create a new client (always required).
_clients.TryGetValue(connectPacket.ClientId, out oldClient);
if (oldClient != null)
lock (_clients)
{
// This will stop the current client from sending and receiving but remains the connection active
// for a later DISCONNECT packet.
oldClient.IsTakenOver = true;
_clients.TryGetValue(connectPacket.ClientId, out oldClient);
if (oldClient != null)
{
// 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
{
+3 -3
View File
@@ -200,7 +200,7 @@ namespace MQTTnet.Server
{
ThrowIfNotStarted();
return _clientSessionsManager.GetClientStatusesAsync();
return _clientSessionsManager.GetClientsStatus();
}
public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
@@ -226,7 +226,7 @@ namespace MQTTnet.Server
{
ThrowIfNotStarted();
return _clientSessionsManager.GetSessionStatusAsync();
return _clientSessionsManager.GetSessionsStatus();
}
public Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage, CancellationToken cancellationToken = default)
@@ -292,7 +292,7 @@ namespace MQTTnet.Server
_cancellationTokenSource.Cancel(false);
await _clientSessionsManager.CloseAllConnectionsAsync().ConfigureAwait(false);
await _clientSessionsManager.CloseAllConnections().ConfigureAwait(false);
foreach (var adapter in _adapters)
{