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

Refactor subscribe and unsubscribe to support MQTTv5.

Browse Source
This commit is contained in:
Christian Kratky
2019-01-21 21:38:39 +01:00
parent 0d6e244f84
commit ebf0e91ab8
39 changed files with 1183 additions and 476 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs
+9 -2
View File
@@ -5,6 +5,7 @@ using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client;
using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Internal;
@@ -52,6 +53,12 @@ namespace MQTTnet.Extensions.ManagedClient
public event EventHandler<MqttClientConnectedEventArgs> Connected;
public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
public IMqttApplicationMessageHandler ReceivedApplicationMessageHandler
{
get => _mqttClient.ReceivedApplicationMessageHandler;
set => _mqttClient.ReceivedApplicationMessageHandler = value;
}
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
public event EventHandler<ApplicationMessageSkippedEventArgs> ApplicationMessageSkipped;
@@ -372,12 +379,12 @@ namespace MQTTnet.Extensions.ManagedClient
{
if (unsubscriptions.Any())
{
await _mqttClient.UnsubscribeAsync(unsubscriptions).ConfigureAwait(false);
await _mqttClient.UnsubscribeAsync(unsubscriptions.ToArray()).ConfigureAwait(false);
}
if (subscriptions.Any())
{
await _mqttClient.SubscribeAsync(subscriptions).ConfigureAwait(false);
await _mqttClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false);
}
}
catch (Exception exception)
Source/MQTTnet/Adapter/MqttChannelAdapter.cs
+16 -3
View File
@@ -174,12 +174,18 @@ namespace MQTTnet.Adapter
private async Task<ReceivedMqttPacket> ReceiveAsync(CancellationToken cancellationToken)
{
var fixedHeader = await _packetReader.ReadFixedHeaderAsync(_fixedHeaderBuffer, cancellationToken).ConfigureAwait(false);
var readFixedHeaderResult = await _packetReader.ReadFixedHeaderAsync(_fixedHeaderBuffer, cancellationToken).ConfigureAwait(false);
try
{
if (readFixedHeaderResult.ConnectionClosed)
{
return null;
}
ReadingPacketStarted?.Invoke(this, EventArgs.Empty);
var fixedHeader = readFixedHeaderResult.FixedHeader;
if (fixedHeader.RemainingLength == 0)
{
return new ReceivedMqttPacket(fixedHeader.Flags, null, 2);
@@ -205,9 +211,16 @@ namespace MQTTnet.Adapter
var readBytes = _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult();
#endif
cancellationToken.ThrowIfCancellationRequested();
ExceptionHelper.ThrowIfGracefulSocketClose(readBytes);
if (cancellationToken.IsCancellationRequested)
{
return null;
}
if (readBytes == 0)
{
return null;
}
bodyOffset += readBytes;
} while (bodyOffset < body.Length);
Source/MQTTnet/Client/IMqttClient.cs
+2 -3
View File
@@ -1,5 +1,4 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
@@ -20,7 +19,7 @@ namespace MQTTnet.Client
Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options);
Task DisconnectAsync(MqttClientDisconnectOptions options);
Task<MqttClientSubscribeResult> SubscribeAsync(IEnumerable<TopicFilter> topicFilters);
Task<MqttClientUnsubscribeResult> UnsubscribeAsync(IEnumerable<string> topics);
Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options);
Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options);
}
}
Source/MQTTnet/Client/MqttClient.cs
+192 -121
View File
@@ -1,20 +1,19 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Client.PacketDispatcher;
using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving;
using MQTTnet.Client.Subscribing;
using MQTTnet.Client.Unsubscribing;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.PacketDispatcher;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -23,16 +22,18 @@ namespace MQTTnet.Client
public class MqttClient : IMqttClient
{
private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
private readonly Stopwatch _sendTracker = new Stopwatch();
private readonly object _disconnectLock = new object();
private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
private readonly IMqttClientAdapterFactory _adapterFactory;
private readonly IMqttNetChildLogger _logger;
private CancellationTokenSource _cancellationTokenSource;
internal Task _packetReceiverTask;
internal Task _keepAliveMessageSenderTask;
private Task _packetReceiverTask;
private Task _keepAlivePacketsSenderTask;
private Task _backgroundWorkerTask;
private IMqttChannelAdapter _adapter;
private bool _cleanDisconnectInitiated;
private long _disconnectGate;
@@ -47,6 +48,8 @@ namespace MQTTnet.Client
public event EventHandler<MqttClientConnectedEventArgs> Connected;
public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
public IMqttApplicationMessageHandler ReceivedApplicationMessageHandler { get; set; }
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public bool IsConnected { get; private set; }
@@ -66,28 +69,31 @@ namespace MQTTnet.Client
_packetIdentifierProvider.Reset();
_packetDispatcher.Reset();
var cancellationTokenSource = new CancellationTokenSource();
_cancellationTokenSource = cancellationTokenSource;
_cancellationTokenSource = new CancellationTokenSource();
var cancellationToken = _cancellationTokenSource.Token;
_disconnectGate = 0;
_adapter = _adapterFactory.CreateClientAdapter(options, _logger);
var adapter = _adapterFactory.CreateClientAdapter(options, _logger);
_adapter = adapter;
_logger.Verbose($"Trying to connect with server ({Options.ChannelOptions}).");
await _adapter.ConnectAsync(Options.CommunicationTimeout, cancellationTokenSource.Token).ConfigureAwait(false);
await _adapter.ConnectAsync(Options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
_logger.Verbose("Connection with server established.");
StartReceivingPackets(cancellationTokenSource.Token);
_packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(cancellationToken), cancellationToken);
var connectResult = await AuthenticateAsync(options.WillMessage, cancellationTokenSource.Token).ConfigureAwait(false);
var connectResult = await AuthenticateAsync(adapter, options.WillMessage, cancellationToken).ConfigureAwait(false);
_logger.Verbose("MQTT connection with server established.");
_sendTracker.Restart();
if (Options.KeepAlivePeriod != TimeSpan.Zero)
{
StartSendingKeepAliveMessages(cancellationTokenSource.Token);
_keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessagesAsync(cancellationToken), cancellationToken);
}
_backgroundWorkerTask = Task.Run(() => TryProcessReceivedPacketsAsync(cancellationToken), cancellationToken);
IsConnected = true;
Connected?.Invoke(this, new MqttClientConnectedEventArgs(connectResult));
@@ -129,35 +135,27 @@ namespace MQTTnet.Client
}
}
public async Task<MqttClientSubscribeResult> SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options)
{
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
if (options == null) throw new ArgumentNullException(nameof(options));
ThrowIfNotConnected();
var subscribePacket = new MqttSubscribePacket
{
PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier()
};
subscribePacket.TopicFilters.AddRange(topicFilters);
var subscribePacket = _adapter.PacketFormatterAdapter.DataConverter.CreateSubscribePacket(options);
subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
var subAckPacket = await SendAndReceiveAsync<MqttSubAckPacket>(subscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false);
return _adapter.PacketFormatterAdapter.DataConverter.CreateClientSubscribeResult(subscribePacket, subAckPacket);
}
public async Task<MqttClientUnsubscribeResult> UnsubscribeAsync(IEnumerable<string> topicFilters)
public async Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options)
{
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
if (options == null) throw new ArgumentNullException(nameof(options));
ThrowIfNotConnected();
var unsubscribePacket = new MqttUnsubscribePacket
{
PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier()
};
unsubscribePacket.TopicFilters.AddRange(topicFilters);
var unsubscribePacket = _adapter.PacketFormatterAdapter.DataConverter.CreateUnsubscribePacket(options);
unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
var unsubAckPacket = await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false);
return _adapter.PacketFormatterAdapter.DataConverter.CreateClientUnsubscribeResult(unsubscribePacket, unsubAckPacket);
@@ -181,7 +179,7 @@ namespace MQTTnet.Client
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
var response = await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket, _cancellationTokenSource.Token).ConfigureAwait(false);
var result = new MqttClientPublishResult();
@@ -213,15 +211,14 @@ namespace MQTTnet.Client
_adapter = null;
}
private async Task<MqttClientConnectResult> AuthenticateAsync(MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken)
private async Task<MqttClientConnectResult> AuthenticateAsync(IMqttChannelAdapter channelAdapter, MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken)
{
var connectPacket = _adapter.PacketFormatterAdapter.DataConverter.CreateConnectPacket(
var connectPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnectPacket(
willApplicationMessage,
Options);
var connAckPacket = await SendAndReceiveAsync<MqttConnAckPacket>(connectPacket, cancellationToken).ConfigureAwait(false);
var result = _adapter.PacketFormatterAdapter.DataConverter.CreateClientConnectResult(connAckPacket);
var result = channelAdapter.PacketFormatterAdapter.DataConverter.CreateClientConnectResult(connAckPacket);
if (result.ResultCode != MqttClientConnectResultCode.Success)
{
@@ -257,7 +254,8 @@ namespace MQTTnet.Client
}
await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false);
await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false);
await WaitForTaskAsync(_keepAlivePacketsSenderTask, sender).ConfigureAwait(false);
await WaitForTaskAsync(_backgroundWorkerTask, sender).ConfigureAwait(false);
_logger.Verbose("Disconnected from adapter.");
}
@@ -316,26 +314,22 @@ namespace MQTTnet.Client
identifier = packetWithIdentifier.PacketIdentifier.Value;
}
var packetAwaiter = _packetDispatcher.AddPacketAwaiter<TResponsePacket>(identifier);
try
using (var packetAwaiter = _packetDispatcher.AddPacketAwaiter<TResponsePacket>(identifier))
{
await _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false);
return await packetAwaiter.WaitOneAsync(Options.CommunicationTimeout);
//return (TResponsePacket)await Internal.TaskExtensions.TimeoutAfterAsync(ct => packetAwaiter.Task, Options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
}
catch (MqttCommunicationTimedOutException)
{
_logger.Warning(null, "Timeout while waiting for packet of type '{0}'.", typeof(TResponsePacket).Namespace);
throw;
}
finally
{
_packetDispatcher.RemovePacketAwaiter<TResponsePacket>(identifier);
try
{
await _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false);
return await packetAwaiter.WaitOneAsync(Options.CommunicationTimeout).ConfigureAwait(false);
}
catch (MqttCommunicationTimedOutException)
{
_logger.Warning(null, "Timeout while waiting for packet of type '{0}'.", typeof(TResponsePacket).Namespace);
throw;
}
}
}
private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
private async Task TrySendKeepAliveMessagesAsync(CancellationToken cancellationToken)
{
try
{
@@ -380,7 +374,7 @@ namespace MQTTnet.Client
if (!DisconnectIsPending())
{
await DisconnectInternalAsync(_keepAliveMessageSenderTask, exception).ConfigureAwait(false);
await DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception).ConfigureAwait(false);
}
}
finally
@@ -389,7 +383,7 @@ namespace MQTTnet.Client
}
}
private async Task ReceivePacketsAsync(CancellationToken cancellationToken)
private async Task TryReceivePacketsAsync(CancellationToken cancellationToken)
{
try
{
@@ -399,10 +393,17 @@ namespace MQTTnet.Client
{
var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
if (packet != null && !cancellationToken.IsCancellationRequested)
if (cancellationToken.IsCancellationRequested)
{
await ProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false);
return;
}
if (packet == null && !DisconnectIsPending())
{
await DisconnectInternalAsync(_packetReceiverTask, null).ConfigureAwait(false);
}
_packetDispatcher.Dispatch(packet);
}
}
catch (Exception exception)
@@ -437,79 +438,146 @@ namespace MQTTnet.Client
}
}
private Task ProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
private async Task TryProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
{
if (packet is MqttPublishPacket publishPacket)
try
{
return ProcessReceivedPublishPacketAsync(publishPacket, cancellationToken);
if (packet is MqttPublishPacket publishPacket)
{
await TryProcessReceivedPublishPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false);
}
else if (packet is MqttPingReqPacket)
{
await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false);
}
else if (packet is MqttDisconnectPacket)
{
await DisconnectAsync(null).ConfigureAwait(false);
}
else
{
throw new MqttProtocolViolationException($"Received packet '{packet}' at an unexpected time.");
}
}
if (packet is MqttPingReqPacket)
catch (Exception exception)
{
return SendAsync(new MqttPingRespPacket(), cancellationToken);
}
if (_cleanDisconnectInitiated)
{
return;
}
if (packet is MqttDisconnectPacket)
{
return DisconnectAsync(null);
}
if (exception is OperationCanceledException)
{
}
else if (exception is MqttCommunicationException)
{
_logger.Warning(exception, "MQTT communication exception while receiving packets.");
}
else
{
_logger.Error(exception, "Unhandled exception while receiving packets.");
}
if (packet is MqttPubRelPacket pubRelPacket)
{
return ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken);
}
_packetDispatcher.Dispatch(exception);
_packetDispatcher.Dispatch(packet);
return Task.FromResult(0);
if (!DisconnectIsPending())
{
await DisconnectInternalAsync(_packetReceiverTask, exception).ConfigureAwait(false);
}
}
}
private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
private async Task TryProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
try
{
FireApplicationMessageReceivedEvent(publishPacket);
return Task.FromResult(0);
}
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
{
await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
}
else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{
await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
await SendAsync(new MqttPubAckPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubAckReasonCode.Success
}, cancellationToken).ConfigureAwait(false);
}
else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
{
var pubRecPacket = new MqttPubRecPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubRecReasonCode.Success
};
var pubRelPacket = await SendAndReceiveAsync<MqttPubRelPacket>(pubRecPacket, cancellationToken).ConfigureAwait(false);
// QoS 2 is implement as method "B" (4.3.3 QoS 2: Exactly once delivery)
await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
await SendAsync(new MqttPubCompPacket
{
PacketIdentifier = pubRelPacket.PacketIdentifier,
ReasonCode = MqttPubCompReasonCode.Success
}, cancellationToken).ConfigureAwait(false);
}
else
{
throw new MqttProtocolViolationException("Received a not supported QoS level.");
}
}
catch (Exception exception)
{
FireApplicationMessageReceivedEvent(publishPacket);
return SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier, ReasonCode = MqttPubAckReasonCode.Success }, cancellationToken);
_logger.Error(exception, "Unhandled exception while handling application message.");
}
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
{
// QoS 2 is implement as method "B" (4.3.3 QoS 2: Exactly once delivery)
FireApplicationMessageReceivedEvent(publishPacket);
return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier, ReasonCode = MqttPubRecReasonCode.Success }, cancellationToken);
}
throw new MqttCommunicationException("Received a not supported QoS level.");
}
private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
private async Task TryProcessReceivedPacketsAsync(CancellationToken cancellationToken)
{
var response = new MqttPubCompPacket
try
{
PacketIdentifier = pubRelPacket.PacketIdentifier,
ReasonCode = MqttPubCompReasonCode.Success
};
while (!cancellationToken.IsCancellationRequested)
{
var packet = _packetDispatcher.Take(cancellationToken);
return SendAsync(response, cancellationToken);
if (cancellationToken.IsCancellationRequested)
{
return;
}
await TryProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
_logger.Error(exception, "Error while processing packet.");
}
}
private async Task<MqttClientPublishResult> PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
// TODO: Check response code.
var pubRelPacket = new MqttPubRelPacket
{
PacketIdentifier = pubRecPacket.PacketIdentifier,
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubRelReasonCode.Success
};
var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
// TODO: Check response code.
var result = new MqttClientPublishResult();
if (pubRecPacket.ReasonCode != null)
@@ -520,35 +588,38 @@ namespace MQTTnet.Client
return result;
}
private void StartReceivingPackets(CancellationToken cancellationToken)
{
_packetReceiverTask = Task.Factory.StartNew(
() => ReceivePacketsAsync(cancellationToken),
cancellationToken,
TaskCreationOptions.LongRunning,
TaskScheduler.Default).Unwrap();
}
////private void StartReceivingPackets(CancellationToken cancellationToken)
////{
//// _packetReceiverTask = Task.Factory.StartNew(
//// () => TryReceivePacketsAsync(cancellationToken),
//// cancellationToken,
//// TaskCreationOptions.LongRunning,
//// TaskScheduler.Default).Unwrap();
////}
private void StartSendingKeepAliveMessages(CancellationToken cancellationToken)
{
_keepAliveMessageSenderTask = Task.Factory.StartNew(
() => SendKeepAliveMessagesAsync(cancellationToken),
cancellationToken,
TaskCreationOptions.LongRunning,
TaskScheduler.Default).Unwrap();
}
////private void StartSendingKeepAliveMessages(CancellationToken cancellationToken)
////{
//// _keepAlivePacketsSenderTask = Task.Factory.StartNew(
//// () => TrySendKeepAliveMessagesAsync(cancellationToken),
//// cancellationToken,
//// TaskCreationOptions.LongRunning,
//// TaskScheduler.Default).Unwrap();
////}
private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket)
private Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
{
try
var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket);
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage));
var handler = ReceivedApplicationMessageHandler;
if (handler != null)
{
var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket);
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage));
}
catch (Exception exception)
{
_logger.Error(exception, "Unhandled exception while handling application message.");
return handler.HandleApplicationMessageAsync(
new MqttApplicationMessageHandlerContext(Options.ClientId, applicationMessage));
}
return Task.FromResult(0);
}
private static async Task WaitForTaskAsync(Task task, Task sender)
Source/MQTTnet/Client/MqttClientExtensions.cs
+54 -2
View File
@@ -1,5 +1,6 @@
using System;
using System.Threading.Tasks;
using MQTTnet.Client.Receiving;
using MQTTnet.Client.Subscribing;
using MQTTnet.Client.Unsubscribing;
using MQTTnet.Protocol;
@@ -8,6 +9,51 @@ namespace MQTTnet.Client
{
public static class MqttClientExtensions
{
public static IMqttClient UseReceivedApplicationMessageHandler(this IMqttClient client, Func<MqttApplicationMessageHandlerContext, Task> handler)
{
if (handler == null)
{
client.ReceivedApplicationMessageHandler = null;
return client;
}
client.ReceivedApplicationMessageHandler = new MqttApplicationMessageHandlerDelegate(handler);
return client;
}
public static IMqttClient UseReceivedApplicationMessageHandler(this IMqttClient client, Action<MqttApplicationMessageHandlerContext> handler)
{
if (handler == null)
{
client.ReceivedApplicationMessageHandler = null;
return client;
}
client.ReceivedApplicationMessageHandler = new MqttApplicationMessageHandlerDelegate(handler);
return client;
}
public static IMqttClient UseReceivedApplicationMessageHandler(this IMqttClient client, IMqttApplicationMessageHandler handler)
{
client.ReceivedApplicationMessageHandler = handler;
return client;
}
public static Task ReconnectAsync(this IMqttClient client)
{
if (client == null) throw new ArgumentNullException(nameof(client));
if (client.Options == null)
{
throw new InvalidOperationException("_ReconnectAsync_ can be used only if _ConnectAsync_ was called before.");
}
return client.ConnectAsync(client.Options);
}
public static Task DisconnectAsync(this IMqttClient client)
{
if (client == null) throw new ArgumentNullException(nameof(client));
@@ -20,7 +66,10 @@ namespace MQTTnet.Client
if (client == null) throw new ArgumentNullException(nameof(client));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
return client.SubscribeAsync(topicFilters);
var options = new MqttClientSubscribeOptions();
options.TopicFilters.AddRange(topicFilters);
return client.SubscribeAsync(options);
}
public static Task<MqttClientSubscribeResult> SubscribeAsync(this IMqttClient client, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
@@ -44,7 +93,10 @@ namespace MQTTnet.Client
if (client == null) throw new ArgumentNullException(nameof(client));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
return client.UnsubscribeAsync(topicFilters);
var options = new MqttClientUnsubscribeOptions();
options.TopicFilters.AddRange(topicFilters);
return client.UnsubscribeAsync(options);
}
}
}
Source/MQTTnet/Client/MqttPacketIdentifierProvider.cs
+1 -1
View File
@@ -13,7 +13,7 @@
}
}
public ushort GetNewPacketIdentifier()
public ushort GetNextPacketIdentifier()
{
lock (_syncRoot)
{
Source/MQTTnet/Client/Options/MqttClientOptions.cs
+1 -1
View File
@@ -7,7 +7,7 @@ namespace MQTTnet.Client.Options
{
public string ClientId { get; set; } = Guid.NewGuid().ToString("N");
public bool CleanSession { get; set; } = true;
public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials();
public IMqttClientCredentials Credentials { get; set; }
public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
public IMqttClientChannelOptions ChannelOptions { get; set; }
Source/MQTTnet/Client/PacketDispatcher/MqttPacketAwaiter.cs
-37
View File
@@ -1,37 +0,0 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Packets;
namespace MQTTnet.Client.PacketDispatcher
{
public sealed class MqttPacketAwaiter<TPacket> : IMqttPacketAwaiter where TPacket : MqttBasePacket
{
private readonly TaskCompletionSource<MqttBasePacket> _packet = new TaskCompletionSource<MqttBasePacket>();
public async Task<TPacket> WaitOneAsync(TimeSpan timeout)
{
using (var timeoutToken = new CancellationTokenSource(timeout))
{
timeoutToken.Token.Register(() => _packet.TrySetCanceled());
var packet = await _packet.Task.ConfigureAwait(false);
return (TPacket)packet;
}
}
public void Complete(MqttBasePacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
_packet.TrySetResult(packet);
}
public void Fail(Exception exception)
{
if (exception == null) throw new ArgumentNullException(nameof(exception));
_packet.TrySetException(exception);
}
}
}
Source/MQTTnet/Client/Receiving/IMqttApplicationMessageHandler.cs
+9
View File
@@ -0,0 +1,9 @@
using System.Threading.Tasks;
namespace MQTTnet.Client.Receiving
{
public interface IMqttApplicationMessageHandler
{
Task HandleApplicationMessageAsync(MqttApplicationMessageHandlerContext context);
}
}
Source/MQTTnet/Client/Receiving/MqttApplicationMessageHandlerContext.cs
+15
View File
@@ -0,0 +1,15 @@
namespace MQTTnet.Client.Receiving
{
public class MqttApplicationMessageHandlerContext
{
public MqttApplicationMessageHandlerContext(string senderClientId, MqttApplicationMessage applicationMessage)
{
SenderClientId = senderClientId;
ApplicationMessage = applicationMessage;
}
public string SenderClientId { get; }
public MqttApplicationMessage ApplicationMessage { get; }
}
}
Source/MQTTnet/Client/Receiving/MqttApplicationMessageHandlerDelegate.cs
+31
View File
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Client.Receiving
{
public class MqttApplicationMessageHandlerDelegate : IMqttApplicationMessageHandler
{
private readonly Func<MqttApplicationMessageHandlerContext, Task> _handler;
public MqttApplicationMessageHandlerDelegate(Action<MqttApplicationMessageHandlerContext> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = context =>
{
handler(context);
return Task.FromResult(0);
};
}
public MqttApplicationMessageHandlerDelegate(Func<MqttApplicationMessageHandlerContext, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleApplicationMessageAsync(MqttApplicationMessageHandlerContext context)
{
return _handler(context);
}
}
}
Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs
+12
View File
@@ -0,0 +1,12 @@
using System.Collections.Generic;
using MQTTnet.Packets;
namespace MQTTnet.Client.Subscribing
{
public class MqttClientSubscribeOptions
{
public List<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>();
public List<MqttUserProperty> UserProperties { get; set; } = new List<MqttUserProperty>();
}
}
Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeOptions.cs
+12
View File
@@ -0,0 +1,12 @@
using System.Collections.Generic;
using MQTTnet.Packets;
namespace MQTTnet.Client.Unsubscribing
{
public class MqttClientUnsubscribeOptions
{
public List<string> TopicFilters { get; set; } = new List<string>();
public List<MqttUserProperty> UserProperties { get; set; } = new List<MqttUserProperty>();
}
}
Source/MQTTnet/Exceptions/MqttCommunicationClosedGracefullyException.cs
-6
View File
@@ -1,6 +0,0 @@
namespace MQTTnet.Exceptions
{
public class MqttCommunicationClosedGracefullyException : MqttCommunicationException
{
}
}
Source/MQTTnet/Formatter/IMqttDataConverter.cs
+4
View File
@@ -19,5 +19,9 @@ namespace MQTTnet.Formatter
MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket);
MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket);
MqttSubscribePacket CreateSubscribePacket(MqttClientSubscribeOptions options);
MqttUnsubscribePacket CreateUnsubscribePacket(MqttClientUnsubscribeOptions options);
}
}
Source/MQTTnet/Formatter/MqttPacketReader.cs
+66 -21
View File
@@ -18,7 +18,7 @@ namespace MQTTnet.Formatter
_channel = channel ?? throw new ArgumentNullException(nameof(channel));
}
public async Task<MqttFixedHeader> ReadFixedHeaderAsync(byte[] fixedHeaderBuffer, CancellationToken cancellationToken)
public async Task<ReadFixedHeaderResult> ReadFixedHeaderAsync(byte[] fixedHeaderBuffer, CancellationToken cancellationToken)
{
// The MQTT fixed header contains 1 byte of flags and at least 1 byte for the remaining data length.
// So in all cases at least 2 bytes must be read for a complete MQTT packet.
@@ -32,15 +32,25 @@ namespace MQTTnet.Formatter
var bytesRead = await _channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
ExceptionHelper.ThrowIfGracefulSocketClose(bytesRead);
if (bytesRead == 0)
{
return new ReadFixedHeaderResult
{
ConnectionClosed = true
};
}
totalBytesRead += bytesRead;
}
var hasRemainingLength = buffer[1] != 0;
if (!hasRemainingLength)
{
return new MqttFixedHeader(buffer[0], 0, totalBytesRead);
return new ReadFixedHeaderResult
{
FixedHeader = new MqttFixedHeader(buffer[0], 0, totalBytesRead)
};
}
#if WINDOWS_UWP
@@ -54,12 +64,23 @@ namespace MQTTnet.Formatter
var bodyLength = ReadBodyLength(buffer[1], cancellationToken);
#endif
totalBytesRead += bodyLength;
return new MqttFixedHeader(buffer[0], bodyLength, totalBytesRead);
if (!bodyLength.HasValue)
{
return new ReadFixedHeaderResult
{
ConnectionClosed = true
};
}
totalBytesRead += bodyLength.Value;
return new ReadFixedHeaderResult
{
FixedHeader = new MqttFixedHeader(buffer[0], bodyLength.Value, totalBytesRead)
};
}
#if !WINDOWS_UWP
private int ReadBodyLength(byte initialEncodedByte, CancellationToken cancellationToken)
private int? ReadBodyLength(byte initialEncodedByte, CancellationToken cancellationToken)
{
var offset = 0;
var multiplier = 128;
@@ -74,9 +95,18 @@ namespace MQTTnet.Formatter
throw new MqttProtocolViolationException("Remaining length is invalid.");
}
cancellationToken.ThrowIfCancellationRequested();
if (cancellationToken.IsCancellationRequested)
{
return null;
}
encodedByte = ReadByte(cancellationToken);
var buffer = ReadByte(cancellationToken);
if (!buffer.HasValue)
{
return null;
}
encodedByte = buffer.Value;
value += (encodedByte & 127) * multiplier;
multiplier *= 128;
@@ -85,15 +115,18 @@ namespace MQTTnet.Formatter
return value;
}
private byte ReadByte(CancellationToken cancellationToken)
private byte? ReadByte(CancellationToken cancellationToken)
{
var readCount = _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult();
cancellationToken.ThrowIfCancellationRequested();
if (readCount <= 0)
if (cancellationToken.IsCancellationRequested)
{
ExceptionHelper.ThrowGracefulSocketClose();
return null;
}
if (readCount == 0)
{
return null;
}
return _singleByteBuffer[0];
@@ -101,7 +134,7 @@ namespace MQTTnet.Formatter
#else
private async Task<int> ReadBodyLengthAsync(byte initialEncodedByte, CancellationToken cancellationToken)
private async Task<int?> ReadBodyLengthAsync(byte initialEncodedByte, CancellationToken cancellationToken)
{
var offset = 0;
var multiplier = 128;
@@ -116,9 +149,18 @@ namespace MQTTnet.Formatter
throw new MqttProtocolViolationException("Remaining length is invalid.");
}
cancellationToken.ThrowIfCancellationRequested();
if (cancellationToken.IsCancellationRequested)
{
return null;
}
encodedByte = await ReadByteAsync(cancellationToken).ConfigureAwait(false);
var buffer = await ReadByteAsync(cancellationToken).ConfigureAwait(false);
if (!buffer.HasValue)
{
return null;
}
encodedByte = buffer.Value;
value += (encodedByte & 127) * multiplier;
multiplier *= 128;
@@ -127,15 +169,18 @@ namespace MQTTnet.Formatter
return value;
}
private async Task<byte> ReadByteAsync(CancellationToken cancellationToken)
private async Task<byte?> ReadByteAsync(CancellationToken cancellationToken)
{
var readCount = await _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
if (readCount <= 0)
if (cancellationToken.IsCancellationRequested)
{
ExceptionHelper.ThrowGracefulSocketClose();
return null;
}
if (readCount == 0)
{
return null;
}
return _singleByteBuffer[0];
Source/MQTTnet/Formatter/ReadFixedHeaderResult.cs
+9
View File
@@ -0,0 +1,9 @@
namespace MQTTnet.Formatter
{
public class ReadFixedHeaderResult
{
public bool ConnectionClosed { get; set; }
public MqttFixedHeader FixedHeader { get; set; }
}
}
Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs
+30
View File
@@ -143,5 +143,35 @@ namespace MQTTnet.Formatter.V3
return result;
}
public MqttSubscribePacket CreateSubscribePacket(MqttClientSubscribeOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (options.UserProperties?.Any() == true)
{
throw new MqttProtocolViolationException("User properties are not supported in MQTT version 3.");
}
var subscribePacket = new MqttSubscribePacket();
subscribePacket.TopicFilters.AddRange(options.TopicFilters);
return subscribePacket;
}
public MqttUnsubscribePacket CreateUnsubscribePacket(MqttClientUnsubscribeOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (options.UserProperties?.Any() == true)
{
throw new MqttProtocolViolationException("User properties are not supported in MQTT version 3.");
}
var unsubscribePacket = new MqttUnsubscribePacket();
unsubscribePacket.TopicFilters.AddRange(options.TopicFilters);
return unsubscribePacket;
}
}
}
Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs
+30
View File
@@ -128,5 +128,35 @@ namespace MQTTnet.Formatter.V5
return result;
}
public MqttSubscribePacket CreateSubscribePacket(MqttClientSubscribeOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
var packet = new MqttSubscribePacket
{
Properties = new MqttSubscribePacketProperties()
};
packet.TopicFilters.AddRange(options.TopicFilters);
packet.Properties.UserProperties.AddRange(options.UserProperties);
return packet;
}
public MqttUnsubscribePacket CreateUnsubscribePacket(MqttClientUnsubscribeOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
var packet = new MqttUnsubscribePacket
{
Properties = new MqttUnsubscribePacketProperties()
};
packet.TopicFilters.AddRange(options.TopicFilters);
packet.Properties.UserProperties.AddRange(options.UserProperties);
return packet;
}
}
}
Source/MQTTnet/IApplicationMessageReceiver.cs
+4
View File
@@ -1,9 +1,13 @@
using System;
using MQTTnet.Client.Receiving;
namespace MQTTnet
{
public interface IApplicationMessageReceiver
{
IMqttApplicationMessageHandler ReceivedApplicationMessageHandler { get; set; }
[Obsolete("Use _ReceivedApplicationMessageHandler_ instead.")]
event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
}
}
Source/MQTTnet/Internal/ExceptionHelper.cs
-20
View File
@@ -1,20 +0,0 @@
using MQTTnet.Exceptions;
namespace MQTTnet.Internal
{
public static class ExceptionHelper
{
public static void ThrowGracefulSocketClose()
{
throw new MqttCommunicationClosedGracefullyException();
}
public static void ThrowIfGracefulSocketClose(int readBytesCount)
{
if (readBytesCount <= 0)
{
throw new MqttCommunicationClosedGracefullyException();
}
}
}
}
Source/MQTTnet/MQTTnet.csproj
-12
View File
@@ -63,20 +63,8 @@
<ItemGroup Condition="'$(TargetFramework)'=='net461'">
</ItemGroup>
<ItemGroup>
<Compile Remove="NewFolder\**" />
<EmbeddedResource Remove="NewFolder\**" />
<None Remove="NewFolder\**" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.2" />
</ItemGroup>
<ItemGroup>
<Reference Include="System.Net.Requests">
<HintPath>C:\Program Files\dotnet\sdk\NuGetFallbackFolder\microsoft.netcore.app\2.1.0\ref\netcoreapp2.1\System.Net.Requests.dll</HintPath>
</Reference>
</ItemGroup>
</Project>
Source/MQTTnet/MessageStream/MqttMessageStream.cs
+79
View File
@@ -0,0 +1,79 @@
using System;
using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Internal;
using MQTTnet.Packets;
namespace MQTTnet.MessageStream
{
public class MqttMessageStream
{
private readonly LinkedList<MqttBasePacket> _queue = new LinkedList<MqttBasePacket>();
private readonly LinkedList<TaskCompletionSource<MqttBasePacket>> _waitHandles = new LinkedList<TaskCompletionSource<MqttBasePacket>>();
private readonly SemaphoreSlim _lock = new SemaphoreSlim(1, 1);
private readonly BlockingQueue<MqttBasePacket> _packets = new BlockingQueue<MqttBasePacket>();
public void Enqueue(MqttBasePacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
lock (_queue)
{
//_queue.AddLast(packet);
_packets.Enqueue(packet);
}
//lock (_queue)
//{
// _queue.AddLast(packet);
// foreach (var waitHandle in _waitHandles)
// {
// waitHandle.TrySetResult(true);
// }
// _waitHandles.Clear();
//}
}
public Task<MqttBasePacket> TakeAsync(CancellationToken cancellationToken)
{
lock (_packets)
{
var packet = _packets.Dequeue();
return Task.FromResult(packet);
}
//while (!cancellationToken.IsCancellationRequested)
//{
// TaskCompletionSource<bool> waitHandle;
// lock (_queue)
// {
// if (_queue.Count > 0)
// {
// var node = _queue.First;
// _queue.RemoveFirst();
// return node.Value;
// }
// waitHandle = new TaskCompletionSource<bool>();
// _waitHandles.Add(waitHandle);
// }
// await waitHandle.Task;
//}
//return null;
}
}
}
Source/MQTTnet/Client/PacketDispatcher/IMqttPacketAwaiter.cs → Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs
+4 -2
View File
@@ -1,12 +1,14 @@
using System;
using MQTTnet.Packets;
namespace MQTTnet.Client.PacketDispatcher
namespace MQTTnet.PacketDispatcher
{
public interface IMqttPacketAwaiter
public interface IMqttPacketAwaiter : IDisposable
{
void Complete(MqttBasePacket packet);
void Fail(Exception exception);
void Cancel();
}
}
Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs
+59
View File
@@ -0,0 +1,59 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Packets;
namespace MQTTnet.PacketDispatcher
{
public sealed class MqttPacketAwaiter<TPacket> : IMqttPacketAwaiter where TPacket : MqttBasePacket
{
private readonly TaskCompletionSource<MqttBasePacket> _taskCompletionSource = new TaskCompletionSource<MqttBasePacket>();
private readonly ushort? _packetIdentifier;
private readonly MqttPacketDispatcher _owningPacketDispatcher;
public MqttPacketAwaiter(ushort? packetIdentifier, MqttPacketDispatcher owningPacketDispatcher)
{
_packetIdentifier = packetIdentifier;
_owningPacketDispatcher = owningPacketDispatcher ?? throw new ArgumentNullException(nameof(owningPacketDispatcher));
}
public async Task<TPacket> WaitOneAsync(TimeSpan timeout)
{
using (var timeoutToken = new CancellationTokenSource(timeout))
{
timeoutToken.Token.Register(() => _taskCompletionSource.TrySetCanceled());
var packet = await _taskCompletionSource.Task.ConfigureAwait(false);
return (TPacket)packet;
}
}
public void Complete(MqttBasePacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
// To prevent deadlocks it is required to call the _TrySetResult_ method
// from a new thread because the awaiting code will not(!) be executed in
// a new thread automatically (due to await). Furthermore _this_ thread will
// do it. But _this_ thread is also reading incoming packets -> deadlock.
Task.Run(() => _taskCompletionSource.TrySetResult(packet));
}
public void Fail(Exception exception)
{
if (exception == null) throw new ArgumentNullException(nameof(exception));
Task.Run(() => _taskCompletionSource.TrySetException(exception));
}
public void Cancel()
{
Task.Run(() => _taskCompletionSource.TrySetCanceled());
}
public void Dispose()
{
_owningPacketDispatcher.RemovePacketAwaiter<TPacket>(_packetIdentifier);
}
}
}
Source/MQTTnet/Client/PacketDispatcher/MqttPacketDispatcher.cs → Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs
+42 -15
View File
@@ -1,22 +1,24 @@
using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;
using System.Threading;
using MQTTnet.Packets;
namespace MQTTnet.Client.PacketDispatcher
namespace MQTTnet.PacketDispatcher
{
public class MqttPacketDispatcher
{
private readonly ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter> _awaiters = new ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter>();
private readonly ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter> _packetAwaiters = new ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter>();
private BlockingCollection<MqttBasePacket> _inboundPackagesQueue = new BlockingCollection<MqttBasePacket>();
public void Dispatch(Exception exception)
{
foreach (var awaiter in _awaiters)
foreach (var awaiter in _packetAwaiters)
{
Task.Run(() => awaiter.Value.Fail(exception)); // Task.Run fixes a dead lock. Without this the client only receives one message.
awaiter.Value.Fail(exception);
}
_awaiters.Clear();
_packetAwaiters.Clear();
}
public void Dispatch(MqttBasePacket packet)
@@ -32,31 +34,56 @@ namespace MQTTnet.Client.PacketDispatcher
var type = packet.GetType();
var key = new Tuple<ushort, Type>(identifier, type);
if (_awaiters.TryRemove(key, out var awaiter))
if (_packetAwaiters.TryRemove(key, out var awaiter))
{
Task.Run(() => awaiter.Complete(packet)); // Task.Run fixes a dead lock. Without this the client only receives one message.
awaiter.Complete(packet);
return;
}
throw new InvalidOperationException($"Packet of type '{type.Name}' not handled or dispatched.");
lock (_inboundPackagesQueue)
{
_inboundPackagesQueue.Add(packet);
}
}
public MqttBasePacket Take(CancellationToken cancellationToken)
{
BlockingCollection<MqttBasePacket> inboundPackagesQueue;
lock (_inboundPackagesQueue)
{
inboundPackagesQueue = _inboundPackagesQueue;
}
return inboundPackagesQueue.Take(cancellationToken);
}
public void Reset()
{
_awaiters.Clear();
foreach (var awaiter in _packetAwaiters)
{
awaiter.Value.Cancel();
}
lock (_inboundPackagesQueue)
{
_inboundPackagesQueue?.Dispose();
_inboundPackagesQueue = new BlockingCollection<MqttBasePacket>();
}
_packetAwaiters.Clear();
}
public MqttPacketAwaiter<TResponsePacket> AddPacketAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
{
var awaiter = new MqttPacketAwaiter<TResponsePacket>();
if (!identifier.HasValue)
{
identifier = 0;
}
var awaiter = new MqttPacketAwaiter<TResponsePacket>(identifier, this);
var key = new Tuple<ushort, Type>(identifier.Value, typeof(TResponsePacket));
if (!_awaiters.TryAdd(key, awaiter))
if (!_packetAwaiters.TryAdd(key, awaiter))
{
throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{key.Item2.Name}' with identifier {key.Item1}.");
}
@@ -71,8 +98,8 @@ namespace MQTTnet.Client.PacketDispatcher
identifier = 0;
}
var key = new Tuple<ushort, Type>(identifier ?? 0, typeof(TResponsePacket));
_awaiters.TryRemove(key, out _);
var key = new Tuple<ushort, Type>(identifier.Value, typeof(TResponsePacket));
_packetAwaiters.TryRemove(key, out _);
}
}
}
Source/MQTTnet/Server/MqttClientSession.cs
+93 -64
View File
@@ -7,6 +7,8 @@ using MQTTnet.Adapter;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.MessageStream;
using MQTTnet.PacketDispatcher;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -15,6 +17,7 @@ namespace MQTTnet.Server
public class MqttClientSession : IMqttClientSession
{
private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
private readonly MqttRetainedMessagesManager _retainedMessagesManager;
private readonly MqttServerEventDispatcher _eventDispatcher;
@@ -56,7 +59,7 @@ namespace MQTTnet.Server
_keepAliveMonitor = new MqttClientKeepAliveMonitor(this, _logger);
_subscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, eventDispatcher);
_pendingMessagesQueue = new MqttClientSessionPendingMessagesQueue(_options, this, _logger);
_pendingMessagesQueue = new MqttClientSessionPendingMessagesQueue(_options, this, _packetDispatcher, _logger);
}
public string ClientId { get; }
@@ -144,13 +147,14 @@ namespace MQTTnet.Server
}
var publishPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(applicationMessage);
publishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
// 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();
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
}
if (_options.ClientMessageQueueInterceptor != null)
@@ -178,6 +182,8 @@ namespace MQTTnet.Server
_pendingMessagesQueue.Enqueue(publishPacket);
}
private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
{
if (channelAdapter == null) throw new ArgumentNullException(nameof(channelAdapter));
@@ -212,13 +218,41 @@ namespace MQTTnet.Server
_isCleanSession = false;
Task.Run(async () =>
{
while (!cancellationTokenSource.IsCancellationRequested)
{
var packet = _packetDispatcher.Take(cancellationTokenSource.Token);
await ProcessReceivedPacketAsync(packet, cancellationTokenSource.Token).ConfigureAwait(false);
}
}, cancellationTokenSource.Token);
Task.Run(async () =>
{
while (!cancellationTokenSource.IsCancellationRequested)
{
try
{
var packet = await _outboundMessageStream.TakeAsync(cancellationTokenSource.Token);
await channelAdapter.SendPacketAsync(packet, cancellationTokenSource.Token);
}
catch (Exception e)
{
_logger.Error(e, "sdfsdf");
await StopAsync(MqttClientDisconnectType.NotClean);
}
}
},cancellationTokenSource.Token);
while (!cancellationTokenSource.IsCancellationRequested)
{
var packet = await channelAdapter.ReceivePacketAsync(TimeSpan.Zero, cancellationTokenSource.Token).ConfigureAwait(false);
if (packet != null)
{
_keepAliveMonitor.PacketReceived(packet);
await ProcessReceivedPacketAsync(channelAdapter, packet, cancellationTokenSource.Token).ConfigureAwait(false);
_packetDispatcher.Dispatch(packet);
}
}
}
@@ -229,17 +263,7 @@ namespace MQTTnet.Server
{
if (exception is MqttCommunicationException)
{
if (exception is MqttCommunicationClosedGracefullyException)
{
_logger.Verbose("Client '{0}': Connection closed gracefully.", ClientId);
StopInternal(MqttClientDisconnectType.Clean);
return;
}
else
{
_logger.Warning(exception, "Client '{0}': Communication exception while receiving client packets.", ClientId);
}
_logger.Warning(exception, "Client '{0}': Communication exception while receiving client packets.", ClientId);
}
else
{
@@ -256,18 +280,18 @@ namespace MQTTnet.Server
}
_willMessage = null;
_channelAdapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
_channelAdapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
_channelAdapter = null;
_logger.Info("Client '{0}': Session stopped.", ClientId);
_eventDispatcher.OnClientDisconnected(ClientId, _wasCleanDisconnect);
_workerTask = null;
}
}
private void StopInternal(MqttClientDisconnectType type)
{
var cts = _cancellationTokenSource;
@@ -278,55 +302,33 @@ namespace MQTTnet.Server
_wasCleanDisconnect = type == MqttClientDisconnectType.Clean;
_cancellationTokenSource?.Cancel(false);
_packetDispatcher.Reset();
}
private Task ProcessReceivedPacketAsync(IMqttChannelAdapter channelAdapter, MqttBasePacket packet, CancellationToken cancellationToken)
private readonly MqttMessageStream _outboundMessageStream = new MqttMessageStream();
private Task ProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
{
if (packet is MqttPublishPacket publishPacket)
{
return HandleIncomingPublishPacketAsync(channelAdapter, publishPacket, cancellationToken);
return HandleIncomingPublishPacketAsync(publishPacket, cancellationToken);
}
if (packet is MqttPingReqPacket)
{
return channelAdapter.SendPacketAsync(new MqttPingRespPacket(), cancellationToken);
}
if (packet is MqttPubRelPacket pubRelPacket)
{
var responsePacket = new MqttPubCompPacket
{
PacketIdentifier = pubRelPacket.PacketIdentifier,
ReasonCode = MqttPubCompReasonCode.Success
};
return channelAdapter.SendPacketAsync(responsePacket, cancellationToken);
}
if (packet is MqttPubRecPacket pubRecPacket)
{
var responsePacket = new MqttPubRelPacket
{
PacketIdentifier = pubRecPacket.PacketIdentifier,
ReasonCode = MqttPubRelReasonCode.Success
};
return channelAdapter.SendPacketAsync(responsePacket, cancellationToken);
}
if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
{
//return channelAdapter.SendPacketAsync(new MqttPingRespPacket(), cancellationToken);
_outboundMessageStream.Enqueue(new MqttPingRespPacket());
return Task.FromResult(0);
}
if (packet is MqttSubscribePacket subscribePacket)
{
return HandleIncomingSubscribePacketAsync(channelAdapter, subscribePacket, cancellationToken);
return HandleIncomingSubscribePacketAsync(subscribePacket, cancellationToken);
}
if (packet is MqttUnsubscribePacket unsubscribePacket)
{
return HandleIncomingUnsubscribePacketAsync(channelAdapter, unsubscribePacket, cancellationToken);
return HandleIncomingUnsubscribePacketAsync(unsubscribePacket, cancellationToken);
}
if (packet is MqttDisconnectPacket)
@@ -335,6 +337,18 @@ namespace MQTTnet.Server
return Task.FromResult(0);
}
//if (packet is MqttAuthPacket ||
// packet is MqttSubAckPacket ||
// packet is MqttUnsubAckPacket ||
// packet is MqttPubAckPacket ||
// packet is MqttPubCompPacket ||
// packet is MqttPubRecPacket ||
// packet is MqttPubRelPacket)
//{
// _packetDispatcher.TryDispatch(packet);
// return Task.FromResult(0);
//}
_logger.Warning(null, "Client '{0}': Received invalid packet ({1}). Closing connection.", ClientId, packet);
StopInternal(MqttClientDisconnectType.NotClean);
@@ -350,10 +364,14 @@ namespace MQTTnet.Server
}
}
private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
private async Task HandleIncomingSubscribePacketAsync(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{
var subscribeResult = await _subscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false);
await adapter.SendPacketAsync(subscribeResult.ResponsePacket, cancellationToken).ConfigureAwait(false);
_outboundMessageStream.Enqueue(subscribeResult.ResponsePacket);
//await adapter.SendPacketAsync(subscribeResult.ResponsePacket, cancellationToken).ConfigureAwait(false);
// TODO: Add "WaitForDelivery".
if (subscribeResult.CloseConnection)
{
@@ -364,13 +382,17 @@ namespace MQTTnet.Server
await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false);
}
private Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
private Task HandleIncomingUnsubscribePacketAsync(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
{
var unsubscribeResult = _subscriptionsManager.Unsubscribe(unsubscribePacket);
return adapter.SendPacketAsync(unsubscribeResult, cancellationToken);
_outboundMessageStream.Enqueue(unsubscribeResult);
//return adapter.SendPacketAsync(unsubscribeResult, cancellationToken);
return Task.FromResult(0);
}
private Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
private Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
Interlocked.Increment(ref _receivedMessagesCount);
@@ -382,11 +404,11 @@ namespace MQTTnet.Server
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
return HandleIncomingPublishPacketWithQoS1Async(adapter, publishPacket, cancellationToken);
return HandleIncomingPublishPacketWithQoS1Async(publishPacket, cancellationToken);
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
return HandleIncomingPublishPacketWithQoS2Async(adapter, publishPacket, cancellationToken);
return HandleIncomingPublishPacketWithQoS2Async(publishPacket, cancellationToken);
}
default:
{
@@ -405,7 +427,6 @@ namespace MQTTnet.Server
}
private Task HandleIncomingPublishPacketWithQoS1Async(
IMqttChannelAdapter adapter,
MqttPublishPacket publishPacket,
CancellationToken cancellationToken)
{
@@ -419,24 +440,32 @@ namespace MQTTnet.Server
ReasonCode = MqttPubAckReasonCode.Success
};
return adapter.SendPacketAsync(response, cancellationToken);
_outboundMessageStream.Enqueue(response);
//return adapter.SendPacketAsync(response, cancellationToken);
return Task.FromResult(0);
}
private Task HandleIncomingPublishPacketWithQoS2Async(
IMqttChannelAdapter adapter,
private async Task HandleIncomingPublishPacketWithQoS2Async(
MqttPublishPacket publishPacket,
CancellationToken cancellationToken)
{
// QoS 2 is implement as method "B" (4.3.3 QoS 2: Exactly once delivery)
_sessionsManager.EnqueueApplicationMessage(this, _channelAdapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket));
var response = new MqttPubRecPacket
using (var pubRelPacketAwaiter = _packetDispatcher.AddPacketAwaiter<MqttPubRelPacket>(publishPacket.PacketIdentifier))
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubRecReasonCode.Success
};
var pubRecPacket = new MqttPubRecPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubRecReasonCode.Success
};
return adapter.SendPacketAsync(response, cancellationToken);
//await adapter.SendPacketAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
_outboundMessageStream.Enqueue(pubRecPacket);
await pubRelPacketAwaiter.WaitOneAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
}
}
private void OnAdapterReadingPacketCompleted(object sender, EventArgs e)
Source/MQTTnet/Server/MqttClientSessionPendingMessagesQueue.cs
+42 -11
View File
@@ -6,6 +6,7 @@ using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Internal;
using MQTTnet.PacketDispatcher;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -13,20 +14,26 @@ namespace MQTTnet.Server
{
public class MqttClientSessionPendingMessagesQueue : IDisposable
{
private readonly Queue<MqttBasePacket> _queue = new Queue<MqttBasePacket>();
private readonly Queue<MqttPublishPacket> _queue = new Queue<MqttPublishPacket>();
private readonly AsyncAutoResetEvent _queueLock = new AsyncAutoResetEvent();
private readonly IMqttServerOptions _options;
private readonly MqttClientSession _clientSession;
private readonly MqttPacketDispatcher _packetDispatcher;
private readonly IMqttNetChildLogger _logger;
private long _sentPacketsCount;
public MqttClientSessionPendingMessagesQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetChildLogger logger)
public MqttClientSessionPendingMessagesQueue(
IMqttServerOptions options,
MqttClientSession clientSession,
MqttPacketDispatcher packetDispatcher,
IMqttNetChildLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
_options = options ?? throw new ArgumentNullException(nameof(options));
_clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
_packetDispatcher = packetDispatcher ?? throw new ArgumentNullException(nameof(packetDispatcher));
_logger = logger.CreateChildLogger(nameof(MqttClientSessionPendingMessagesQueue));
}
@@ -115,7 +122,7 @@ namespace MQTTnet.Server
private async Task TrySendNextQueuedPacketAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
{
MqttBasePacket packet = null;
MqttPublishPacket packet = null;
try
{
if (cancellationToken.IsCancellationRequested)
@@ -137,7 +144,34 @@ namespace MQTTnet.Server
return;
}
await adapter.SendPacketAsync(packet, cancellationToken).ConfigureAwait(false);
if (packet.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
{
await adapter.SendPacketAsync(packet, cancellationToken).ConfigureAwait(false);
}
else if (packet.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{
var awaiter = _packetDispatcher.AddPacketAwaiter<MqttPubAckPacket>(packet.PacketIdentifier);
await adapter.SendPacketAsync(packet, cancellationToken).ConfigureAwait(false);
await awaiter.WaitOneAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
}
else if (packet.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
{
var awaiter1 = _packetDispatcher.AddPacketAwaiter<MqttPubRecPacket>(packet.PacketIdentifier);
var awaiter2 = _packetDispatcher.AddPacketAwaiter<MqttPubCompPacket>(packet.PacketIdentifier);
try
{
await adapter.SendPacketAsync(packet, cancellationToken).ConfigureAwait(false);
await awaiter1.WaitOneAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
await adapter.SendPacketAsync(new MqttPubRelPacket { PacketIdentifier = packet.PacketIdentifier }, cancellationToken).ConfigureAwait(false);
await awaiter2.WaitOneAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
}
finally
{
_packetDispatcher.RemovePacketAwaiter<MqttPubRecPacket>(packet.PacketIdentifier);
_packetDispatcher.RemovePacketAwaiter<MqttPubCompPacket>(packet.PacketIdentifier);
}
}
_logger.Verbose("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
@@ -153,7 +187,7 @@ namespace MQTTnet.Server
{
_logger.Warning(exception, "Sending publish packet failed: Communication exception (ClientId: {0}).", _clientSession.ClientId);
}
else if (exception is OperationCanceledException)
else if (exception is OperationCanceledException && cancellationToken.IsCancellationRequested)
{
}
else
@@ -161,14 +195,11 @@ namespace MQTTnet.Server
_logger.Error(exception, "Sending publish packet failed (ClientId: {0}).", _clientSession.ClientId);
}
if (packet is MqttPublishPacket publishPacket)
if (packet?.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
publishPacket.Dup = true;
packet.Dup = true;
Enqueue(publishPacket);
}
Enqueue(packet);
}
if (!cancellationToken.IsCancellationRequested)
Source/MQTTnet/Server/MqttServer.cs
+14 -1
View File
@@ -5,6 +5,7 @@ using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving;
using MQTTnet.Diagnostics;
namespace MQTTnet.Server
@@ -31,7 +32,18 @@ namespace MQTTnet.Server
_eventDispatcher.ClientDisconnected += (s, e) => ClientDisconnected?.Invoke(s, e);
_eventDispatcher.ClientSubscribedTopic += (s, e) => ClientSubscribedTopic?.Invoke(s, e);
_eventDispatcher.ClientUnsubscribedTopic += (s, e) => ClientUnsubscribedTopic?.Invoke(s, e);
_eventDispatcher.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
_eventDispatcher.ApplicationMessageReceived += async (s, e) =>
{
// TODO: Migrate EventDispatcher to proper handlers and no events anymore.
ApplicationMessageReceived?.Invoke(s, e);
var handler = ReceivedApplicationMessageHandler;
if (handler != null)
{
await handler.HandleApplicationMessageAsync(
new MqttApplicationMessageHandlerContext(e.ClientId, e.ApplicationMessage)).ConfigureAwait(false);
}
};
}
public event EventHandler Started;
@@ -42,6 +54,7 @@ namespace MQTTnet.Server
public event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
public event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
public IMqttApplicationMessageHandler ReceivedApplicationMessageHandler { get; set; }
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public IMqttServerOptions Options { get; private set; }