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

Process all queued message before server stops (#1234)

Browse Source 
* process all queued message befor server stops

* fix an issue

Co-authored-by: ablfzl <ablfzl21@outlook.com>
This commit is contained in:
Ablfzl Edgelolli
2021-09-19 13:01:47 +02:00
committed by GitHub
co-authored by ablfzl
parent 68f1043ef5
commit dd0aba0640
1 changed files with 7 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+7 -11
View File
@@ -80,7 +80,7 @@ namespace MQTTnet.Server.Internal
_logger.Warning("Client '{0}': First received packet was no 'CONNECT' packet [MQTT-3.1.0-1].", channelAdapter.Endpoint); _logger.Warning("Client '{0}': First received packet was no 'CONNECT' packet [MQTT-3.1.0-1].", channelAdapter.Endpoint);
return null; return null;
} }
public async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken) public async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
{ {
MqttClientConnection clientConnection = null; MqttClientConnection clientConnection = null;
@@ -95,7 +95,7 @@ namespace MQTTnet.Server.Internal
} }
MqttConnAckPacket connAckPacket; MqttConnAckPacket connAckPacket;
var connectionValidatorContext = await ValidateConnection(connectPacket, channelAdapter).ConfigureAwait(false); var connectionValidatorContext = await ValidateConnection(connectPacket, channelAdapter).ConfigureAwait(false);
if (connectionValidatorContext.ReasonCode != MqttConnectReasonCode.Success) if (connectionValidatorContext.ReasonCode != MqttConnectReasonCode.Success)
{ {
@@ -272,11 +272,12 @@ namespace MQTTnet.Server.Internal
async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken) async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken)
{ {
while (!cancellationToken.IsCancellationRequested) //make sure all queued messages are proccessed befor server stops
while (!cancellationToken.IsCancellationRequested || _messageQueue.Any())
{ {
try try
{ {
await TryProcessNextQueuedApplicationMessageAsync(cancellationToken).ConfigureAwait(false); await TryProcessNextQueuedApplicationMessageAsync().ConfigureAwait(false);
} }
catch (OperationCanceledException) catch (OperationCanceledException)
{ {
@@ -288,16 +289,14 @@ namespace MQTTnet.Server.Internal
} }
} }
async Task TryProcessNextQueuedApplicationMessageAsync(CancellationToken cancellationToken) async Task TryProcessNextQueuedApplicationMessageAsync()
{ {
try try
{ {
cancellationToken.ThrowIfCancellationRequested();
MqttPendingApplicationMessage queuedApplicationMessage; MqttPendingApplicationMessage queuedApplicationMessage;
try try
{ {
queuedApplicationMessage = _messageQueue.Take(cancellationToken); queuedApplicationMessage = _messageQueue.Take();
} }
catch (ArgumentNullException) catch (ArgumentNullException)
{ {
@@ -370,9 +369,6 @@ namespace MQTTnet.Server.Internal
await InterceptApplicationMessageAsync(undeliveredMessageInterceptor, clientConnection, applicationMessage).ConfigureAwait(false); await InterceptApplicationMessageAsync(undeliveredMessageInterceptor, clientConnection, applicationMessage).ConfigureAwait(false);
} }
} }
catch (OperationCanceledException)
{
}
catch (Exception exception) catch (Exception exception)
{ {
_logger.Error(exception, "Unhandled exception while processing next queued application message."); _logger.Error(exception, "Unhandled exception while processing next queued application message.");