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

Merge branch 'develop' into ServerShutdown

Browse Source
This commit is contained in:
Christian
2018-07-26 14:45:25 +02:00
committed by GitHub
parent 043cfdb3f9 f70c79aaf9
commit ac36c9c406
16 changed files with 277 additions and 86 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/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
+6 -6
View File
@@ -67,13 +67,13 @@ namespace MQTTnet.Server
status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived;
}
public Task<bool> RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
{
_workerTask = RunInternalAsync(connectPacket, adapter);
return _workerTask;
}
private async Task<bool> RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
{
if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket));
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -138,8 +138,6 @@ namespace MQTTnet.Server
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
}
return _wasCleanDisconnect;
}
private async Task Cleanup()
@@ -199,6 +197,8 @@ namespace MQTTnet.Server
finally
{
_logger.Info("Client '{0}': Session stopped.", ClientId);
_sessionsManager.Server.OnClientDisconnected(ClientId, _wasCleanDisconnect);
}
}
@@ -447,12 +447,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 -39
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)
{
@@ -253,8 +268,6 @@ namespace MQTTnet.Server
{
DeleteSession(clientId);
}
Server.OnClientDisconnected(clientId, wasCleanDisconnect);
}
}