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

Merge branch 'master' into 1291-access-client-status-in-client-disconnected-handler

Browse Source
This commit is contained in:
Christian
2022-07-23 18:09:05 +02:00
parent 09385b1b36 cbf413f04e
commit cef61900df
27 changed files with 739 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs
-1
View File
@@ -98,7 +98,6 @@ namespace MQTTnet.Extensions.WebSocket4Net
#else
certificates.Add(certificate);
#endif
}
}
Source/MQTTnet.TestApp/ServerTest.cs
+1 -1
View File
@@ -118,7 +118,7 @@ namespace MQTTnet.TestApp
{
if (e.ClientId == "SpecialClient")
{
if (e.Username != "USER" || e.Password != "PASS")
if (e.UserName != "USER" || e.Password != "PASS")
{
e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
}
Source/MQTTnet.Tests/Server/General.cs
+2 -144
View File
@@ -798,38 +798,7 @@ namespace MQTTnet.Tests.Server
Assert.IsTrue(longBody.SequenceEqual(receivedBody ?? new byte[0]));
}
}
[TestMethod]
public async Task Session_Takeover()
{
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
var options = new MqttClientOptionsBuilder().WithCleanSession(false)
.WithProtocolVersion(MqttProtocolVersion.V500) // Disconnect reason is only available in MQTT 5+
.WithClientId("a");
var client1 = await testEnvironment.ConnectClient(options);
await Task.Delay(500);
var disconnectReason = MqttClientDisconnectReason.NormalDisconnection;
client1.DisconnectedAsync += c =>
{
disconnectReason = c.Reason;
return PlatformAbstractionLayer.CompletedTask;
};
var client2 = await testEnvironment.ConnectClient(options);
await Task.Delay(500);
Assert.IsFalse(client1.IsConnected);
Assert.IsTrue(client2.IsConnected);
Assert.AreEqual(MqttClientDisconnectReason.SessionTakenOver, disconnectReason);
}
}
[TestMethod]
public async Task Set_Subscription_At_Server()
{
@@ -944,26 +913,6 @@ namespace MQTTnet.Tests.Server
}
}
[TestMethod]
public async Task Use_Clean_Session()
{
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
var client = testEnvironment.CreateClient();
var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build());
// Create the session including the subscription.
var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a").WithCleanSession(false));
await client1.SubscribeAsync("x");
await client1.DisconnectAsync();
await Task.Delay(500);
Assert.IsFalse(connectResult.IsSessionPresent);
}
}
[TestMethod]
public async Task Use_Empty_Client_ID()
{
@@ -986,98 +935,7 @@ namespace MQTTnet.Tests.Server
Assert.IsTrue(client.IsConnected);
}
}
[TestMethod]
public async Task Will_Message_Do_Not_Send_On_Clean_Disconnect()
{
using (var testEnvironment = CreateTestEnvironment())
{
var receivedMessagesCount = 0;
await testEnvironment.StartServer();
var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will");
var c1 = await testEnvironment.ConnectClient();
c1.ApplicationMessageReceivedAsync += e =>
{
Interlocked.Increment(ref receivedMessagesCount);
return PlatformAbstractionLayer.CompletedTask;
};
await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
var c2 = await testEnvironment.ConnectClient(clientOptions);
await c2.DisconnectAsync().ConfigureAwait(false);
await Task.Delay(1000);
Assert.AreEqual(0, receivedMessagesCount);
}
}
[TestMethod]
public async Task Will_Message_Do_Not_Send_On_Takeover()
{
using (var testEnvironment = CreateTestEnvironment())
{
var receivedMessagesCount = 0;
await testEnvironment.StartServer();
// C1 will receive the last will!
var c1 = await testEnvironment.ConnectClient();
c1.ApplicationMessageReceivedAsync += e =>
{
Interlocked.Increment(ref receivedMessagesCount);
return PlatformAbstractionLayer.CompletedTask;
};
await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
// C2 has the last will defined.
var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithClientId("WillOwner");
var c2 = await testEnvironment.ConnectClient(clientOptions);
// C3 will do the connection takeover.
var c3 = await testEnvironment.ConnectClient(clientOptions);
await Task.Delay(1000);
Assert.AreEqual(0, receivedMessagesCount);
}
}
[TestMethod]
public async Task Will_Message_Send()
{
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder());
var receivedMessagesCount = 0;
c1.ApplicationMessageReceivedAsync += e =>
{
Interlocked.Increment(ref receivedMessagesCount);
return PlatformAbstractionLayer.CompletedTask;
};
await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce);
var c2 = await testEnvironment.ConnectClient(clientOptions);
c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent.
await Task.Delay(1000);
Assert.AreEqual(1, receivedMessagesCount);
}
}
void ConnectionValidationHandler(ValidatingConnectionEventArgs eventArgs)
{
if (_connected.ContainsKey(eventArgs.ClientId))
Source/MQTTnet.Tests/Server/QoS_Tests.cs
+124
View File
@@ -0,0 +1,124 @@
// Licensed to the .NET Foundation under one or more agreements.
// 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.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Implementations;
using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.Server
{
[TestClass]
public sealed class QoS_Tests : BaseTestClass
{
[TestMethod]
public async Task Fire_Event_On_Client_Acknowledges_QoS_0()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
ClientAcknowledgedPublishPacketEventArgs eventArgs = null;
server.ClientAcknowledgedPublishPacketAsync += args =>
{
eventArgs = args;
return PlatformAbstractionLayer.CompletedTask;
};
var client1 = await testEnvironment.ConnectClient();
await client1.SubscribeAsync("A");
var client2 = await testEnvironment.ConnectClient();
await client2.PublishStringAsync("A");
await LongTestDelay();
// Must be null because no event should be fired for QoS 0.
Assert.IsNull(eventArgs);
}
}
[TestMethod]
public async Task Fire_Event_On_Client_Acknowledges_QoS_1()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
ClientAcknowledgedPublishPacketEventArgs eventArgs = null;
server.ClientAcknowledgedPublishPacketAsync += args =>
{
eventArgs = args;
return PlatformAbstractionLayer.CompletedTask;
};
var client1 = await testEnvironment.ConnectClient();
await client1.SubscribeAsync("A", MqttQualityOfServiceLevel.AtLeastOnce);
var client2 = await testEnvironment.ConnectClient();
await client2.PublishStringAsync("A", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await LongTestDelay();
Assert.IsNotNull(eventArgs);
Assert.IsNotNull(eventArgs.PublishPacket);
Assert.IsNotNull(eventArgs.AcknowledgePacket);
Assert.IsTrue(eventArgs.IsCompleted);
Assert.AreEqual("A", eventArgs.PublishPacket.Topic);
}
}
[TestMethod]
public async Task Fire_Event_On_Client_Acknowledges_QoS_2()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
var eventArgs = new List<ClientAcknowledgedPublishPacketEventArgs>();
server.ClientAcknowledgedPublishPacketAsync += args =>
{
lock (eventArgs)
{
eventArgs.Add(args);
}
return PlatformAbstractionLayer.CompletedTask;
};
var client1 = await testEnvironment.ConnectClient();
await client1.SubscribeAsync("A", MqttQualityOfServiceLevel.ExactlyOnce);
var client2 = await testEnvironment.ConnectClient();
await client2.PublishStringAsync("A", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
await LongTestDelay();
Assert.AreEqual(2, eventArgs.Count);
var firstEvent = eventArgs[0];
Assert.IsNotNull(firstEvent);
Assert.IsNotNull(firstEvent.PublishPacket);
Assert.IsNotNull(firstEvent.AcknowledgePacket);
Assert.IsFalse(firstEvent.IsCompleted);
Assert.AreEqual("A", firstEvent.PublishPacket.Topic);
var secondEvent = eventArgs[1];
Assert.IsNotNull(secondEvent);
Assert.IsNotNull(secondEvent.PublishPacket);
Assert.IsNotNull(secondEvent.AcknowledgePacket);
Assert.IsTrue(secondEvent.IsCompleted);
Assert.AreEqual("A", secondEvent.PublishPacket.Topic);
}
}
}
}
Source/MQTTnet.Tests/Server/Security_Tests.cs
+2 -4
View File
@@ -2,10 +2,8 @@
// 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.Threading.Tasks;
using Microsoft.CodeAnalysis;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Adapter;
using MQTTnet.Client;
@@ -40,7 +38,7 @@ namespace MQTTnet.Tests.Server
testEnvironment.Server.ValidatingConnectionAsync += eventArgs =>
{
if (eventArgs.Username == "SECRET")
if (eventArgs.UserName == "SECRET")
{
eventArgs.ReasonCode = MqttConnectReasonCode.Success;
}
@@ -149,7 +147,7 @@ namespace MQTTnet.Tests.Server
server.ValidatingConnectionAsync += e =>
{
if (e.Username != "UserName1")
if (e.UserName != "UserName1")
{
e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
}
Source/MQTTnet.Tests/Server/Session_Tests.cs
+84 -1
View File
@@ -14,7 +14,6 @@ using MQTTnet.Implementations;
using MQTTnet.Protocol;
using MQTTnet.Server;
using MQTTnet.Tests.Mockups;
using MqttClient = MQTTnet.Client.MqttClient;
namespace MQTTnet.Tests.Server
{
@@ -220,6 +219,37 @@ namespace MQTTnet.Tests.Server
}
}
[TestMethod]
public async Task Session_Takeover()
{
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
var options = new MqttClientOptionsBuilder().WithCleanSession(false)
.WithProtocolVersion(MqttProtocolVersion.V500) // Disconnect reason is only available in MQTT 5+
.WithClientId("a");
var client1 = await testEnvironment.ConnectClient(options);
await Task.Delay(500);
var disconnectReason = MqttClientDisconnectReason.NormalDisconnection;
client1.DisconnectedAsync += c =>
{
disconnectReason = c.Reason;
return PlatformAbstractionLayer.CompletedTask;
};
var client2 = await testEnvironment.ConnectClient(options);
await Task.Delay(500);
Assert.IsFalse(client1.IsConnected);
Assert.IsTrue(client2.IsConnected);
Assert.AreEqual(MqttClientDisconnectReason.SessionTakenOver, disconnectReason);
}
}
[TestMethod]
public async Task Set_Session_Item()
{
@@ -277,6 +307,59 @@ namespace MQTTnet.Tests.Server
}
}
[TestMethod]
public async Task Use_Clean_Session()
{
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
var client = testEnvironment.CreateClient();
var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build());
// Create the session including the subscription.
var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a").WithCleanSession(false));
await client1.SubscribeAsync("x");
await client1.DisconnectAsync();
await Task.Delay(500);
Assert.IsFalse(connectResult.IsSessionPresent);
}
}
[TestMethod]
public async Task Will_Message_Do_Not_Send_On_Takeover()
{
using (var testEnvironment = CreateTestEnvironment())
{
var receivedMessagesCount = 0;
await testEnvironment.StartServer();
// C1 will receive the last will!
var c1 = await testEnvironment.ConnectClient();
c1.ApplicationMessageReceivedAsync += e =>
{
Interlocked.Increment(ref receivedMessagesCount);
return PlatformAbstractionLayer.CompletedTask;
};
await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
// C2 has the last will defined.
var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithClientId("WillOwner");
var c2 = await testEnvironment.ConnectClient(clientOptions);
// C3 will do the connection takeover.
var c3 = await testEnvironment.ConnectClient(clientOptions);
await Task.Delay(1000);
Assert.AreEqual(0, receivedMessagesCount);
}
}
async Task<IMqttClient> ConnectAndSubscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive)
{
try
Source/MQTTnet.Tests/Server/Will_Tests.cs
+71
View File
@@ -0,0 +1,71 @@
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Implementations;
using MQTTnet.Protocol;
namespace MQTTnet.Tests.Server
{
[TestClass]
public sealed class Will_Tests : BaseTestClass
{
[TestMethod]
public async Task Will_Message_Do_Not_Send_On_Clean_Disconnect()
{
using (var testEnvironment = CreateTestEnvironment())
{
var receivedMessagesCount = 0;
await testEnvironment.StartServer();
var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will");
var c1 = await testEnvironment.ConnectClient();
c1.ApplicationMessageReceivedAsync += e =>
{
Interlocked.Increment(ref receivedMessagesCount);
return PlatformAbstractionLayer.CompletedTask;
};
await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
var c2 = await testEnvironment.ConnectClient(clientOptions);
await c2.DisconnectAsync().ConfigureAwait(false);
await Task.Delay(1000);
Assert.AreEqual(0, receivedMessagesCount);
}
}
[TestMethod]
public async Task Will_Message_Send()
{
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder());
var receivedMessagesCount = 0;
c1.ApplicationMessageReceivedAsync += e =>
{
Interlocked.Increment(ref receivedMessagesCount);
return PlatformAbstractionLayer.CompletedTask;
};
await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce);
var c2 = await testEnvironment.ConnectClient(clientOptions);
c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent.
await Task.Delay(1000);
Assert.AreEqual(1, receivedMessagesCount);
}
}
}
}
Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs
+1 -2
View File
@@ -264,8 +264,7 @@ namespace MQTTnet.Client
return this;
}
// TODO: Consider creating _MqttClientTcpOptionsBuilder_ as overload.
public MqttClientOptionsBuilder WithTcpServer(Action<MqttClientTcpOptions> optionsBuilder)
{
if (optionsBuilder == null)
Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs
+12
View File
@@ -2,6 +2,7 @@
// 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.Net;
using System.Net.Sockets;
namespace MQTTnet.Client
@@ -10,8 +11,19 @@ namespace MQTTnet.Client
{
public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
/// <summary>
/// Gets the local endpoint (network card) which is used by the client.
/// Set it to _null_ to let the OS select the network card.
/// </summary>
public EndPoint LocalEndpoint { get; set; }
public int BufferSize { get; set; } = 8192;
/// <summary>
/// Gets or sets whether the underlying socket should run in dual mode.
/// Leaving this _null_ will avoid setting this value at socket level.
/// Setting this a value other than _null_ will throw an exception when only IPv4 is supported on the machine.
/// </summary>
public bool? DualMode { get; set; }
public LingerOption LingerState { get; set; } = new LingerOption(true, 0);
Source/MQTTnet/Client/Options/MqttClientWebSocketOptions.cs
+7 -7
View File
@@ -7,23 +7,23 @@ using System.Net;
namespace MQTTnet.Client
{
public class MqttClientWebSocketOptions : IMqttClientChannelOptions
public sealed class MqttClientWebSocketOptions : IMqttClientChannelOptions
{
public string Uri { get; set; }
public CookieContainer CookieContainer { get; set; }
public MqttClientWebSocketProxyOptions ProxyOptions { get; set; }
public IDictionary<string, string> RequestHeaders { get; set; }
public ICollection<string> SubProtocols { get; set; } = new List<string> { "mqtt" };
public CookieContainer CookieContainer { get; set; }
public MqttClientWebSocketProxyOptions ProxyOptions { get; set; }
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
public string Uri { get; set; }
public override string ToString()
{
return Uri;
}
}
}
}
Source/MQTTnet/Client/Options/MqttClientWebSocketProxyOptions.cs
+1 -1
View File
@@ -4,7 +4,7 @@
namespace MQTTnet.Client
{
public class MqttClientWebSocketProxyOptions
public sealed class MqttClientWebSocketProxyOptions
{
public string Address { get; set; }
Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedEventArgs.cs
+8 -2
View File
@@ -25,6 +25,7 @@ namespace MQTTnet.Client
ClientId = clientId;
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
PublishPacket = publishPacket;
PacketIdentifier = publishPacket.PacketIdentifier;
_acknowledgeHandler = acknowledgeHandler;
}
@@ -62,12 +63,17 @@ namespace MQTTnet.Client
public bool IsHandled { get; set; }
/// <summary>
/// Gets ir sets whether the library should send MQTT ACK packets automatically if required.
/// Gets or sets whether the library should send MQTT ACK packets automatically if required.
/// </summary>
public bool AutoAcknowledge { get; set; } = true;
public object Tag { get; set; }
/// <summary>
/// Gets the indentifier of MQTT packet
/// </summary>
public ushort PacketIdentifier { get; }
public Task AcknowledgeAsync(CancellationToken cancellationToken)
{
if (_acknowledgeHandler == null)
Source/MQTTnet/Implementations/MqttTcpChannel.cs
+5
View File
@@ -68,6 +68,11 @@ namespace MQTTnet.Implementations
socket = new CrossPlatformSocket(_tcpOptions.AddressFamily);
}
if (_tcpOptions.LocalEndpoint != null)
{
socket.Bind(_tcpOptions.LocalEndpoint);
}
socket.ReceiveBufferSize = _tcpOptions.BufferSize;
socket.SendBufferSize = _tcpOptions.BufferSize;
socket.SendTimeout = (int)_clientOptions.Timeout.TotalMilliseconds;
Source/MQTTnet/Implementations/MqttWebSocketChannel.cs
-1
View File
@@ -118,7 +118,6 @@ namespace MQTTnet.Implementations
void SetupClientWebSocket(ClientWebSocket clientWebSocket)
{
if (_options.ProxyOptions != null)
{
clientWebSocket.Options.Proxy = CreateProxy();
Source/MQTTnet/Server/Events/ClientAcknowledgedPublishPacketEventArgs.cs
+38
View File
@@ -0,0 +1,38 @@
// Licensed to the .NET Foundation under one or more agreements.
// 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;
using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class ClientAcknowledgedPublishPacketEventArgs : EventArgs
{
/// <summary>
/// Gets the packet which was used for acknowledge. This can be a PubAck or PubComp packet.
/// </summary>
public MqttPacketWithIdentifier AcknowledgePacket { get; internal set; }
/// <summary>
/// Gets the ID of the client which acknowledged a PUBLISH packet.
/// </summary>
public string ClientId { get; internal set; }
/// <summary>
/// Gets whether the PUBLISH packet is fully acknowledged. This is the case for PUBACK (QoS 1) and PUBCOMP (QoS 2.
/// </summary>
public bool IsCompleted => AcknowledgePacket is MqttPubAckPacket || AcknowledgePacket is MqttPubCompPacket;
/// <summary>
/// Gets the PUBLISH packet which was acknowledged.
/// </summary>
public MqttPublishPacket PublishPacket { get; internal set; }
/// <summary>
/// Gets the session items which contain custom user data per session.
/// </summary>
public IDictionary SessionItems { get; internal set; }
}
}
Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs
+14 -6
View File
@@ -17,13 +17,12 @@ namespace MQTTnet.Server
{
public sealed class ValidatingConnectionEventArgs : EventArgs
{
readonly IMqttChannelAdapter _clientAdapter;
readonly MqttConnectPacket _connectPacket;
public ValidatingConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter)
{
_connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
_clientAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter));
ChannelAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter));
}
/// <summary>
@@ -44,6 +43,12 @@ namespace MQTTnet.Server
/// </summary>
public string AuthenticationMethod => _connectPacket.AuthenticationMethod;
/// <summary>
/// Gets the channel adapter. This can be a _MqttConnectionContext_ (used in ASP.NET), a _MqttChannelAdapter_ (used for
/// TCP or WebSockets) or a custom implementation.
/// </summary>
public IMqttChannelAdapter ChannelAdapter { get; }
/// <summary>
/// Gets or sets a value indicating whether clean sessions are used or not.
/// When a client connects to a broker it can connect using either a non persistent connection (clean session) or a
@@ -56,7 +61,7 @@ namespace MQTTnet.Server
/// </summary>
public bool? CleanSession => _connectPacket.CleanSession;
public X509Certificate2 ClientCertificate => _clientAdapter.ClientCertificate;
public X509Certificate2 ClientCertificate => ChannelAdapter.ClientCertificate;
/// <summary>
/// Gets the client identifier.
@@ -64,9 +69,9 @@ namespace MQTTnet.Server
/// </summary>
public string ClientId => _connectPacket.ClientId;
public string Endpoint => _clientAdapter.Endpoint;
public string Endpoint => ChannelAdapter.Endpoint;
public bool IsSecureConnection => _clientAdapter.IsSecureConnection;
public bool IsSecureConnection => ChannelAdapter.IsSecureConnection;
/// <summary>
/// Gets or sets the keep alive period.
@@ -85,7 +90,7 @@ namespace MQTTnet.Server
public string Password => Encoding.UTF8.GetString(RawPassword ?? PlatformAbstractionLayer.EmptyByteArray);
public MqttProtocolVersion ProtocolVersion => _clientAdapter.PacketFormatterAdapter.ProtocolVersion;
public MqttProtocolVersion ProtocolVersion => ChannelAdapter.PacketFormatterAdapter.ProtocolVersion;
public byte[] RawPassword => _connectPacket.Password;
@@ -161,7 +166,10 @@ namespace MQTTnet.Server
/// </summary>
public ushort TopicAliasMaximum => _connectPacket.TopicAliasMaximum;
[Obsolete("This property name has a typo. Use 'UserName' instead. This one will be removed soon.")]
public string Username => _connectPacket.Username;
public string UserName => _connectPacket.Username;
/// <summary>
/// Gets or sets the user properties.
Source/MQTTnet/Server/Internal/MqttClient.cs
+250 -46
View File
@@ -11,6 +11,7 @@ using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Implementations;
using MQTTnet.Internal;
using MQTTnet.PacketDispatcher;
using MQTTnet.Packets;
@@ -20,20 +21,20 @@ namespace MQTTnet.Server
{
public sealed class MqttClient
{
readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
readonly MqttConnectPacket _connectPacket;
readonly MqttServerEventContainer _eventContainer;
readonly MqttNetSourceLogger _logger;
readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
readonly MqttClientSessionsManager _sessionsManager;
readonly MqttNetSourceLogger _logger;
readonly MqttServerOptions _serverOptions;
readonly MqttServerEventContainer _eventContainer;
readonly MqttConnectPacket _connectPacket;
readonly MqttClientSessionsManager _sessionsManager;
readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
CancellationTokenSource _cancellationToken;
public MqttClient(
MqttConnectPacket connectPacket,
IMqttChannelAdapter channelAdapter,
@@ -53,43 +54,31 @@ namespace MQTTnet.Server
Session = session ?? throw new ArgumentNullException(nameof(session));
_connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
if (logger == null) throw new ArgumentNullException(nameof(logger));
if (logger == null)
{
throw new ArgumentNullException(nameof(logger));
}
_logger = logger.WithSource(nameof(MqttClient));
}
public string Id => _connectPacket.ClientId;
public IMqttChannelAdapter ChannelAdapter { get; }
public string Endpoint { get; }
public IMqttChannelAdapter ChannelAdapter { get; }
public string Id => _connectPacket.ClientId;
public MqttClientStatistics Statistics { get; } = new MqttClientStatistics();
public bool IsCleanDisconnect { get; private set; }
public bool IsRunning { get; private set; }
public bool IsTakenOver { get; set; }
public ushort KeepAlivePeriod => _connectPacket.KeepAlivePeriod;
public MqttSession Session { get; }
public bool IsTakenOver { get; set; }
public bool IsCleanDisconnect { get; private set; }
public async Task StopAsync(MqttDisconnectReasonCode reason)
{
IsRunning = false;
if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout)
{
// Is is very important to send the DISCONNECT packet here BEFORE cancelling the
// token because the entire connection is closed (disposed) as soon as the cancellation
// token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
// at the client!
await TrySendDisconnectPacket(reason).ConfigureAwait(false);
}
StopInternal();
}
public MqttClientStatistics Statistics { get; } = new MqttClientStatistics();
public void ResetStatistics()
{
@@ -110,7 +99,7 @@ namespace MQTTnet.Server
try
{
Task.Run(() => SendPacketsLoop(cancellationToken), cancellationToken).RunInBackground(_logger);
IsRunning = true;
await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false);
@@ -130,10 +119,10 @@ namespace MQTTnet.Server
{
var willPublishPacket = _packetFactories.Publish.Create(Session.LatestConnectPacket);
var willApplicationMessage = _applicationMessageFactory.Create(willPublishPacket);
_= _sessionsManager.DispatchApplicationMessage(Id, willApplicationMessage);
_ = _sessionsManager.DispatchApplicationMessage(Id, willApplicationMessage);
Session.WillMessageSent = true;
_logger.Info("Client '{0}': Published will message.", Id);
}
@@ -149,11 +138,226 @@ namespace MQTTnet.Server
// This might break the protocol but the user wants that.
return;
}
await ChannelAdapter.SendPacketAsync(packet, cancellationToken).ConfigureAwait(false);
Statistics.HandleSentPacket(packet);
}
public async Task StopAsync(MqttDisconnectReasonCode reason)
{
IsRunning = false;
if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout)
{
// Is is very important to send the DISCONNECT packet here BEFORE cancelling the
// token because the entire connection is closed (disposed) as soon as the cancellation
// token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
// at the client!
await TrySendDisconnectPacket(reason).ConfigureAwait(false);
}
StopInternal();
}
Task ClientAcknowledgedPublishPacket(MqttPublishPacket publishPacket, MqttPacketWithIdentifier acknowledgePacket)
{
if (_eventContainer.ClientAcknowledgedPublishPacketEvent.HasHandlers)
{
var eventArgs = new ClientAcknowledgedPublishPacketEventArgs
{
PublishPacket = publishPacket,
AcknowledgePacket = acknowledgePacket,
ClientId = Id,
SessionItems = Session.Items
};
return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger);
}
return PlatformAbstractionLayer.CompletedTask;
}
Task HandleIncomingPubAckPacket(MqttPubAckPacket pubAckPacket)
{
var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier);
if (acknowledgedPublishPacket != null)
{
return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubAckPacket);
}
return PlatformAbstractionLayer.CompletedTask;
}
Task HandleIncomingPubCompPacket(MqttPubCompPacket pubCompPacket)
{
var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier);
if (acknowledgedPublishPacket != null)
{
return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubCompPacket);
}
return PlatformAbstractionLayer.CompletedTask;
}
async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
HandleTopicAlias(publishPacket);
InterceptingPublishEventArgs interceptingPublishEventArgs = null;
var applicationMessage = _applicationMessageFactory.Create(publishPacket);
var closeConnection = false;
var processPublish = true;
if (_eventContainer.InterceptingPublishEvent.HasHandlers)
{
interceptingPublishEventArgs = new InterceptingPublishEventArgs
{
ClientId = Id,
ApplicationMessage = applicationMessage,
SessionItems = Session.Items,
ProcessPublish = true,
CloseConnection = false,
CancellationToken = cancellationToken
};
if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic))
{
// This can happen if a topic alias us used but the topic is
// unknown to the server.
interceptingPublishEventArgs.Response.ReasonCode = MqttPubAckReasonCode.TopicNameInvalid;
interceptingPublishEventArgs.ProcessPublish = false;
}
await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false);
applicationMessage = interceptingPublishEventArgs.ApplicationMessage;
closeConnection = interceptingPublishEventArgs.CloseConnection;
processPublish = interceptingPublishEventArgs.ProcessPublish;
}
if (closeConnection)
{
await StopAsync(MqttDisconnectReasonCode.UnspecifiedError);
return;
}
if (processPublish && applicationMessage != null)
{
await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false);
}
switch (publishPacket.QualityOfServiceLevel)
{
case MqttQualityOfServiceLevel.AtMostOnce:
{
// Do nothing since QoS 0 has no ACK at all!
break;
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket));
break;
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket));
break;
}
default:
{
throw new MqttCommunicationException("Received a not supported QoS level.");
}
}
}
async Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
{
var acknowledgedPublishPacket = Session.PeekAcknowledgePublishPacket(pubRecPacket.PacketIdentifier);
if (acknowledgedPublishPacket != null)
{
await ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubRecPacket).ConfigureAwait(false);
}
var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
}
void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)
{
var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket));
}
async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{
var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult);
Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket));
if (subscribeResult.CloseConnection)
{
StopInternal();
return;
}
if (subscribeResult.RetainedMessages != null)
{
foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages)
{
var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
}
}
async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
{
var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false);
var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult);
Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket));
if (unsubscribeResult.CloseConnection)
{
StopInternal();
}
}
void HandleTopicAlias(MqttPublishPacket publishPacket)
{
if (publishPacket.TopicAlias == 0)
{
return;
}
lock (_topicAlias)
{
if (!string.IsNullOrEmpty(publishPacket.Topic))
{
_topicAlias[publishPacket.TopicAlias] = publishPacket.Topic;
}
else
{
if (_topicAlias.TryGetValue(publishPacket.TopicAlias, out var topic))
{
publishPacket.Topic = topic;
}
else
{
_logger.Warning("Client '{0}': Received invalid topic alias ({1}).", Id, publishPacket.TopicAlias);
}
}
}
}
async Task<MqttPacket> InterceptPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
{
if (!_eventContainer.InterceptingOutboundPacketEvent.HasHandlers)
@@ -163,12 +367,12 @@ namespace MQTTnet.Server
var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items);
await _eventContainer.InterceptingOutboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
if (!interceptingPacketEventArgs.ProcessPacket || packet == null)
{
return null;
}
return interceptingPacketEventArgs.Packet;
}
@@ -190,7 +394,7 @@ namespace MQTTnet.Server
}
var processPacket = true;
if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers)
{
var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items);
@@ -198,7 +402,7 @@ namespace MQTTnet.Server
packet = interceptingPacketEventArgs.Packet;
processPacket = interceptingPacketEventArgs.ProcessPacket;
}
if (!processPacket || packet == null)
{
// Restart the receiving process to get the next packet ignoring the current one..
@@ -213,15 +417,15 @@ namespace MQTTnet.Server
}
else if (packet is MqttPubAckPacket pubAckPacket)
{
Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier);
await HandleIncomingPubAckPacket(pubAckPacket).ConfigureAwait(false);
}
else if (packet is MqttPubCompPacket pubCompPacket)
{
Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier);
await HandleIncomingPubCompPacket(pubCompPacket).ConfigureAwait(false);
}
else if (packet is MqttPubRecPacket pubRecPacket)
{
HandleIncomingPubRecPacket(pubRecPacket);
await HandleIncomingPubRecPacket(pubRecPacket).ConfigureAwait(false);
}
else if (packet is MqttPubRelPacket pubRelPacket)
{
@@ -339,7 +543,7 @@ namespace MQTTnet.Server
StopInternal();
}
}
void StopInternal()
{
try
@@ -499,7 +703,7 @@ namespace MQTTnet.Server
try
{
var disconnectPacket = _packetFactories.Disconnect.Create(reasonCode);
using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
{
await SendPacketAsync(disconnectPacket, timeout.Token).ConfigureAwait(false);
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+5 -5
View File
@@ -603,25 +603,25 @@ namespace MQTTnet.Server
async Task<ValidatingConnectionEventArgs> ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
{
var context = new ValidatingConnectionEventArgs(connectPacket, channelAdapter)
var eventArgs = new ValidatingConnectionEventArgs(connectPacket, channelAdapter)
{
SessionItems = new ConcurrentDictionary<object, object>()
};
await _eventContainer.ValidatingConnectionEvent.InvokeAsync(context).ConfigureAwait(false);
await _eventContainer.ValidatingConnectionEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
// Check the client ID and set a random one if supported.
if (string.IsNullOrEmpty(connectPacket.ClientId) && channelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
{
connectPacket.ClientId = context.AssignedClientIdentifier;
connectPacket.ClientId = eventArgs.AssignedClientIdentifier;
}
if (string.IsNullOrEmpty(connectPacket.ClientId))
{
context.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid;
eventArgs.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid;
}
return context;
return eventArgs;
}
}
}
Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs
+32 -31
View File
@@ -9,40 +9,41 @@ namespace MQTTnet.Server
{
public sealed class MqttServerEventContainer
{
public AsyncEvent<InterceptingSubscriptionEventArgs> InterceptingSubscriptionEvent { get; } = new AsyncEvent<InterceptingSubscriptionEventArgs>();
public AsyncEvent<InterceptingUnsubscriptionEventArgs> InterceptingUnsubscriptionEvent { get; } = new AsyncEvent<InterceptingUnsubscriptionEventArgs>();
public AsyncEvent<InterceptingPublishEventArgs> InterceptingPublishEvent { get; } = new AsyncEvent<InterceptingPublishEventArgs>();
public AsyncEvent<ValidatingConnectionEventArgs> ValidatingConnectionEvent { get; } = new AsyncEvent<ValidatingConnectionEventArgs>();
public AsyncEvent<ClientConnectedEventArgs> ClientConnectedEvent { get; } = new AsyncEvent<ClientConnectedEventArgs>();
public AsyncEvent<ClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new AsyncEvent<ClientDisconnectedEventArgs>();
public AsyncEvent<ClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new AsyncEvent<ClientSubscribedTopicEventArgs>();
public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<ClientUnsubscribedTopicEventArgs>();
public AsyncEvent<EventArgs> PreparingSessionEvent { get; } = new AsyncEvent<EventArgs>();
public AsyncEvent<SessionDeletedEventArgs> SessionDeletedEvent { get; } = new AsyncEvent<SessionDeletedEventArgs>();
public AsyncEvent<ApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent<ApplicationMessageNotConsumedEventArgs>();
public AsyncEvent<RetainedMessageChangedEventArgs> RetainedMessageChangedEvent { get; } = new AsyncEvent<RetainedMessageChangedEventArgs>();
public AsyncEvent<LoadingRetainedMessagesEventArgs> LoadingRetainedMessagesEvent { get; } = new AsyncEvent<LoadingRetainedMessagesEventArgs>();
public AsyncEvent<EventArgs> RetainedMessagesClearedEvent { get; } = new AsyncEvent<EventArgs>();
public AsyncEvent<ClientAcknowledgedPublishPacketEventArgs> ClientAcknowledgedPublishPacketEvent { get; } = new AsyncEvent<ClientAcknowledgedPublishPacketEventArgs>();
public AsyncEvent<ClientConnectedEventArgs> ClientConnectedEvent { get; } = new AsyncEvent<ClientConnectedEventArgs>();
public AsyncEvent<ClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new AsyncEvent<ClientDisconnectedEventArgs>();
public AsyncEvent<ClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new AsyncEvent<ClientSubscribedTopicEventArgs>();
public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<ClientUnsubscribedTopicEventArgs>();
public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
public AsyncEvent<InterceptingPacketEventArgs> InterceptingOutboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
public AsyncEvent<InterceptingPublishEventArgs> InterceptingPublishEvent { get; } = new AsyncEvent<InterceptingPublishEventArgs>();
public AsyncEvent<InterceptingSubscriptionEventArgs> InterceptingSubscriptionEvent { get; } = new AsyncEvent<InterceptingSubscriptionEventArgs>();
public AsyncEvent<InterceptingUnsubscriptionEventArgs> InterceptingUnsubscriptionEvent { get; } = new AsyncEvent<InterceptingUnsubscriptionEventArgs>();
public AsyncEvent<LoadingRetainedMessagesEventArgs> LoadingRetainedMessagesEvent { get; } = new AsyncEvent<LoadingRetainedMessagesEventArgs>();
public AsyncEvent<EventArgs> PreparingSessionEvent { get; } = new AsyncEvent<EventArgs>();
public AsyncEvent<RetainedMessageChangedEventArgs> RetainedMessageChangedEvent { get; } = new AsyncEvent<RetainedMessageChangedEventArgs>();
public AsyncEvent<EventArgs> RetainedMessagesClearedEvent { get; } = new AsyncEvent<EventArgs>();
public AsyncEvent<SessionDeletedEventArgs> SessionDeletedEvent { get; } = new AsyncEvent<SessionDeletedEventArgs>();
public AsyncEvent<EventArgs> StartedEvent { get; } = new AsyncEvent<EventArgs>();
public AsyncEvent<EventArgs> StoppedEvent { get; } = new AsyncEvent<EventArgs>();
public AsyncEvent<ValidatingConnectionEventArgs> ValidatingConnectionEvent { get; } = new AsyncEvent<ValidatingConnectionEventArgs>();
}
}
}
Source/MQTTnet/Server/Internal/MqttSession.cs
+11 -2
View File
@@ -71,9 +71,18 @@ namespace MQTTnet.Server
public bool WillMessageSent { get; set; }
public void AcknowledgePublishPacket(ushort packetIdentifier)
public MqttPublishPacket PeekAcknowledgePublishPacket(ushort packetIdentifier)
{
_unacknowledgedPublishPackets.TryRemove(packetIdentifier, out _);
// This will only return the matching PUBLISH packet but does not remove it.
// This is required for QoS 2.
_unacknowledgedPublishPackets.TryGetValue(packetIdentifier, out var publishPacket);
return publishPacket;
}
public MqttPublishPacket AcknowledgePublishPacket(ushort packetIdentifier)
{
_unacknowledgedPublishPackets.TryRemove(packetIdentifier, out var publishPacket);
return publishPacket;
}
public void AddSubscribedTopic(string topic)
Source/MQTTnet/Server/MqttServer.cs
+6
View File
@@ -60,6 +60,12 @@ namespace MQTTnet.Server
add => _eventContainer.ClientConnectedEvent.AddHandler(value);
remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value);
}
public event Func<ClientAcknowledgedPublishPacketEventArgs, Task> ClientAcknowledgedPublishPacketAsync
{
add => _eventContainer.ClientAcknowledgedPublishPacketEvent.AddHandler(value);
remove => _eventContainer.ClientAcknowledgedPublishPacketEvent.RemoveHandler(value);
}
public event Func<ClientDisconnectedEventArgs, Task> ClientDisconnectedAsync
{