diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec
index d8558191..c15d5321 100644
--- a/Build/MQTTnet.nuspec
+++ b/Build/MQTTnet.nuspec
@@ -12,10 +12,7 @@
true
MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.
-* [Client] Fixed wrong value for "ClientWasConnected" in "MqttClientDisconnectedEventArgs" #976 (thanks to @dbeinder).
-* [Client] Added direct support for Amazon AWS connections (requires .NET Core 3.1) (thanks to @henning-krause).
-* [ManagedClient] Exposed the internal MQTT client.
-* [Server] Added client message queue interceptor for QoS level (tahnks to @msallin).
+* [Server] ..
Copyright Christian Kratky 2016-2020
MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor
diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs
index 8a045bed..cb87120a 100644
--- a/Source/MQTTnet/Client/MqttClient.cs
+++ b/Source/MQTTnet/Client/MqttClient.cs
@@ -245,16 +245,22 @@ namespace MQTTnet.Client
void Cleanup()
{
- _backgroundCancellationTokenSource?.Cancel(false);
- _backgroundCancellationTokenSource?.Dispose();
- _backgroundCancellationTokenSource = null;
+ try
+ {
+ _backgroundCancellationTokenSource?.Cancel(false);
+ }
+ finally
+ {
+ _backgroundCancellationTokenSource?.Dispose();
+ _backgroundCancellationTokenSource = null;
- _publishPacketReceiverQueue?.Dispose();
- _publishPacketReceiverQueue = null;
+ _publishPacketReceiverQueue?.Dispose();
+ _publishPacketReceiverQueue = null;
- _adapter?.Dispose();
+ _adapter?.Dispose();
+ }
}
-
+
protected override void Dispose(bool disposing)
{
if (disposing)
@@ -387,7 +393,7 @@ namespace MQTTnet.Client
async Task SendAndReceiveAsync(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket
{
cancellationToken.ThrowIfCancellationRequested();
-
+
ushort identifier = 0;
if (requestPacket is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier.HasValue)
{
@@ -436,7 +442,7 @@ namespace MQTTnet.Client
{
// Values described here: [MQTT-3.1.2-24].
var waitTime = keepAlivePeriod - _sendTracker.Elapsed;
-
+
if (waitTime <= TimeSpan.Zero)
{
await SendAndReceiveAsync(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false);
@@ -561,12 +567,15 @@ namespace MQTTnet.Client
{
await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false);
}
- else if (packet is MqttDisconnectPacket)
+ else if (packet is MqttDisconnectPacket disconnectPacket)
{
// Also dispatch disconnect to waiting threads to generate a proper exception.
_packetDispatcher.Dispatch(packet);
- await DisconnectAsync(null, cancellationToken).ConfigureAwait(false);
+ await DisconnectAsync(new MqttClientDisconnectOptions()
+ {
+ ReasonCode = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.UnspecifiedError)
+ }, cancellationToken).ConfigureAwait(false);
}
else if (packet is MqttAuthPacket authPacket)
{
@@ -628,7 +637,6 @@ namespace MQTTnet.Client
try
{
var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken);
-
if (!publishPacketDequeueResult.IsSuccess)
{
return;
diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs
index 5d96740d..8a349ecc 100644
--- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs
+++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs
@@ -87,7 +87,7 @@ namespace MQTTnet.Implementations
ApplicationProtocols = _options.TlsOptions.ApplicationProtocols,
ClientCertificates = LoadCertificates(),
EnabledSslProtocols = _options.TlsOptions.SslProtocol,
- CertificateRevocationCheckMode = _options.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.Online : X509RevocationMode.NoCheck,
+ CertificateRevocationCheckMode = _options.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : X509RevocationMode.Online,
TargetHost = _options.Server
};
diff --git a/Source/MQTTnet/Internal/AsyncQueue.cs b/Source/MQTTnet/Internal/AsyncQueue.cs
index 43ad938a..868ed617 100644
--- a/Source/MQTTnet/Internal/AsyncQueue.cs
+++ b/Source/MQTTnet/Internal/AsyncQueue.cs
@@ -7,16 +7,15 @@ namespace MQTTnet.Internal
{
public sealed class AsyncQueue : IDisposable
{
- private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
-
- private ConcurrentQueue _queue = new ConcurrentQueue();
-
+ SemaphoreSlim _semaphore = new SemaphoreSlim(0);
+ ConcurrentQueue _queue = new ConcurrentQueue();
+
public int Count => _queue.Count;
public void Enqueue(TItem item)
{
_queue.Enqueue(item);
- _semaphore.Release();
+ _semaphore?.Release();
}
public async Task> TryDequeueAsync(CancellationToken cancellationToken)
@@ -25,12 +24,17 @@ namespace MQTTnet.Internal
{
try
{
+ if (_semaphore == null)
+ {
+ return new AsyncQueueDequeueResult(false, default);
+ }
+
await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
}
catch (OperationCanceledException)
{
- return new AsyncQueueDequeueResult(false, default(TItem));
+ return new AsyncQueueDequeueResult(false, default);
}
if (_queue.TryDequeue(out var item))
@@ -39,7 +43,7 @@ namespace MQTTnet.Internal
}
}
- return new AsyncQueueDequeueResult(false, default(TItem));
+ return new AsyncQueueDequeueResult(false, default);
}
public AsyncQueueDequeueResult TryDequeue()
@@ -49,7 +53,7 @@ namespace MQTTnet.Internal
return new AsyncQueueDequeueResult(true, item);
}
- return new AsyncQueueDequeueResult(false, default(TItem));
+ return new AsyncQueueDequeueResult(false, default);
}
public void Clear()
@@ -60,6 +64,7 @@ namespace MQTTnet.Internal
public void Dispose()
{
_semaphore?.Dispose();
+ _semaphore = null;
}
}
}
diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs
index bf51269c..d501c8eb 100644
--- a/Source/MQTTnet/Server/MqttClientConnection.cs
+++ b/Source/MQTTnet/Server/MqttClientConnection.cs
@@ -23,8 +23,6 @@ namespace MQTTnet.Server
readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource();
readonly IMqttRetainedMessagesManager _retainedMessagesManager;
- readonly Func _onStart;
- readonly Func _onStop;
readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
readonly MqttClientSessionsManager _sessionsManager;
@@ -32,6 +30,7 @@ namespace MQTTnet.Server
readonly IMqttServerOptions _serverOptions;
readonly IMqttChannelAdapter _channelAdapter;
+ readonly MqttConnectionValidatorContext _connectionValidatorContext;
readonly IMqttDataConverter _dataConverter;
readonly string _endpoint;
readonly DateTime _connectedTimestamp;
@@ -47,25 +46,22 @@ namespace MQTTnet.Server
volatile bool _isTakeover;
- public MqttClientConnection(
- MqttConnectPacket connectPacket,
+ public MqttClientConnection(MqttConnectPacket connectPacket,
IMqttChannelAdapter channelAdapter,
MqttClientSession session,
+ MqttConnectionValidatorContext connectionValidatorContext,
IMqttServerOptions serverOptions,
MqttClientSessionsManager sessionsManager,
IMqttRetainedMessagesManager retainedMessagesManager,
- Func onStart,
- Func onStop,
IMqttNetLogger logger)
{
Session = session ?? throw new ArgumentNullException(nameof(session));
_serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
_sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
_retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
- _onStart = onStart ?? throw new ArgumentNullException(nameof(onStart));
- _onStop = onStop ?? throw new ArgumentNullException(nameof(onStop));
_channelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
+ _connectionValidatorContext = connectionValidatorContext ?? throw new ArgumentNullException(nameof(connectionValidatorContext));
_dataConverter = _channelAdapter.PacketFormatterAdapter.DataConverter;
_endpoint = _channelAdapter.Endpoint;
ConnectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
@@ -131,18 +127,17 @@ namespace MQTTnet.Server
_cancellationToken.Dispose();
}
- public Task RunAsync(MqttConnectionValidatorContext connectionValidatorContext)
+ public Task RunAsync()
{
- _packageReceiverTask = RunInternalAsync(connectionValidatorContext);
+ _packageReceiverTask = RunInternalAsync();
return _packageReceiverTask;
}
- async Task RunInternalAsync(MqttConnectionValidatorContext connectionValidatorContext)
+ async Task RunInternalAsync()
{
var disconnectType = MqttClientDisconnectType.NotClean;
try
{
- await _onStart();
_logger.Info("Client '{0}': Session started.", ClientId);
_channelAdapter.ReadingPacketStartedCallback = OnAdapterReadingPacketStarted;
@@ -155,9 +150,7 @@ namespace MQTTnet.Server
// TODO: Change to single thread in SessionManager. Or use SessionManager and stats from KeepAliveMonitor.
_keepAliveMonitor.Start(ConnectPacket.KeepAlivePeriod, _cancellationToken.Token);
- await SendAsync(
- _channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext)
- ).ConfigureAwait(false);
+ await SendAsync(_channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(_connectionValidatorContext)).ConfigureAwait(false);
Session.IsCleanSession = false;
@@ -262,17 +255,33 @@ namespace MQTTnet.Server
_channelAdapter.ReadingPacketStartedCallback = null;
_channelAdapter.ReadingPacketCompletedCallback = null;
- _logger.Info("Client '{0}': Connection stopped.", ClientId);
-
_packageReceiverTask = null;
+ if (_isTakeover)
+ {
+ try
+ {
+ // Don't use SendAsync here _cancellationToken is already cancelled.
+ await _channelAdapter.SendPacketAsync(new MqttDisconnectPacket
+ {
+ ReasonCode = MqttDisconnectReasonCode.SessionTakenOver
+ }, TimeSpan.Zero, CancellationToken.None).ConfigureAwait(false);
+ }
+ catch (Exception exception)
+ {
+ _logger.Error(exception, "Client '{0}': Error while sending DISCONNECT packet after takeover.", ClientId);
+ }
+ }
+
+ _logger.Info("Client '{0}': Connection stopped.", ClientId);
+
try
{
- await _onStop(disconnectType);
+ await _sessionsManager.CleanUpClient(ClientId, _channelAdapter, disconnectType);
}
catch (Exception e)
{
- _logger.Error(e, "client '{0}': Error while cleaning up", ClientId);
+ _logger.Error(e, "Client '{0}': Error while cleaning up", ClientId);
}
}
}
diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs
index 48f69809..19c8478d 100644
--- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs
+++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs
@@ -21,6 +21,7 @@ namespace MQTTnet.Server
readonly AsyncLock _createConnectionGate = new AsyncLock();
readonly ConcurrentDictionary _connections = new ConcurrentDictionary();
readonly ConcurrentDictionary _sessions = new ConcurrentDictionary();
+
readonly IDictionary _serverSessionItems = new ConcurrentDictionary();
readonly CancellationToken _cancellationToken;
@@ -183,7 +184,6 @@ namespace MQTTnet.Server
}
var dequeueResult = await _messageQueue.TryDequeueAsync(cancellationToken).ConfigureAwait(false);
-
if (!dequeueResult.IsSuccess)
{
return;
@@ -258,8 +258,6 @@ namespace MQTTnet.Server
async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
{
- string clientId = null;
-
try
{
MqttConnectPacket connectPacket;
@@ -291,17 +289,9 @@ namespace MQTTnet.Server
return;
}
- clientId = connectPacket.ClientId;
-
- var connection = await CreateClientConnectionAsync(
- connectPacket,
- connectionValidatorContext,
- channelAdapter,
- async () => await _eventDispatcher.SafeNotifyClientConnectedAsync(clientId).ConfigureAwait(false),
- async disconnectType => await CleanUpClient(clientId, channelAdapter, disconnectType)
- ).ConfigureAwait(false);
-
- await connection.RunAsync(connectionValidatorContext).ConfigureAwait(false);
+ var connection = await CreateClientConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false);
+ await _eventDispatcher.SafeNotifyClientConnectedAsync(connectPacket.ClientId).ConfigureAwait(false);
+ await connection.RunAsync().ConfigureAwait(false);
}
catch (OperationCanceledException)
{
@@ -312,15 +302,19 @@ namespace MQTTnet.Server
}
}
- private async Task CleanUpClient(string clientId, IMqttChannelAdapter channelAdapter, MqttClientDisconnectType disconnectType)
+ public async Task CleanUpClient(string clientId, IMqttChannelAdapter channelAdapter, MqttClientDisconnectType disconnectType)
{
if (clientId != null)
{
- _connections.TryRemove(clientId, out _);
-
- if (!_options.EnablePersistentSessions)
+ // in case it is a takeover _connections already contains the new connection
+ if (disconnectType != MqttClientDisconnectType.Takeover)
{
- await DeleteSessionAsync(clientId).ConfigureAwait(false);
+ _connections.TryRemove(clientId, out _);
+
+ if (!_options.EnablePersistentSessions)
+ {
+ await DeleteSessionAsync(clientId).ConfigureAwait(false);
+ }
}
}
@@ -360,45 +354,52 @@ namespace MQTTnet.Server
return context;
}
- async Task CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter, Func onStart, Func onStop)
+ async Task CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
{
+ MqttClientConnection connection;
+ MqttClientConnection existingConnection = null;
+
using (await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false))
{
- var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var session);
-
- var isConnectionPresent = _connections.TryGetValue(connectPacket.ClientId, out var existingConnection);
- if (isConnectionPresent)
- {
- await existingConnection.StopAsync(true).ConfigureAwait(false);
- }
-
- if (isSessionPresent)
- {
- if (connectPacket.CleanSession)
+ var session = _sessions.AddOrUpdate(
+ connectPacket.ClientId,
+ key =>
{
- session = null;
-
- _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
- }
- else
+ _logger.Verbose("Created a new session for client '{0}'.", key);
+ return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
+ },
+ (key, existingSession) =>
{
+ if (connectPacket.CleanSession)
+ {
+ _logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
+ return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
+ }
+
_logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
- }
- }
+ return existingSession;
+ });
- if (session == null)
- {
- session = new MqttClientSession(connectPacket.ClientId, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
- _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
- }
+ connection = new MqttClientConnection(connectPacket, channelAdapter, session, connectionValidatorContext, _options, this, _retainedMessagesManager, _rootLogger);
- var connection = new MqttClientConnection(connectPacket, channelAdapter, session, _options, this, _retainedMessagesManager, onStart, onStop, _rootLogger);
-
- _connections[connection.ClientId] = connection;
- _sessions[session.ClientId] = session;
-
- return connection;
+ _connections.AddOrUpdate(
+ connectPacket.ClientId,
+ key => connection,
+ (key, tempExistingConnection) =>
+ {
+ existingConnection = tempExistingConnection;
+ return connection;
+ });
}
+
+ // Disconnect the client outside of the lock so that new clients can still connect while a single
+ // one is being disconnected.
+ if (existingConnection != null)
+ {
+ await existingConnection.StopAsync(true).ConfigureAwait(false);
+ }
+
+ return connection;
}
async Task InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)
diff --git a/Tests/MQTTnet.Core.Tests/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Session_Tests.cs
index 973fbf3e..1f4c418d 100644
--- a/Tests/MQTTnet.Core.Tests/Session_Tests.cs
+++ b/Tests/MQTTnet.Core.Tests/Session_Tests.cs
@@ -1,5 +1,6 @@
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
+using MQTTnet.Client.Options;
using MQTTnet.Client.Subscribing;
using MQTTnet.Server;
using MQTTnet.Tests.Mockups;
@@ -84,5 +85,40 @@ namespace MQTTnet.Tests
Assert.AreEqual(true, session.Items["can_subscribe_x"]);
}
}
+
+
+ [TestMethod]
+ public async Task Manage_Session_MaxParallel()
+ {
+ using (var testEnvironment = new TestEnvironment(TestContext))
+ {
+ testEnvironment.IgnoreClientLogErrors = true;
+ var serverOptions = new MqttServerOptionsBuilder();
+ await testEnvironment.StartServerAsync(serverOptions);
+
+ var options = new MqttClientOptionsBuilder()
+ .WithClientId("1")
+ ;
+
+ var clients = await Task.WhenAll(Enumerable.Range(0, 10)
+ .Select(i => TryConnect(testEnvironment, options)));
+
+ var connectedClients = clients.Where(c => c?.IsConnected ?? false).ToList();
+
+ Assert.AreEqual(1, connectedClients.Count);
+ }
+ }
+
+ private async Task TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options)
+ {
+ try
+ {
+ return await testEnvironment.ConnectClientAsync(options);
+ }
+ catch (System.Exception)
+ {
+ return null;
+ }
+ }
}
}