mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-15 08:33:01 +08:00
Compare commits
11
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8ad86928f1 | ||
|
|
d8e5936d45 | ||
|
|
caa717544b | ||
|
|
66b77f8d6d | ||
|
|
ab5264c8ca | ||
|
|
db12309bae | ||
|
|
2953e0dcc5 | ||
|
|
1b18ac9fe3 | ||
|
|
2cdad6622a | ||
|
|
7601d11a7a | ||
|
|
0bb4c6ef09 |
@@ -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();
|
||||
|
||||
@@ -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) =>
|
||||
{
|
||||
|
||||
@@ -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
|
||||
{
|
||||
|
||||
@@ -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)
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user