mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-18 02:13:00 +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] 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;
|
||||||
|
|
||||||
@@ -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();
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
@@ -79,6 +79,16 @@ 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,8 +2,10 @@
|
|||||||
// 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
|
||||||
{
|
{
|
||||||
@@ -11,6 +13,8 @@ namespace MQTTnet.Client
|
|||||||
{
|
{
|
||||||
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; }
|
||||||
@@ -25,5 +29,27 @@ namespace MQTTnet.Client
|
|||||||
{
|
{
|
||||||
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);
|
||||||
}
|
}
|
||||||
|
|
||||||
return 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 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) =>
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -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
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -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)
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user