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

Fix session takeover.

Browse Source
This commit is contained in:
Christian Kratky
2018-12-27 23:05:57 +01:00
parent c0507fcc55
commit b527059b72
3 changed files with 28 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/MqttClientSession.cs
+22 -12
View File
@@ -30,7 +30,7 @@ namespace MQTTnet.Server
private MqttApplicationMessage _willMessage; private MqttApplicationMessage _willMessage;
private bool _wasCleanDisconnect; private bool _wasCleanDisconnect;
private Task _workerTask; private Task _workerTask;
private IDisposable _cleanupHandle; //private IDisposable _cleanupHandle;
private string _adapterEndpoint; private string _adapterEndpoint;
private MqttProtocolVersion? _adapterProtocolVersion; private MqttProtocolVersion? _adapterProtocolVersion;
@@ -73,11 +73,11 @@ namespace MQTTnet.Server
status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived; status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived;
} }
public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) //public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
{ //{
_workerTask = RunInternalAsync(connectPacket, adapter); // _workerTask = RunInternalAsync(connectPacket, adapter);
return _workerTask; // return _workerTask;
} //}
public Task StopAsync(MqttClientDisconnectType type) public Task StopAsync(MqttClientDisconnectType type)
{ {
@@ -121,16 +121,22 @@ namespace MQTTnet.Server
_cancellationTokenSource = null; _cancellationTokenSource = null;
} }
public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
{
_workerTask = RunInternalAsync(connectPacket, adapter);
return _workerTask;
}
private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
{ {
if (adapter == null) throw new ArgumentNullException(nameof(adapter)); if (adapter == null) throw new ArgumentNullException(nameof(adapter));
try try
{ {
if (_cancellationTokenSource != null) //if (_cancellationTokenSource != null)
{ //{
await StopAsync(MqttClientDisconnectType.Clean, true).ConfigureAwait(false); // await StopAsync(MqttClientDisconnectType.Clean, true).ConfigureAwait(false);
} //}
adapter.ReadingPacketStarted += OnAdapterReadingPacketStarted; adapter.ReadingPacketStarted += OnAdapterReadingPacketStarted;
adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted; adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted;
@@ -203,12 +209,14 @@ namespace MQTTnet.Server
//await TryDisconnectAdapterAsync(adapter).ConfigureAwait(false); //await TryDisconnectAdapterAsync(adapter).ConfigureAwait(false);
//TryDisposeAdapter(adapter); //TryDisposeAdapter(adapter);
_cleanupHandle?.Dispose(); //_cleanupHandle?.Dispose();
_cleanupHandle = null; //_cleanupHandle = null;
_cancellationTokenSource?.Cancel(false); _cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose(); _cancellationTokenSource?.Dispose();
_cancellationTokenSource = null; _cancellationTokenSource = null;
_workerTask = null;
} }
} }
@@ -264,6 +272,8 @@ namespace MQTTnet.Server
await _workerTask.ConfigureAwait(false); await _workerTask.ConfigureAwait(false);
} }
} }
await Task.FromResult(0);
} }
finally finally
{ {
Source/MQTTnet/Server/MqttClientSessionsManager.cs
+5 -5
View File
@@ -60,9 +60,9 @@ namespace MQTTnet.Server
} }
} }
public Task StartSession(IMqttChannelAdapter clientAdapter) public Task HandleConnectionAsync(IMqttChannelAdapter clientAdapter)
{ {
return Task.Run(() => RunSessionAsync(clientAdapter, _cancellationToken), _cancellationToken); return Task.Run(() => HandleConnectionAsync(clientAdapter, _cancellationToken), _cancellationToken);
} }
public async Task<IList<IMqttClientSessionStatus>> GetClientStatusAsync() public async Task<IList<IMqttClientSessionStatus>> GetClientStatusAsync()
@@ -207,7 +207,7 @@ namespace MQTTnet.Server
} }
} }
private async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken) private async Task HandleConnectionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
{ {
var clientId = string.Empty; var clientId = string.Empty;
@@ -295,11 +295,11 @@ namespace MQTTnet.Server
var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession); var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession);
if (isSessionPresent) if (isSessionPresent)
{ {
await clientSession.StopAsync(MqttClientDisconnectType.Clean).ConfigureAwait(false);
if (connectPacket.CleanSession) if (connectPacket.CleanSession)
{ {
_sessions.Remove(connectPacket.ClientId); _sessions.Remove(connectPacket.ClientId);
await clientSession.StopAsync(MqttClientDisconnectType.Clean).ConfigureAwait(false);
clientSession.Dispose(); clientSession.Dispose();
clientSession = null; clientSession = null;
Source/MQTTnet/Server/MqttServer.cs
+1 -1
View File
@@ -148,7 +148,7 @@ namespace MQTTnet.Server
private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs) private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
{ {
eventArgs.SessionTask = _clientSessionsManager.StartSession(eventArgs.Client); eventArgs.SessionTask = _clientSessionsManager.HandleConnectionAsync(eventArgs.Client);
} }
} }
} }