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

Fix broken connection takeover.

Browse Source
This commit is contained in:
Christian Kratky
2020-10-15 01:01:35 +02:00
parent 7be3685850
commit a26b641284
4 changed files with 101 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Client/MqttClient.cs
+16 -11
View File
@@ -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<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(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<MqttPingRespPacket>(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false);
@@ -566,7 +572,7 @@ namespace MQTTnet.Client
// Also dispatch disconnect to waiting threads to generate a proper exception.
_packetDispatcher.Dispatch(packet);
await DisconnectAsync(new MqttClientDisconnectOptions()
await DisconnectAsync(new MqttClientDisconnectOptions()
{
ReasonCode = (MqttClientDisconnectReason)(disconnectPacket.ReasonCode ?? MqttDisconnectReasonCode.UnspecifiedError)
}, cancellationToken).ConfigureAwait(false);
@@ -631,7 +637,6 @@ namespace MQTTnet.Client
try
{
var publishPacketDequeueResult = await _publishPacketReceiverQueue.TryDequeueAsync(cancellationToken);
if (!publishPacketDequeueResult.IsSuccess)
{
return;
Source/MQTTnet/Internal/AsyncQueue.cs
+13 -8
View File
@@ -7,16 +7,15 @@ namespace MQTTnet.Internal
{
public sealed class AsyncQueue<TItem> : IDisposable
{
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
private ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
SemaphoreSlim _semaphore = new SemaphoreSlim(0);
ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
public int Count => _queue.Count;
public void Enqueue(TItem item)
{
_queue.Enqueue(item);
_semaphore.Release();
_semaphore?.Release();
}
public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
@@ -25,12 +24,17 @@ namespace MQTTnet.Internal
{
try
{
if (_semaphore == null)
{
return new AsyncQueueDequeueResult<TItem>(false, default);
}
await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
}
catch (OperationCanceledException)
{
return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
return new AsyncQueueDequeueResult<TItem>(false, default);
}
if (_queue.TryDequeue(out var item))
@@ -39,7 +43,7 @@ namespace MQTTnet.Internal
}
}
return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
return new AsyncQueueDequeueResult<TItem>(false, default);
}
public AsyncQueueDequeueResult<TItem> TryDequeue()
@@ -49,7 +53,7 @@ namespace MQTTnet.Internal
return new AsyncQueueDequeueResult<TItem>(true, item);
}
return new AsyncQueueDequeueResult<TItem>(false, default(TItem));
return new AsyncQueueDequeueResult<TItem>(false, default);
}
public void Clear()
@@ -60,6 +64,7 @@ namespace MQTTnet.Internal
public void Dispose()
{
_semaphore?.Dispose();
_semaphore = null;
}
}
}
Source/MQTTnet/Server/MqttClientConnection.cs
+28 -28
View File
@@ -23,8 +23,6 @@ namespace MQTTnet.Server
readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource();
readonly IMqttRetainedMessagesManager _retainedMessagesManager;
readonly Func<Task> _onStart;
readonly Func<MqttClientDisconnectType, Task> _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<Task> onStart,
Func<MqttClientDisconnectType, Task> 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;
@@ -257,31 +250,38 @@ namespace MQTTnet.Server
Session.WillMessage = null;
}
if (_isTakeover)
{
// dont use SendAsync here _cancellationToken is already cancelled
await _channelAdapter.SendPacketAsync(new MqttDisconnectPacket()
{
ReasonCode = MqttDisconnectReasonCode.SessionTakenOver
}, TimeSpan.Zero, CancellationToken.None).ConfigureAwait(false);
}
_packetDispatcher.Reset();
_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);
}
}
}
Source/MQTTnet/Server/MqttClientSessionsManager.cs
+44 -48
View File
@@ -21,6 +21,7 @@ namespace MQTTnet.Server
readonly AsyncLock _createConnectionGate = new AsyncLock();
readonly ConcurrentDictionary<string, MqttClientConnection> _connections = new ConcurrentDictionary<string, MqttClientConnection>();
readonly ConcurrentDictionary<string, MqttClientSession> _sessions = new ConcurrentDictionary<string, MqttClientSession>();
readonly IDictionary<object, object> _serverSessionItems = new ConcurrentDictionary<object, object>();
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,7 +302,7 @@ 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)
{
@@ -320,11 +310,11 @@ namespace MQTTnet.Server
if (disconnectType != MqttClientDisconnectType.Takeover)
{
_connections.TryRemove(clientId, out _);
}
if (!_options.EnablePersistentSessions)
{
await DeleteSessionAsync(clientId).ConfigureAwait(false);
if (!_options.EnablePersistentSessions)
{
await DeleteSessionAsync(clientId).ConfigureAwait(false);
}
}
}
@@ -364,46 +354,52 @@ namespace MQTTnet.Server
return context;
}
async Task<MqttClientConnection> CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter, Func<Task> onStart, Func<MqttClientDisconnectType, Task> onStop)
async Task<MqttClientConnection> CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
{
MqttClientConnection connection;
MqttClientConnection existingConnection = null;
using (await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false))
{
var session = _sessions.AddOrUpdate(connectPacket.ClientId, key =>
{
_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)
var session = _sessions.AddOrUpdate(
connectPacket.ClientId,
key =>
{
_logger.Verbose("Deleting existing session of client '{0}'.", connectPacket.ClientId);
_logger.Verbose("Created a new session for client '{0}'.", key);
return new MqttClientSession(key, connectionValidatorContext.SessionItems, _eventDispatcher, _options, _retainedMessagesManager, _rootLogger);
}
else
},
(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;
}
});
});
var connection = new MqttClientConnection(connectPacket, channelAdapter, session, _options, this, _retainedMessagesManager, onStart, onStop, _rootLogger);
MqttClientConnection existingConnection = null;
_connections.AddOrUpdate(connectPacket.ClientId, key =>
{
return connection;
}, (key, tempExistingConnection) =>
{
existingConnection = tempExistingConnection;
return connection;
});
connection = new MqttClientConnection(connectPacket, channelAdapter, session, connectionValidatorContext, _options, this, _retainedMessagesManager, _rootLogger);
if (existingConnection != null)
{
await existingConnection.StopAsync(true).ConfigureAwait(false);
}
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<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)