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

Fix client dead lock and protocol version issues.

Browse Source
This commit is contained in:
Christian
2018-04-11 21:22:08 +02:00
parent c06d84e834
commit ec41efd860
22 changed files with 190 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Frameworks/MQTTnet.NetStandard/Server/MqttApplicationMessageInterceptorContext.cs
+4
View File
@@ -11,5 +11,9 @@
public string ClientId { get; }
public MqttApplicationMessage ApplicationMessage { get; set; }
public bool AcceptPublish { get; set; } = true;
public bool CloseConnection { get; set; }
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientDisconnectedEventArgs.cs
+4 -1
View File
@@ -4,11 +4,14 @@ namespace MQTTnet.Server
{
public class MqttClientDisconnectedEventArgs : EventArgs
{
public MqttClientDisconnectedEventArgs(ConnectedMqttClient client)
public MqttClientDisconnectedEventArgs(ConnectedMqttClient client, bool wasCleanDisconnect)
{
Client = client ?? throw new ArgumentNullException(nameof(client));
WasCleanDisconnect = wasCleanDisconnect;
}
public ConnectedMqttClient Client { get; }
public bool WasCleanDisconnect { get; }
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientKeepAliveMonitor.cs
+1 -1
View File
@@ -81,7 +81,7 @@ namespace MQTTnet.Server
}
finally
{
_logger.Trace<MqttClientSession>("Client {0}: Stopped checking keep alive timeout.", _clientId);
_logger.Verbose<MqttClientSession>("Client {0}: Stopped checking keep alive timeout.", _clientId);
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs
+2 -2
View File
@@ -57,7 +57,7 @@ namespace MQTTnet.Server
_queue.Enqueue(packet);
_queueWaitSemaphore.Release();
_logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
_logger.Verbose<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
}
private async Task SendQueuedPacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
@@ -96,7 +96,7 @@ namespace MQTTnet.Server
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { packet }).ConfigureAwait(false);
_logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
_logger.Verbose<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
}
catch (Exception exception)
{
Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs
+17 -6
View File
@@ -23,6 +23,7 @@ namespace MQTTnet.Server
private IMqttChannelAdapter _adapter;
private CancellationTokenSource _cancellationTokenSource;
private MqttApplicationMessage _willMessage;
private bool _wasCleanDisconnect;
public MqttClientSession(
string clientId,
@@ -55,7 +56,7 @@ namespace MQTTnet.Server
public bool IsConnected => _adapter != null;
public async Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
public async Task<bool> RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
{
if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket));
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -64,6 +65,7 @@ namespace MQTTnet.Server
{
var cancellationTokenSource = new CancellationTokenSource();
_wasCleanDisconnect = false;
_willMessage = connectPacket.WillMessage;
_adapter = adapter;
_cancellationTokenSource = cancellationTokenSource;
@@ -84,9 +86,11 @@ namespace MQTTnet.Server
{
_logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
}
return _wasCleanDisconnect;
}
public async Task StopAsync()
public async Task StopAsync(bool wasCleanDisconnect = false)
{
try
{
@@ -95,6 +99,8 @@ namespace MQTTnet.Server
return;
}
_wasCleanDisconnect = wasCleanDisconnect;
_cancellationTokenSource?.Cancel(false);
PendingMessagesQueue.WaitForCompletion();
@@ -110,9 +116,10 @@ namespace MQTTnet.Server
finally
{
var willMessage = _willMessage;
if (willMessage != null)
_willMessage = null; // clear willmessage so it is send just once
if (willMessage != null && !wasCleanDisconnect)
{
_willMessage = null; // clear willmessage so it is send just once
await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false);
}
}
@@ -246,7 +253,12 @@ namespace MQTTnet.Server
return HandleIncomingUnsubscribePacketAsync(adapter, unsubscribePacket, cancellationToken);
}
if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
if (packet is MqttDisconnectPacket)
{
return StopAsync(true);
}
if (packet is MqttConnectPacket)
{
return StopAsync();
}
@@ -262,7 +274,6 @@ namespace MQTTnet.Server
if (subscribeResult.CloseConnection)
{
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new[] { new MqttDisconnectPacket() }).ConfigureAwait(false);
await StopAsync().ConfigureAwait(false);
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs
+24 -16
View File
@@ -29,7 +29,7 @@ namespace MQTTnet.Server
}
public Action<ConnectedMqttClient> ClientConnectedCallback { get; set; }
public Action<ConnectedMqttClient> ClientDisconnectedCallback { get; set; }
public Action<ConnectedMqttClient, bool> ClientDisconnectedCallback { get; set; }
public Action<string, TopicFilter> ClientSubscribedTopicCallback { get; set; }
public Action<string, string> ClientUnsubscribedTopicCallback { get; set; }
public Action<string, MqttApplicationMessage> ApplicationMessageReceivedCallback { get; set; }
@@ -37,7 +37,9 @@ namespace MQTTnet.Server
public async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
{
var clientId = string.Empty;
var wasCleanDisconnect = false;
MqttClientSession clientSession = null;
try
{
if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken)
@@ -84,7 +86,7 @@ namespace MQTTnet.Server
ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
});
await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
wasCleanDisconnect = await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
@@ -110,7 +112,8 @@ namespace MQTTnet.Server
ClientId = clientId,
ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion,
PendingApplicationMessages = clientSession?.PendingMessagesQueue.Count ?? 0
});
},
wasCleanDisconnect);
}
}
@@ -156,8 +159,13 @@ namespace MQTTnet.Server
{
try
{
applicationMessage = InterceptApplicationMessage(senderClientSession, applicationMessage);
if (applicationMessage == null)
var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage);
if (interceptorContext.CloseConnection)
{
await senderClientSession.StopAsync().ConfigureAwait(false);
}
if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
{
return;
}
@@ -230,20 +238,20 @@ namespace MQTTnet.Server
}
}
private MqttApplicationMessage InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
private MqttApplicationMessageInterceptorContext InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
{
var interceptor = _options.ApplicationMessageInterceptor;
if (interceptor == null)
{
return applicationMessage;
}
var interceptorContext = new MqttApplicationMessageInterceptorContext(
senderClientSession?.ClientId,
applicationMessage);
var interceptor = _options.ApplicationMessageInterceptor;
if (interceptor == null)
{
return interceptorContext;
}
interceptor(interceptorContext);
return interceptorContext.ApplicationMessage;
return interceptorContext;
}
private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
@@ -279,11 +287,11 @@ namespace MQTTnet.Server
clientSession.Dispose();
clientSession = null;
_logger.Trace<MqttClientSessionsManager>("Stopped existing session of client '{0}'.", connectPacket.ClientId);
_logger.Verbose<MqttClientSessionsManager>("Stopped existing session of client '{0}'.", connectPacket.ClientId);
}
else
{
_logger.Trace<MqttClientSessionsManager>("Reusing existing session of client '{0}'.", connectPacket.ClientId);
_logger.Verbose<MqttClientSessionsManager>("Reusing existing session of client '{0}'.", connectPacket.ClientId);
}
}
@@ -302,7 +310,7 @@ namespace MQTTnet.Server
_sessions[connectPacket.ClientId] = clientSession;
_logger.Trace<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
_logger.Verbose<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
}
return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs
+1 -1
View File
@@ -132,7 +132,7 @@ namespace MQTTnet.Server
if (!saveIsRequired)
{
_logger.Trace<MqttRetainedMessagesManager>("Skipped saving retained messages because no changes were detected.");
_logger.Verbose<MqttRetainedMessagesManager>("Skipped saving retained messages because no changes were detected.");
}
if (saveIsRequired && _options.Storage != null)
Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs
+4 -4
View File
@@ -138,10 +138,10 @@ namespace MQTTnet.Server
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(client));
}
private void OnClientDisconnected(ConnectedMqttClient client)
private void OnClientDisconnected(ConnectedMqttClient client, bool wasCleanDisconnect)
{
_logger.Info<MqttServer>("Client '{0}': Disconnected.", client.ClientId);
ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client));
_logger.Info<MqttServer>("Client '{0}': Disconnected (clean={1}).", client.ClientId, wasCleanDisconnect);
ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client, wasCleanDisconnect));
}
private void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
@@ -162,7 +162,7 @@ namespace MQTTnet.Server
private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
{
eventArgs.SessionTask = Task.Run(
async () => await _clientSessionsManager.RunSessionAsync(eventArgs.Client, _cancellationTokenSource.Token).ConfigureAwait(false),
() => _clientSessionsManager.RunSessionAsync(eventArgs.Client, _cancellationTokenSource.Token),
_cancellationTokenSource.Token);
}
}