Browse Source

Fix cleanup of client sessions

release/3.x.x
Christian Kratky 7 years ago
parent
commit
7e48746944
4 changed files with 6 additions and 13 deletions
  1. +2
    -2
      MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs
  2. +0
    -1
      MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs
  3. +2
    -8
      MQTTnet.Core/Server/MqttClientSession.cs
  4. +2
    -2
      MQTTnet.Core/Server/MqttClientSessionsManager.cs

+ 2
- 2
MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs View File

@@ -14,10 +14,10 @@ namespace MQTTnet.Core.Adapter
{ {
public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter
{ {
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly ILogger<MqttChannelCommunicationAdapter> _logger; private readonly ILogger<MqttChannelCommunicationAdapter> _logger;
private readonly IMqttCommunicationChannel _channel; private readonly IMqttCommunicationChannel _channel;
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);

public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer, ILogger<MqttChannelCommunicationAdapter> logger) public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer, ILogger<MqttChannelCommunicationAdapter> logger)
{ {
_logger = logger ?? throw new ArgumentNullException(nameof(logger)); _logger = logger ?? throw new ArgumentNullException(nameof(logger));


+ 0
- 1
MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs View File

@@ -68,7 +68,6 @@ namespace MQTTnet.Core.Server


try try
{ {
// ReSharper disable once UseNullPropagation
if (_options.Storage != null) if (_options.Storage != null)
{ {
await _options.Storage.SaveRetainedMessagesAsync(allRetainedMessages); await _options.Storage.SaveRetainedMessagesAsync(allRetainedMessages);


+ 2
- 8
MQTTnet.Core/Server/MqttClientSession.cs View File

@@ -13,11 +13,11 @@ using Microsoft.Extensions.Options;


namespace MQTTnet.Core.Server namespace MQTTnet.Core.Server
{ {
public sealed class MqttClientSession : IDisposable
public sealed class MqttClientSession
{ {
private readonly IMqttClientRetainedMessageManager _clientRetainedMessageManager;
private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>(); private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();


private readonly IMqttClientRetainedMessageManager _clientRetainedMessageManager;
private readonly MqttClientSubscriptionsManager _subscriptionsManager; private readonly MqttClientSubscriptionsManager _subscriptionsManager;
private readonly MqttClientSessionsManager _sessionsManager; private readonly MqttClientSessionsManager _sessionsManager;
private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue; private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
@@ -113,12 +113,6 @@ namespace MQTTnet.Core.Server
_pendingMessagesQueue.Enqueue(publishPacket); _pendingMessagesQueue.Enqueue(publishPacket);
} }


public void Dispose()
{
_cancellationTokenSource?.Cancel();
_cancellationTokenSource?.Dispose();
}

private async Task ReceivePacketsAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken) private async Task ReceivePacketsAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
{ {
try try


+ 2
- 2
MQTTnet.Core/Server/MqttClientSessionsManager.cs View File

@@ -164,10 +164,10 @@ namespace MQTTnet.Core.Server
if (connectPacket.CleanSession) if (connectPacket.CleanSession)
{ {
_clientSessions.Remove(connectPacket.ClientId); _clientSessions.Remove(connectPacket.ClientId);
clientSession.Dispose();
clientSession.Stop();
clientSession = null; clientSession = null;


_logger.LogTrace("Disposed existing session of client '{0}'.", connectPacket.ClientId);
_logger.LogTrace("Stopped existing session of client '{0}'.", connectPacket.ClientId);
} }
else else
{ {


Loading…
Cancel
Save