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 issue in processing of server messages.

Browse Source
This commit is contained in:
Christian Kratky
2018-07-24 13:49:23 +02:00
parent 773fd09166
commit 603639126e
1 changed files with 49 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/MqttClientSessionsManager.cs
+49 -34
View File
@@ -44,7 +44,7 @@ namespace MQTTnet.Server
public void Start() 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() public void Stop()
@@ -133,54 +133,69 @@ namespace MQTTnet.Server
_messageQueue?.Dispose(); _messageQueue?.Dispose();
} }
private void ProcessQueuedApplicationMessages(CancellationToken cancellationToken) private void TryProcessQueuedApplicationMessages(CancellationToken cancellationToken)
{ {
while (!cancellationToken.IsCancellationRequested) while (!cancellationToken.IsCancellationRequested)
{ {
try try
{ {
var enqueuedApplicationMessage = _messageQueue.Take(cancellationToken); TryProcessNextQueuedApplicationMessage(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 (OperationCanceledException)
{ {
} }
catch (Exception exception) 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() private List<MqttClientSession> GetSessions()
{ {
lock (_sessions) lock (_sessions)