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

Refactor interceptors etc. to support async.

Browse Source
This commit is contained in:
Christian Kratky
2018-12-27 22:41:41 +01:00
parent 38f2453c92
commit c0507fcc55
33 changed files with 537 additions and 311 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/IMqttClientSession.cs
+2 -1
View File
@@ -1,4 +1,5 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
@@ -6,6 +7,6 @@ namespace MQTTnet.Server
{
string ClientId { get; }
void Stop(MqttClientDisconnectType disconnectType);
Task StopAsync(MqttClientDisconnectType disconnectType);
}
}
Source/MQTTnet/Server/IMqttServer.cs
-3
View File
@@ -16,11 +16,8 @@ namespace MQTTnet.Server
IMqttServerOptions Options { get; }
[Obsolete("This method is no longer async. Use the not async method.")]
Task<IList<IMqttClientSessionStatus>> GetClientSessionsStatusAsync();
IList<IMqttClientSessionStatus> GetClientSessionsStatus();
IList<MqttApplicationMessage> GetRetainedMessages();
Task ClearRetainedMessagesAsync();
Source/MQTTnet/Server/IMqttServerApplicationMessageInterceptor.cs
+9
View File
@@ -0,0 +1,9 @@
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerApplicationMessageInterceptor
{
Task InterceptApplicationMessagePublishAsync(MqttApplicationMessageInterceptorContext context);
}
}
Source/MQTTnet/Server/IMqttServerClientMessageQueueInterceptor.cs
+9
View File
@@ -0,0 +1,9 @@
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerClientMessageQueueInterceptor
{
Task InterceptClientMessageQueueEnqueueAsync(MqttClientMessageQueueInterceptorContext context);
}
}
Source/MQTTnet/Server/IMqttServerConnectionValidator.cs
+9
View File
@@ -0,0 +1,9 @@
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerConnectionValidator
{
Task ValidateConnection(MqttConnectionValidatorContext context);
}
}
Source/MQTTnet/Server/IMqttServerOptions.cs
+4 -4
View File
@@ -11,10 +11,10 @@ namespace MQTTnet.Server
TimeSpan DefaultCommunicationTimeout { get; }
Action<MqttConnectionValidatorContext> ConnectionValidator { get; }
Action<MqttSubscriptionInterceptorContext> SubscriptionInterceptor { get; }
Action<MqttApplicationMessageInterceptorContext> ApplicationMessageInterceptor { get; }
Action<MqttClientMessageQueueInterceptorContext> ClientMessageQueueInterceptor { get; }
IMqttServerConnectionValidator ConnectionValidator { get; }
IMqttServerSubscriptionInterceptor SubscriptionInterceptor { get; }
IMqttServerApplicationMessageInterceptor ApplicationMessageInterceptor { get; }
IMqttServerClientMessageQueueInterceptor ClientMessageQueueInterceptor { get; }
MqttServerTcpEndpointOptions DefaultEndpointOptions { get; }
MqttServerTlsTcpEndpointOptions TlsEndpointOptions { get; }
Source/MQTTnet/Server/IMqttServerSubscriptionInterceptor.cs
+9
View File
@@ -0,0 +1,9 @@
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerSubscriptionInterceptor
{
Task InterceptSubscriptionAsync(MqttSubscriptionInterceptorContext context);
}
}
Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs
+1 -1
View File
@@ -81,7 +81,7 @@ namespace MQTTnet.Server
if (!_isPaused && _lastPacketReceivedTracker.Elapsed.TotalSeconds >= keepAlivePeriod * 1.5D)
{
_logger.Warning(null, "Client '{0}': Did not receive any packet or keep alive signal.", _clientSession.ClientId);
_clientSession.Stop(MqttClientDisconnectType.NotClean);
await _clientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
return;
}
Source/MQTTnet/Server/MqttClientPendingPacketsQueue.cs
+2 -2
View File
@@ -133,7 +133,7 @@ namespace MQTTnet.Server
return;
}
adapter.SendPacketAsync(packet, cancellationToken).GetAwaiter().GetResult();
await adapter.SendPacketAsync(packet, cancellationToken).ConfigureAwait(false);
_logger.Verbose("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
}
@@ -167,7 +167,7 @@ namespace MQTTnet.Server
if (!cancellationToken.IsCancellationRequested)
{
_clientSession.Stop(MqttClientDisconnectType.NotClean);
await _clientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
}
}
}
Source/MQTTnet/Server/MqttClientSession.cs
+102 -105
View File
@@ -79,22 +79,20 @@ namespace MQTTnet.Server
return _workerTask;
}
public void Stop(MqttClientDisconnectType type)
public Task StopAsync(MqttClientDisconnectType type)
{
Stop(type, false);
return StopAsync(type, false);
}
public Task SubscribeAsync(IList<TopicFilter> topicFilters)
public async Task SubscribeAsync(IList<TopicFilter> topicFilters)
{
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
var packet = new MqttSubscribePacket();
packet.TopicFilters.AddRange(topicFilters);
_subscriptionsManager.Subscribe(packet);
EnqueueSubscribedRetainedMessages(topicFilters);
return Task.FromResult(0);
await _subscriptionsManager.SubscribeAsync(packet).ConfigureAwait(false);
await EnqueueSubscribedRetainedMessagesAsync(topicFilters).ConfigureAwait(false);
}
public Task UnsubscribeAsync(IList<string> topicFilters)
@@ -122,7 +120,7 @@ namespace MQTTnet.Server
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
}
private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter)
{
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -131,7 +129,7 @@ namespace MQTTnet.Server
{
if (_cancellationTokenSource != null)
{
Stop(MqttClientDisconnectType.Clean, true);
await StopAsync(MqttClientDisconnectType.Clean, true).ConfigureAwait(false);
}
adapter.ReadingPacketStarted += OnAdapterReadingPacketStarted;
@@ -139,15 +137,15 @@ namespace MQTTnet.Server
_cancellationTokenSource = new CancellationTokenSource();
//workaround for https://github.com/dotnet/corefx/issues/24430
#pragma warning disable 4014
_cleanupHandle = _cancellationTokenSource.Token.Register(async () =>
{
await TryDisconnectAdapterAsync(adapter).ConfigureAwait(false);
TryDisposeAdapter(adapter);
});
#pragma warning restore 4014
//end workaround
// //workaround for https://github.com/dotnet/corefx/issues/24430
//#pragma warning disable 4014
// _cleanupHandle = _cancellationTokenSource.Token.Register(async () =>
// {
// await TryDisconnectAdapterAsync(adapter).ConfigureAwait(false);
// TryDisposeAdapter(adapter);
// });
//#pragma warning restore 4014
// //end workaround
_wasCleanDisconnect = false;
_willMessage = connectPacket.WillMessage;
@@ -165,7 +163,7 @@ namespace MQTTnet.Server
if (packet != null)
{
_keepAliveMonitor.PacketReceived(packet);
ProcessReceivedPacket(adapter, packet, _cancellationTokenSource.Token);
await ProcessReceivedPacketAsync(adapter, packet, _cancellationTokenSource.Token).ConfigureAwait(false);
}
}
}
@@ -190,51 +188,55 @@ namespace MQTTnet.Server
_logger.Error(exception, "Client '{0}': Unhandled exception while receiving client packets.", ClientId);
}
Stop(MqttClientDisconnectType.NotClean, true);
await StopAsync(MqttClientDisconnectType.NotClean, true).ConfigureAwait(false);
}
finally
{
adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
_adapterEndpoint = null;
_adapterProtocolVersion = null;
// Uncomment as soon as the workaround above is no longer needed.
// Also called in outer scope!
//await TryDisconnectAdapterAsync(adapter).ConfigureAwait(false);
//TryDisposeAdapter(adapter);
_cleanupHandle?.Dispose();
_cleanupHandle = null;
_cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
}
}
private void TryDisposeAdapter(IMqttChannelAdapter adapter)
{
if (adapter == null)
{
return;
}
////private void TryDisposeAdapter(IMqttChannelAdapter adapter)
////{
//// if (adapter == null)
//// {
//// return;
//// }
try
{
adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
//// try
//// {
//// adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
//// adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
adapter.Dispose();
}
catch (Exception exception)
{
_logger.Error(exception, exception.Message);
}
finally
{
adapter.Dispose();
}
}
//// adapter.Dispose();
//// }
//// catch (Exception exception)
//// {
//// _logger.Error(exception, exception.Message);
//// }
//// finally
//// {
//// adapter.Dispose();
//// }
////}
private void Stop(MqttClientDisconnectType type, bool isInsideSession)
private async Task StopAsync(MqttClientDisconnectType type, bool isInsideSession)
{
try
{
@@ -257,7 +259,10 @@ namespace MQTTnet.Server
if (!isInsideSession)
{
_workerTask?.GetAwaiter().GetResult();
if (_workerTask != null)
{
await _workerTask.ConfigureAwait(false);
}
}
}
finally
@@ -267,7 +272,7 @@ namespace MQTTnet.Server
}
}
public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage, bool isRetainedApplicationMessage)
public async Task EnqueueApplicationMessageAsync(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage, bool isRetainedApplicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
@@ -278,10 +283,10 @@ namespace MQTTnet.Server
}
var publishPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(applicationMessage);
// Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9].
publishPacket.Retain = isRetainedApplicationMessage;
if (publishPacket.QualityOfServiceLevel > 0)
{
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
@@ -294,7 +299,10 @@ namespace MQTTnet.Server
ClientId,
applicationMessage);
_options.ClientMessageQueueInterceptor?.Invoke(context);
if (_options.ClientMessageQueueInterceptor != null)
{
await _options.ClientMessageQueueInterceptor.InterceptClientMessageQueueEnqueueAsync(context).ConfigureAwait(false);
}
if (!context.AcceptEnqueue || context.ApplicationMessage == null)
{
@@ -309,35 +317,33 @@ namespace MQTTnet.Server
_pendingPacketsQueue.Enqueue(publishPacket);
}
private async Task TryDisconnectAdapterAsync(IMqttChannelAdapter adapter)
{
if (adapter == null)
{
return;
}
//private async Task TryDisconnectAdapterAsync(IMqttChannelAdapter adapter)
//{
// if (adapter == null)
// {
// return;
// }
try
{
await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.Error(exception, "Error while disconnecting channel adapter.");
}
}
// try
// {
// await adapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
// }
// catch (Exception exception)
// {
// _logger.Error(exception, "Error while disconnecting channel adapter.");
// }
//}
private void ProcessReceivedPacket(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken)
private Task ProcessReceivedPacketAsync(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken)
{
if (packet is MqttPublishPacket publishPacket)
{
HandleIncomingPublishPacket(adapter, publishPacket, cancellationToken);
return;
return HandleIncomingPublishPacketAsync(adapter, publishPacket, cancellationToken);
}
if (packet is MqttPingReqPacket)
{
adapter.SendPacketAsync(new MqttPingRespPacket(), cancellationToken).GetAwaiter().GetResult();
return;
return adapter.SendPacketAsync(new MqttPingRespPacket(), cancellationToken);
}
if (packet is MqttPubRelPacket pubRelPacket)
@@ -348,8 +354,7 @@ namespace MQTTnet.Server
ReasonCode = MqttPubCompReasonCode.Success
};
adapter.SendPacketAsync(responsePacket, cancellationToken).GetAwaiter().GetResult();
return;
return adapter.SendPacketAsync(responsePacket, cancellationToken);
}
if (packet is MqttPubRecPacket pubRecPacket)
@@ -360,91 +365,83 @@ namespace MQTTnet.Server
ReasonCode = MqttPubRelReasonCode.Success
};
adapter.SendPacketAsync(responsePacket, cancellationToken).GetAwaiter().GetResult();
return;
return adapter.SendPacketAsync(responsePacket, cancellationToken);
}
if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
{
return;
return Task.FromResult(0);
}
if (packet is MqttSubscribePacket subscribePacket)
{
HandleIncomingSubscribePacket(adapter, subscribePacket, cancellationToken);
return;
return HandleIncomingSubscribePacketAsync(adapter, subscribePacket, cancellationToken);
}
if (packet is MqttUnsubscribePacket unsubscribePacket)
{
HandleIncomingUnsubscribePacket(adapter, unsubscribePacket, cancellationToken);
return;
return HandleIncomingUnsubscribePacketAsync(adapter, unsubscribePacket, cancellationToken);
}
if (packet is MqttDisconnectPacket)
{
Stop(MqttClientDisconnectType.Clean, true);
return;
return StopAsync(MqttClientDisconnectType.Clean, true);
}
if (packet is MqttConnectPacket)
{
Stop(MqttClientDisconnectType.NotClean, true);
return;
return StopAsync(MqttClientDisconnectType.NotClean, true);
}
_logger.Warning(null, "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
Stop(MqttClientDisconnectType.NotClean, true);
return StopAsync(MqttClientDisconnectType.NotClean, true);
}
private void EnqueueSubscribedRetainedMessages(ICollection<TopicFilter> topicFilters)
private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters)
{
var retainedMessages = _retainedMessagesManager.GetSubscribedMessages(topicFilters);
var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false);
foreach (var applicationMessage in retainedMessages)
{
EnqueueApplicationMessage(null, applicationMessage, true);
await EnqueueApplicationMessageAsync(null, applicationMessage, true).ConfigureAwait(false);
}
}
private void HandleIncomingSubscribePacket(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{
var subscribeResult = _subscriptionsManager.Subscribe(subscribePacket);
adapter.SendPacketAsync(subscribeResult.ResponsePacket, cancellationToken).GetAwaiter().GetResult();
var subscribeResult = await _subscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false);
await adapter.SendPacketAsync(subscribeResult.ResponsePacket, cancellationToken).ConfigureAwait(false);
if (subscribeResult.CloseConnection)
{
Stop(MqttClientDisconnectType.NotClean, true);
return;
await StopAsync(MqttClientDisconnectType.NotClean, true).ConfigureAwait(false);
}
EnqueueSubscribedRetainedMessages(subscribePacket.TopicFilters);
await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false);
}
private void HandleIncomingUnsubscribePacket(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
private Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
{
var unsubscribeResult = _subscriptionsManager.Unsubscribe(unsubscribePacket);
adapter.SendPacketAsync(unsubscribeResult, cancellationToken).GetAwaiter().GetResult();
return adapter.SendPacketAsync(unsubscribeResult, cancellationToken);
}
private void HandleIncomingPublishPacket(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
private Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
switch (publishPacket.QualityOfServiceLevel)
{
case MqttQualityOfServiceLevel.AtMostOnce:
{
HandleIncomingPublishPacketWithQoS0(publishPacket);
break;
return Task.FromResult(0);
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
HandleIncomingPublishPacketWithQoS1(adapter, publishPacket, cancellationToken);
break;
return HandleIncomingPublishPacketWithQoS1Async(adapter, publishPacket, cancellationToken);
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
HandleIncomingPublishPacketWithQoS2(adapter, publishPacket, cancellationToken);
break;
return HandleIncomingPublishPacketWithQoS2Async(adapter, publishPacket, cancellationToken);
}
default:
{
@@ -456,17 +453,17 @@ namespace MQTTnet.Server
private void HandleIncomingPublishPacketWithQoS0(MqttPublishPacket publishPacket)
{
_sessionsManager.EnqueueApplicationMessage(
this,
this,
_channelAdapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket));
}
private void HandleIncomingPublishPacketWithQoS1(
private Task HandleIncomingPublishPacketWithQoS1Async(
IMqttChannelAdapter adapter,
MqttPublishPacket publishPacket,
CancellationToken cancellationToken)
{
_sessionsManager.EnqueueApplicationMessage(
this,
this,
_channelAdapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket));
var response = new MqttPubAckPacket
@@ -475,10 +472,10 @@ namespace MQTTnet.Server
ReasonCode = MqttPubAckReasonCode.Success
};
adapter.SendPacketAsync(response, cancellationToken).GetAwaiter().GetResult();
return adapter.SendPacketAsync(response, cancellationToken);
}
private void HandleIncomingPublishPacketWithQoS2(
private Task HandleIncomingPublishPacketWithQoS2Async(
IMqttChannelAdapter adapter,
MqttPublishPacket publishPacket,
CancellationToken cancellationToken)
@@ -492,7 +489,7 @@ namespace MQTTnet.Server
ReasonCode = MqttPubRecReasonCode.Success
};
adapter.SendPacketAsync(response, cancellationToken).GetAwaiter().GetResult();
return adapter.SendPacketAsync(response, cancellationToken);
}
private void OnAdapterReadingPacketCompleted(object sender, EventArgs e)
Source/MQTTnet/Server/MqttClientSessionStatus.cs
+4 -7
View File
@@ -25,22 +25,19 @@ namespace MQTTnet.Server
public Task DisconnectAsync()
{
_session.Stop(MqttClientDisconnectType.NotClean);
return Task.FromResult(0);
return _session.StopAsync(MqttClientDisconnectType.NotClean);
}
public Task DeleteSessionAsync()
public async Task DeleteSessionAsync()
{
try
{
_session.Stop(MqttClientDisconnectType.NotClean);
await _session.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
}
finally
{
_sessionsManager.DeleteSession(ClientId);
await _sessionsManager.DeleteSessionAsync(ClientId).ConfigureAwait(false);
}
return Task.FromResult(0);
}
public Task ClearPendingApplicationMessagesAsync()
Source/MQTTnet/Server/MqttClientSessionsManager.cs
+46 -50
View File
@@ -1,12 +1,11 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -16,9 +15,7 @@ namespace MQTTnet.Server
{
private readonly BlockingCollection<MqttEnqueuedApplicationMessage> _messageQueue = new BlockingCollection<MqttEnqueuedApplicationMessage>();
/// <summary>
/// manual locking dictionaries is faster than using concurrent dictionary
/// </summary>
private readonly AsyncLock _sessionsLock = new AsyncLock();
private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
private readonly CancellationToken _cancellationToken;
@@ -47,16 +44,16 @@ namespace MQTTnet.Server
public void Start()
{
Task.Factory.StartNew(() => TryProcessQueuedApplicationMessages(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
Task.Factory.StartNew(() => TryProcessQueuedApplicationMessagesAsync(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
}
public void Stop()
public async Task StopAsync()
{
lock (_sessions)
using (await _sessionsLock.WaitAsync(_cancellationToken).ConfigureAwait(false))
{
foreach (var session in _sessions)
{
session.Value.Stop(MqttClientDisconnectType.NotClean);
await session.Value.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
}
_sessions.Clear();
@@ -68,18 +65,21 @@ namespace MQTTnet.Server
return Task.Run(() => RunSessionAsync(clientAdapter, _cancellationToken), _cancellationToken);
}
public IList<IMqttClientSessionStatus> GetClientStatus()
public async Task<IList<IMqttClientSessionStatus>> GetClientStatusAsync()
{
var result = new List<IMqttClientSessionStatus>();
foreach (var session in GetSessions())
using (await _sessionsLock.WaitAsync(_cancellationToken).ConfigureAwait(false))
{
var status = new MqttClientSessionStatus(this, session);
session.FillStatus(status);
foreach (var session in _sessions.Values)
{
var status = new MqttClientSessionStatus(this, session);
session.FillStatus(status);
result.Add(status);
result.Add(status);
}
}
return result;
}
@@ -122,9 +122,9 @@ namespace MQTTnet.Server
}
}
public void DeleteSession(string clientId)
public async Task DeleteSessionAsync(string clientId)
{
lock (_sessions)
using (await _sessionsLock.WaitAsync(_cancellationToken).ConfigureAwait(false))
{
_sessions.Remove(clientId);
}
@@ -137,13 +137,13 @@ namespace MQTTnet.Server
_messageQueue?.Dispose();
}
private void TryProcessQueuedApplicationMessages(CancellationToken cancellationToken)
private async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
try
{
TryProcessNextQueuedApplicationMessage(cancellationToken);
await TryProcessNextQueuedApplicationMessageAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
@@ -155,7 +155,7 @@ namespace MQTTnet.Server
}
}
private void TryProcessNextQueuedApplicationMessage(CancellationToken cancellationToken)
private async Task TryProcessNextQueuedApplicationMessageAsync(CancellationToken cancellationToken)
{
try
{
@@ -164,12 +164,12 @@ namespace MQTTnet.Server
var sender = enqueuedApplicationMessage.Sender;
var applicationMessage = enqueuedApplicationMessage.ApplicationMessage;
var interceptorContext = InterceptApplicationMessage(sender, applicationMessage);
var interceptorContext = await InterceptApplicationMessageAsync(sender, applicationMessage).ConfigureAwait(false);
if (interceptorContext != null)
{
if (interceptorContext.CloseConnection)
{
enqueuedApplicationMessage.Sender.Stop(MqttClientDisconnectType.NotClean);
await enqueuedApplicationMessage.Sender.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false);
}
if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
@@ -184,12 +184,18 @@ namespace MQTTnet.Server
if (applicationMessage.Retain)
{
_retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).GetAwaiter().GetResult();
await _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false);
}
foreach (var clientSession in GetSessions())
using (await _sessionsLock.WaitAsync(_cancellationToken).ConfigureAwait(false))
{
clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, enqueuedApplicationMessage.ApplicationMessage, false);
foreach (var clientSession in _sessions.Values)
{
await clientSession.EnqueueApplicationMessageAsync(
enqueuedApplicationMessage.Sender,
enqueuedApplicationMessage.ApplicationMessage,
false).ConfigureAwait(false);
}
}
}
catch (OperationCanceledException)
@@ -201,35 +207,22 @@ namespace MQTTnet.Server
}
}
private List<MqttClientSession> GetSessions()
{
lock (_sessions)
{
return _sessions.Values.ToList();
}
}
private async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
{
var clientId = string.Empty;
try
{
// TODO: Catch cancel exception here if the first packet was not received and log properly.
var firstPacket = await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false);
if (firstPacket == null)
{
return;
}
if (!(firstPacket is MqttConnectPacket connectPacket))
{
throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1].");
_logger.Warning(null, "The first packet from client '{0}' was no 'CONNECT' packet [MQTT-3.1.0-1].", clientAdapter.Endpoint);
return;
}
clientId = connectPacket.ClientId;
var connectReturnCode = ValidateConnection(connectPacket, clientAdapter);
var connectReturnCode = await ValidateConnectionAsync(connectPacket, clientAdapter).ConfigureAwait(false);
if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
{
await clientAdapter.SendPacketAsync(
@@ -242,7 +235,7 @@ namespace MQTTnet.Server
return;
}
var result = PrepareClientSession(connectPacket);
var result = await PrepareClientSessionAsync(connectPacket).ConfigureAwait(false);
await clientAdapter.SendPacketAsync(
new MqttConnAckPacket
@@ -267,14 +260,17 @@ namespace MQTTnet.Server
}
finally
{
await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false);
clientAdapter.Dispose();
if (!_options.EnablePersistentSessions)
{
DeleteSession(clientId);
await DeleteSessionAsync(clientId).ConfigureAwait(false);
}
}
}
private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter)
private async Task<MqttConnectReturnCode> ValidateConnectionAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter)
{
if (_options.ConnectionValidator == null)
{
@@ -288,13 +284,13 @@ namespace MQTTnet.Server
connectPacket.WillMessage,
clientAdapter.Endpoint);
_options.ConnectionValidator(context);
await _options.ConnectionValidator.ValidateConnection(context).ConfigureAwait(false);
return context.ReturnCode;
}
private PrepareClientSessionResult PrepareClientSession(MqttConnectPacket connectPacket)
private async Task<PrepareClientSessionResult> PrepareClientSessionAsync(MqttConnectPacket connectPacket)
{
lock (_sessions)
using (await _sessionsLock.WaitAsync(_cancellationToken).ConfigureAwait(false))
{
var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession);
if (isSessionPresent)
@@ -303,7 +299,7 @@ namespace MQTTnet.Server
{
_sessions.Remove(connectPacket.ClientId);
clientSession.Stop(MqttClientDisconnectType.Clean);
await clientSession.StopAsync(MqttClientDisconnectType.Clean).ConfigureAwait(false);
clientSession.Dispose();
clientSession = null;
@@ -330,7 +326,7 @@ namespace MQTTnet.Server
}
}
private MqttApplicationMessageInterceptorContext InterceptApplicationMessage(MqttClientSession sender, MqttApplicationMessage applicationMessage)
private async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(MqttClientSession sender, MqttApplicationMessage applicationMessage)
{
var interceptor = _options.ApplicationMessageInterceptor;
if (interceptor == null)
@@ -339,7 +335,7 @@ namespace MQTTnet.Server
}
var interceptorContext = new MqttApplicationMessageInterceptorContext(sender?.ClientId, applicationMessage);
interceptor(interceptorContext);
await interceptor.InterceptApplicationMessagePublishAsync(interceptorContext).ConfigureAwait(false);
return interceptorContext;
}
}
Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs
+11 -6
View File
@@ -1,6 +1,7 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -20,7 +21,7 @@ namespace MQTTnet.Server
_eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
}
public MqttClientSubscribeResult Subscribe(MqttSubscribePacket subscribePacket)
public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket)
{
if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
@@ -36,7 +37,7 @@ namespace MQTTnet.Server
foreach (var topicFilter in subscribePacket.TopicFilters)
{
var interceptorContext = InterceptSubscribe(topicFilter);
var interceptorContext = await InterceptSubscribeAsync(topicFilter).ConfigureAwait(false);
if (!interceptorContext.AcceptSubscription)
{
result.ResponsePacket.ReturnCodes.Add(MqttSubscribeReturnCode.Failure);
@@ -146,11 +147,15 @@ namespace MQTTnet.Server
}
}
private MqttSubscriptionInterceptorContext InterceptSubscribe(TopicFilter topicFilter)
private async Task<MqttSubscriptionInterceptorContext> InterceptSubscribeAsync(TopicFilter topicFilter)
{
var interceptorContext = new MqttSubscriptionInterceptorContext(_clientId, topicFilter);
_options.SubscriptionInterceptor?.Invoke(interceptorContext);
return interceptorContext;
var context = new MqttSubscriptionInterceptorContext(_clientId, topicFilter);
if (_options.SubscriptionInterceptor != null)
{
await _options.SubscriptionInterceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false);
}
return context;
}
private static CheckSubscriptionsResult CreateSubscriptionResult(MqttQualityOfServiceLevel qosLevel, HashSet<MqttQualityOfServiceLevel> subscribedQoSLevels)
Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
+21 -28
View File
@@ -3,11 +3,14 @@ using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Internal;
namespace MQTTnet.Server
{
public class MqttRetainedMessagesManager
{
private readonly byte[] _emptyArray = new byte[0];
private readonly AsyncLock _messagesLock = new AsyncLock();
private readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>();
private readonly IMqttNetChildLogger _logger;
@@ -31,7 +34,7 @@ namespace MQTTnet.Server
{
var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false);
lock (_messages)
using (await _messagesLock.WaitAsync().ConfigureAwait(false))
{
_messages.Clear();
foreach (var retainedMessage in retainedMessages)
@@ -52,8 +55,7 @@ namespace MQTTnet.Server
try
{
List<MqttApplicationMessage> messagesForSave = null;
lock (_messages)
using (await _messagesLock.WaitAsync().ConfigureAwait(false))
{
var saveIsRequired = false;
@@ -71,7 +73,7 @@ namespace MQTTnet.Server
}
else
{
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? _emptyArray))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
@@ -83,20 +85,13 @@ namespace MQTTnet.Server
if (saveIsRequired)
{
messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
if (_options.Storage != null)
{
var messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
await _options.Storage.SaveRetainedMessagesAsync(messagesForSave).ConfigureAwait(false);
}
}
}
if (messagesForSave == null)
{
_logger.Verbose("Skipped saving retained messages because no changes were detected.");
return;
}
if (_options.Storage != null)
{
await _options.Storage.SaveRetainedMessagesAsync(messagesForSave).ConfigureAwait(false);
}
}
catch (Exception exception)
{
@@ -104,11 +99,11 @@ namespace MQTTnet.Server
}
}
public IList<MqttApplicationMessage> GetSubscribedMessages(ICollection<TopicFilter> topicFilters)
public async Task<List<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<TopicFilter> topicFilters)
{
var retainedMessages = new List<MqttApplicationMessage>();
lock (_messages)
using (await _messagesLock.WaitAsync().ConfigureAwait(false))
{
foreach (var retainedMessage in _messages.Values)
{
@@ -128,27 +123,25 @@ namespace MQTTnet.Server
return retainedMessages;
}
public IList<MqttApplicationMessage> GetMessages()
public async Task<List<MqttApplicationMessage>> GetMessagesAsync()
{
lock (_messages)
using (await _messagesLock.WaitAsync().ConfigureAwait(false))
{
return _messages.Values.ToList();
}
}
public Task ClearMessagesAsync()
public async Task ClearMessagesAsync()
{
lock (_messages)
using (await _messagesLock.WaitAsync().ConfigureAwait(false))
{
_messages.Clear();
}
if (_options.Storage != null)
{
return _options.Storage.SaveRetainedMessagesAsync(new List<MqttApplicationMessage>());
if (_options.Storage != null)
{
await _options.Storage.SaveRetainedMessagesAsync(new List<MqttApplicationMessage>()).ConfigureAwait(false);
}
}
return Task.FromResult((object)null);
}
}
}
Source/MQTTnet/Server/MqttServer.cs
+3 -8
View File
@@ -48,17 +48,12 @@ namespace MQTTnet.Server
public Task<IList<IMqttClientSessionStatus>> GetClientSessionsStatusAsync()
{
return Task.FromResult(_clientSessionsManager.GetClientStatus());
}
public IList<IMqttClientSessionStatus> GetClientSessionsStatus()
{
return _clientSessionsManager.GetClientStatus();
return _clientSessionsManager.GetClientStatusAsync();
}
public IList<MqttApplicationMessage> GetRetainedMessages()
{
return _retainedMessagesManager.GetMessages();
return _retainedMessagesManager.GetMessagesAsync().GetAwaiter().GetResult();
}
public Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
@@ -123,7 +118,7 @@ namespace MQTTnet.Server
_cancellationTokenSource.Cancel(false);
_clientSessionsManager.Stop();
_clientSessionsManager.StopAsync().ConfigureAwait(false);
foreach (var adapter in _adapters)
{
Source/MQTTnet/Server/MqttServerApplicationMessageInterceptorDelegate.cs
+31
View File
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttServerApplicationMessageInterceptorDelegate : IMqttServerApplicationMessageInterceptor
{
private readonly Func<MqttApplicationMessageInterceptorContext, Task> _callback;
public MqttServerApplicationMessageInterceptorDelegate(Action<MqttApplicationMessageInterceptorContext> callback)
{
if (callback == null) throw new ArgumentNullException(nameof(callback));
_callback = context =>
{
callback(context);
return Task.FromResult(0);
};
}
public MqttServerApplicationMessageInterceptorDelegate(Func<MqttApplicationMessageInterceptorContext, Task> callback)
{
_callback = callback ?? throw new ArgumentNullException(nameof(callback));
}
public Task InterceptApplicationMessagePublishAsync(MqttApplicationMessageInterceptorContext context)
{
return _callback(context);
}
}
}
Source/MQTTnet/Server/MqttServerConnectionValidatorDelegate.cs
+31
View File
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttServerConnectionValidatorDelegate : IMqttServerConnectionValidator
{
private readonly Func<MqttConnectionValidatorContext, Task> _callback;
public MqttServerConnectionValidatorDelegate(Action<MqttConnectionValidatorContext> callback)
{
if (callback == null) throw new ArgumentNullException(nameof(callback));
_callback = context =>
{
callback(context);
return Task.FromResult(0);
};
}
public MqttServerConnectionValidatorDelegate(Func<MqttConnectionValidatorContext, Task> callback)
{
_callback = callback ?? throw new ArgumentNullException(nameof(callback));
}
public Task ValidateConnection(MqttConnectionValidatorContext context)
{
return _callback(context);
}
}
}
Source/MQTTnet/Server/MqttServerOptions.cs
+5 -5
View File
@@ -16,13 +16,13 @@ namespace MQTTnet.Server
public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15);
public Action<MqttConnectionValidatorContext> ConnectionValidator { get; set; }
public IMqttServerConnectionValidator ConnectionValidator { get; set; }
public Action<MqttApplicationMessageInterceptorContext> ApplicationMessageInterceptor { get; set; }
public IMqttServerApplicationMessageInterceptor ApplicationMessageInterceptor { get; set; }
public IMqttServerClientMessageQueueInterceptor ClientMessageQueueInterceptor { get; set; }
public Action<MqttClientMessageQueueInterceptorContext> ClientMessageQueueInterceptor { get; set; }
public Action<MqttSubscriptionInterceptorContext> SubscriptionInterceptor { get; set; }
public IMqttServerSubscriptionInterceptor SubscriptionInterceptor { get; set; }
public IMqttServerStorage Storage { get; set; }
}
Source/MQTTnet/Server/MqttServerOptionsBuilder.cs
+21 -3
View File
@@ -99,24 +99,42 @@ namespace MQTTnet.Server
return this;
}
public MqttServerOptionsBuilder WithConnectionValidator(Action<MqttConnectionValidatorContext> value)
public MqttServerOptionsBuilder WithConnectionValidator(IMqttServerConnectionValidator value)
{
_options.ConnectionValidator = value;
return this;
}
public MqttServerOptionsBuilder WithApplicationMessageInterceptor(Action<MqttApplicationMessageInterceptorContext> value)
public MqttServerOptionsBuilder WithConnectionValidator(Action<MqttConnectionValidatorContext> value)
{
_options.ConnectionValidator = new MqttServerConnectionValidatorDelegate(value);
return this;
}
public MqttServerOptionsBuilder WithApplicationMessageInterceptor(IMqttServerApplicationMessageInterceptor value)
{
_options.ApplicationMessageInterceptor = value;
return this;
}
public MqttServerOptionsBuilder WithSubscriptionInterceptor(Action<MqttSubscriptionInterceptorContext> value)
public MqttServerOptionsBuilder WithApplicationMessageInterceptor(Action<MqttApplicationMessageInterceptorContext> value)
{
_options.ApplicationMessageInterceptor = new MqttServerApplicationMessageInterceptorDelegate(value);
return this;
}
public MqttServerOptionsBuilder WithSubscriptionInterceptor(IMqttServerSubscriptionInterceptor value)
{
_options.SubscriptionInterceptor = value;
return this;
}
public MqttServerOptionsBuilder WithSubscriptionInterceptor(Action<MqttSubscriptionInterceptorContext> value)
{
_options.SubscriptionInterceptor = new MqttServerSubscriptionInterceptorDelegate(value);
return this;
}
public MqttServerOptionsBuilder WithPersistentSessions()
{
_options.EnablePersistentSessions = true;
Source/MQTTnet/Server/MqttServerSubscriptionInterceptorDelegate.cs
+31
View File
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttServerSubscriptionInterceptorDelegate : IMqttServerSubscriptionInterceptor
{
private readonly Func<MqttSubscriptionInterceptorContext, Task> _callback;
public MqttServerSubscriptionInterceptorDelegate(Action<MqttSubscriptionInterceptorContext> callback)
{
if (callback == null) throw new ArgumentNullException(nameof(callback));
_callback = context =>
{
callback(context);
return Task.FromResult(0);
};
}
public MqttServerSubscriptionInterceptorDelegate(Func<MqttSubscriptionInterceptorContext, Task> callback)
{
_callback = callback ?? throw new ArgumentNullException(nameof(callback));
}
public Task InterceptSubscriptionAsync(MqttSubscriptionInterceptorContext context)
{
return _callback(context);
}
}
}