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

Merge branch 'develop' into client-cancel-task

Browse Source
This commit is contained in:
Christian
2018-09-28 22:49:00 +02:00
committed by GitHub
parent 400c0f05ad 682298b69f
commit 5c6bcdbca6
31 changed files with 557 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/IMqttClientSession.cs
+1 -1
View File
@@ -14,7 +14,7 @@ namespace MQTTnet.Server
void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket);
void ClearPendingApplicationMessages();
Task<bool> RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter);
Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter);
void Stop(MqttClientDisconnectType disconnectType);
Task SubscribeAsync(IList<TopicFilter> topicFilters);
Source/MQTTnet/Server/IMqttServerFactory.cs
+2
View File
@@ -8,6 +8,8 @@ namespace MQTTnet.Server
{
IMqttServer CreateMqttServer();
IMqttServer CreateMqttServer(IMqttNetLogger logger);
IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger);
}
}
Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs
+4 -4
View File
@@ -14,9 +14,9 @@ namespace MQTTnet.Server
private readonly IMqttClientSession _clientSession;
private readonly IMqttNetChildLogger _logger;
private bool _isPaused;
public MqttClientKeepAliveMonitor(IMqttClientSession clientSession, IMqttNetChildLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
@@ -76,7 +76,7 @@ namespace MQTTnet.Server
{
_logger.Warning(null, "Client '{0}': Did not receive any packet or keep alive signal.", _clientSession.ClientId);
_clientSession.Stop(MqttClientDisconnectType.NotClean);
return;
}
@@ -96,7 +96,7 @@ namespace MQTTnet.Server
}
finally
{
_logger.Verbose("Client {0}: Stopped checking keep alive timeout.", _clientSession.ClientId);
_logger.Verbose("Client '{0}': Stopped checking keep alive timeout.", _clientSession.ClientId);
}
}
}
Source/MQTTnet/Server/MqttClientSession.cs
+67 -15
View File
@@ -29,6 +29,8 @@ namespace MQTTnet.Server
private MqttApplicationMessage _willMessage;
private bool _wasCleanDisconnect;
private IMqttChannelAdapter _adapter;
private Task _workerTask;
private IDisposable _cleanupHandle;
public MqttClientSession(
string clientId,
@@ -65,7 +67,13 @@ namespace MQTTnet.Server
status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived;
}
public async Task<bool> RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
{
_workerTask = RunInternalAsync(connectPacket, adapter);
return _workerTask;
}
private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
{
if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket));
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -73,10 +81,18 @@ namespace MQTTnet.Server
try
{
_adapter = adapter;
adapter.ReadingPacketStarted += OnAdapterReadingPacketStarted;
adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted;
_cancellationTokenSource = new CancellationTokenSource();
//woraround for https://github.com/dotnet/corefx/issues/24430
#pragma warning disable 4014
_cleanupHandle = _cancellationTokenSource.Token.Register(() => CleanupAsync());
#pragma warning restore 4014
//endworkaround
_wasCleanDisconnect = false;
_willMessage = connectPacket.WillMessage;
@@ -114,26 +130,55 @@ namespace MQTTnet.Server
_logger.Error(exception, "Client '{0}': Unhandled exception while receiving client packets.", ClientId);
}
Stop(MqttClientDisconnectType.NotClean);
Stop(MqttClientDisconnectType.NotClean, true);
}
finally
{
if (_adapter != null)
{
_adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
_adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
}
await CleanupAsync().ConfigureAwait(false);
_cleanupHandle?.Dispose();
_cleanupHandle = null;
_adapter = null;
_cancellationTokenSource?.Cancel ();
_cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
}
}
return _wasCleanDisconnect;
private async Task CleanupAsync()
{
var adapter = _adapter;
try
{
if (adapter == null)
{
return;
}
_adapter = null;
adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.Error(exception, exception.Message);
}
finally
{
adapter?.Dispose();
}
}
public void Stop(MqttClientDisconnectType type)
{
Stop(type, false);
}
private void Stop(MqttClientDisconnectType type, bool isInsideSession)
{
try
{
@@ -153,10 +198,17 @@ namespace MQTTnet.Server
}
_willMessage = null;
if (!isInsideSession)
{
_workerTask?.GetAwaiter().GetResult();
}
}
finally
{
_logger.Info("Client '{0}': Session stopped.", ClientId);
_sessionsManager.Server.OnClientDisconnected(ClientId, _wasCleanDisconnect);
}
}
@@ -300,18 +352,18 @@ namespace MQTTnet.Server
if (packet is MqttDisconnectPacket)
{
Stop(MqttClientDisconnectType.Clean);
Stop(MqttClientDisconnectType.Clean, true);
return;
}
if (packet is MqttConnectPacket)
{
Stop(MqttClientDisconnectType.NotClean);
Stop(MqttClientDisconnectType.NotClean, true);
return;
}
_logger.Warning(null, "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
Stop(MqttClientDisconnectType.NotClean);
Stop(MqttClientDisconnectType.NotClean, true);
}
private void EnqueueSubscribedRetainedMessages(ICollection<TopicFilter> topicFilters)
@@ -330,7 +382,7 @@ namespace MQTTnet.Server
if (subscribeResult.CloseConnection)
{
Stop(MqttClientDisconnectType.NotClean);
Stop(MqttClientDisconnectType.NotClean, true);
return;
}
@@ -407,12 +459,12 @@ namespace MQTTnet.Server
private void OnAdapterReadingPacketCompleted(object sender, EventArgs e)
{
_keepAliveMonitor?.Pause();
_keepAliveMonitor?.Resume();
}
private void OnAdapterReadingPacketStarted(object sender, EventArgs e)
{
_keepAliveMonitor?.Resume();
_keepAliveMonitor?.Pause();
}
}
}
Source/MQTTnet/Server/MqttClientSessionsManager.cs
+52 -49
View File
@@ -44,7 +44,7 @@ namespace MQTTnet.Server
public void Start()
{
Task.Factory.StartNew(() => ProcessQueuedApplicationMessages(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
Task.Factory.StartNew(() => TryProcessQueuedApplicationMessages(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
}
public void Stop()
@@ -125,6 +125,7 @@ namespace MQTTnet.Server
{
_sessions.Remove(clientId);
}
_logger.Verbose("Session for client '{0}' deleted.", clientId);
}
@@ -133,54 +134,69 @@ namespace MQTTnet.Server
_messageQueue?.Dispose();
}
private void ProcessQueuedApplicationMessages(CancellationToken cancellationToken)
private void TryProcessQueuedApplicationMessages(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
try
{
var enqueuedApplicationMessage = _messageQueue.Take(cancellationToken);
var sender = enqueuedApplicationMessage.Sender;
var applicationMessage = enqueuedApplicationMessage.PublishPacket.ToApplicationMessage();
var interceptorContext = InterceptApplicationMessage(sender, applicationMessage);
if (interceptorContext != null)
{
if (interceptorContext.CloseConnection)
{
enqueuedApplicationMessage.Sender.Stop(MqttClientDisconnectType.NotClean);
}
if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
{
return;
}
applicationMessage = interceptorContext.ApplicationMessage;
}
Server.OnApplicationMessageReceived(sender?.ClientId, applicationMessage);
if (applicationMessage.Retain)
{
_retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).GetAwaiter().GetResult();
}
foreach (var clientSession in GetSessions())
{
clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage.ToPublishPacket());
}
TryProcessNextQueuedApplicationMessage(cancellationToken);
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
_logger.Error(exception, "Unhandled exception while processing queued application message.");
_logger.Error(exception, "Unhandled exception while processing queued application messages.");
}
}
}
private void TryProcessNextQueuedApplicationMessage(CancellationToken cancellationToken)
{
try
{
var enqueuedApplicationMessage = _messageQueue.Take(cancellationToken);
var sender = enqueuedApplicationMessage.Sender;
var applicationMessage = enqueuedApplicationMessage.PublishPacket.ToApplicationMessage();
var interceptorContext = InterceptApplicationMessage(sender, applicationMessage);
if (interceptorContext != null)
{
if (interceptorContext.CloseConnection)
{
enqueuedApplicationMessage.Sender.Stop(MqttClientDisconnectType.NotClean);
}
if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
{
return;
}
applicationMessage = interceptorContext.ApplicationMessage;
}
Server.OnApplicationMessageReceived(sender?.ClientId, applicationMessage);
if (applicationMessage.Retain)
{
_retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).GetAwaiter().GetResult();
}
foreach (var clientSession in GetSessions())
{
clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage.ToPublishPacket());
}
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
_logger.Error(exception, "Unhandled exception while processing next queued application message.");
}
}
private List<MqttClientSession> GetSessions()
{
lock (_sessions)
@@ -192,8 +208,7 @@ namespace MQTTnet.Server
private async Task RunSession(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
{
var clientId = string.Empty;
var wasCleanDisconnect = false;
try
{
var firstPacket = await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false);
@@ -238,7 +253,7 @@ namespace MQTTnet.Server
Server.OnClientConnected(clientId);
wasCleanDisconnect = await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
@@ -249,22 +264,10 @@ namespace MQTTnet.Server
}
finally
{
try
{
await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
clientAdapter.Dispose();
}
catch (Exception exception)
{
_logger.Error(exception, exception.Message);
}
if (!_options.EnablePersistentSessions)
{
DeleteSession(clientId);
}
Server.OnClientDisconnected(clientId, wasCleanDisconnect);
}
}
Source/MQTTnet/Server/MqttServer.cs
+2 -2
View File
@@ -106,14 +106,14 @@ namespace MQTTnet.Server
_cancellationTokenSource.Cancel(false);
_clientSessionsManager.Stop();
foreach (var adapter in _adapters)
{
adapter.ClientAccepted -= OnClientAccepted;
await adapter.StopAsync().ConfigureAwait(false);
}
_clientSessionsManager.Stop();
_logger.Info("Stopped.");
Stopped?.Invoke(this, EventArgs.Empty);
}