diff --git a/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs b/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs index 15da033d..93583907 100644 --- a/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs +++ b/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs @@ -9,5 +9,10 @@ namespace MQTTnet { return new MqttClient(new MqttCommunicationAdapterFactory()); } + + public IMqttClient CreateMqttQueuedClient() + { + return new MqttClientQueued(new MqttCommunicationAdapterFactory()); + } } } \ No newline at end of file diff --git a/MQTTnet.Core/Client/MqttClientQueued.cs b/MQTTnet.Core/Client/MqttClientQueued.cs new file mode 100644 index 00000000..925b9656 --- /dev/null +++ b/MQTTnet.Core/Client/MqttClientQueued.cs @@ -0,0 +1,530 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using MQTTnet.Core.Adapter; +using MQTTnet.Core.Diagnostics; +using MQTTnet.Core.Exceptions; +using MQTTnet.Core.Packets; +using MQTTnet.Core.Protocol; +using MQTTnet.Core.Internal; + +namespace MQTTnet.Core.Client +{ + public class MqttClientQueued : IMqttClient + { + #region Fields + private readonly IMqttCommunicationAdapterFactory _communicationChannelFactory; + private readonly MqttPacketDispatcher _packetDispatcher; + private readonly HashSet _unacknowledgedPublishPackets; + + private MqttClientOptions _options; + private bool _isReceivingPackets; + private int _latestPacketIdentifier; + private CancellationTokenSource _cancellationTokenSource; + private IMqttCommunicationAdapter _adapter; + + //added + private readonly ConcurrentQueue _inflightQueue; + private bool _usePersistance = false; + private ConcurrentQueue _persistentQueue; + #endregion + + #region Ctrs + public MqttClientQueued(IMqttCommunicationAdapterFactory communicationChannelFactory) + { + _communicationChannelFactory = communicationChannelFactory ?? throw new ArgumentNullException(nameof(communicationChannelFactory)); + _packetDispatcher = new MqttPacketDispatcher(); + _unacknowledgedPublishPackets = new HashSet(); + _inflightQueue = new ConcurrentQueue(); + _persistentQueue = new ConcurrentQueue(); + } + #endregion + + #region Events + public event EventHandler Connected; + public event EventHandler Disconnected; + public event EventHandler ApplicationMessageReceived; + #endregion + + #region Poperties + public bool IsConnected => _cancellationTokenSource != null && !_cancellationTokenSource.IsCancellationRequested; + #endregion + + #region MqttClient Methods + + public async Task ConnectAsync(MqttClientOptions options) + { + if (options == null) throw new ArgumentNullException(nameof(options)); + + ThrowIfConnected("It is not allowed to connect with a server after the connection is established."); + + try + { + _options = options; + _cancellationTokenSource = new CancellationTokenSource(); + _latestPacketIdentifier = 0; + _packetDispatcher.Reset(); + + _adapter = _communicationChannelFactory.CreateMqttCommunicationAdapter(options); + + MqttNetTrace.Verbose(nameof(MqttClient), "Trying to connect with server."); + await _adapter.ConnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false); + MqttNetTrace.Verbose(nameof(MqttClient), "Connection with server established."); + + await SetupIncomingPacketProcessingAsync(); + SetupOutgoingPacketProcessingAsync(); + await AuthenticateAsync(options.WillMessage); + + MqttNetTrace.Verbose(nameof(MqttClient), "MQTT connection with server established."); + + if (_options.KeepAlivePeriod != TimeSpan.Zero) + { + StartSendKeepAliveMessages(_cancellationTokenSource.Token); + } + + Connected?.Invoke(this, EventArgs.Empty); + } + catch (Exception) + { + await DisconnectInternalAsync().ConfigureAwait(false); + throw; + } + } + + public async Task DisconnectAsync() + { + if (!IsConnected) + { + return; + } + + try + { + await SendAsync(new MqttDisconnectPacket()).ConfigureAwait(false); + } + finally + { + await DisconnectInternalAsync().ConfigureAwait(false); + } + } + + public async Task PublishAsync(IEnumerable applicationMessages) + { +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + ThrowIfNotConnected(); + + foreach (var applicationMessage in applicationMessages) + { + if (_usePersistance) + _persistentQueue.Enqueue(applicationMessage); + _inflightQueue.Enqueue(applicationMessage); + } +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + } + + public async Task> SubscribeAsync(IEnumerable topicFilters) + { + if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); + + ThrowIfNotConnected(); + + var subscribePacket = new MqttSubscribePacket + { + PacketIdentifier = GetNewPacketIdentifier(), + TopicFilters = topicFilters.ToList() + }; + + var response = await SendAndReceiveAsync(subscribePacket).ConfigureAwait(false); + + if (response.SubscribeReturnCodes.Count != subscribePacket.TopicFilters.Count) + { + throw new MqttProtocolViolationException("The return codes are not matching the topic filters [MQTT-3.9.3-1]."); + } + + return subscribePacket.TopicFilters.Select((t, i) => new MqttSubscribeResult(t, response.SubscribeReturnCodes[i])).ToList(); + } + + public async Task UnsubscribeAsync(IEnumerable topicFilters) + { + if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); + + ThrowIfNotConnected(); + + var unsubscribePacket = new MqttUnsubscribePacket + { + PacketIdentifier = GetNewPacketIdentifier(), + TopicFilters = topicFilters.ToList() + }; + + await SendAndReceiveAsync(unsubscribePacket); + } + + #region private + private void ThrowIfNotConnected() + { + if (!IsConnected) throw new MqttCommunicationException("The client is not connected."); + } + + private void ThrowIfConnected(string message) + { + if (IsConnected) throw new MqttProtocolViolationException(message); + } + + private async Task SetupIncomingPacketProcessingAsync() + { + _isReceivingPackets = false; + +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Factory.StartNew( + () => ReceivePackets(_cancellationTokenSource.Token), + _cancellationTokenSource.Token, + TaskCreationOptions.LongRunning, + TaskScheduler.Default).ConfigureAwait(false); +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + + while (!_isReceivingPackets && _cancellationTokenSource != null && !_cancellationTokenSource.IsCancellationRequested) + { + await Task.Delay(TimeSpan.FromMilliseconds(100)); + } + } + + private async Task AuthenticateAsync(MqttApplicationMessage willApplicationMessage) + { + var connectPacket = new MqttConnectPacket + { + ClientId = _options.ClientId, + Username = _options.UserName, + Password = _options.Password, + CleanSession = _options.CleanSession, + KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds, + WillMessage = willApplicationMessage + }; + + var response = await SendAndReceiveAsync(connectPacket).ConfigureAwait(false); + if (response.ConnectReturnCode != MqttConnectReturnCode.ConnectionAccepted) + { + throw new MqttConnectingFailedException(response.ConnectReturnCode); + } + } + + private async Task ReceivePackets(CancellationToken cancellationToken) + { + MqttNetTrace.Information(nameof(MqttClient), "Start receiving packets."); + + try + { + while (!cancellationToken.IsCancellationRequested) + { + _isReceivingPackets = true; + + var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false); + if (cancellationToken.IsCancellationRequested) + { + return; + } + + StartProcessReceivedPacket(packet, cancellationToken); + } + } + catch (OperationCanceledException) + { + } + catch (MqttCommunicationException exception) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + MqttNetTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while receiving packets."); + await DisconnectInternalAsync().ConfigureAwait(false); + } + catch (Exception exception) + { + MqttNetTrace.Error(nameof(MqttClient), exception, "Unhandled exception while receiving packets."); + await DisconnectInternalAsync().ConfigureAwait(false); + } + finally + { + MqttNetTrace.Information(nameof(MqttClient), "Stopped receiving packets."); + } + } + + private void StartProcessReceivedPacket(MqttBasePacket packet, CancellationToken cancellationToken) + { +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Run(() => ProcessReceivedPacketAsync(packet), cancellationToken).ConfigureAwait(false); +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + } + + private async Task ProcessReceivedPacketAsync(MqttBasePacket packet) + { + try + { + MqttNetTrace.Information(nameof(MqttClient), "Received <<< {0}", packet); + + if (packet is MqttPingReqPacket) + { + await SendAsync(new MqttPingRespPacket()); + return; + } + + if (packet is MqttDisconnectPacket) + { + await DisconnectAsync(); + return; + } + + if (packet is MqttPublishPacket publishPacket) + { + await ProcessReceivedPublishPacket(publishPacket); + return; + } + + if (packet is MqttPubRelPacket pubRelPacket) + { + await ProcessReceivedPubRelPacket(pubRelPacket); + return; + } + + _packetDispatcher.Dispatch(packet); + } + catch (Exception exception) + { + MqttNetTrace.Error(nameof(MqttClient), exception, "Unhandled exception while processing received packet."); + } + } + + private async Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket) + { + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) + { + FireApplicationMessageReceivedEvent(publishPacket); + return; + } + + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) + { + FireApplicationMessageReceivedEvent(publishPacket); + await SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }); + return; + } + + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) + { + // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery] + lock (_unacknowledgedPublishPackets) + { + _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier); + } + + FireApplicationMessageReceivedEvent(publishPacket); + await SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }); + return; + } + + throw new MqttCommunicationException("Received a not supported QoS level."); + } + + private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket) + { + lock (_unacknowledgedPublishPackets) + { + _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier); + } + + return SendAsync(pubRelPacket.CreateResponse()); + } + + private async Task DisconnectInternalAsync() + { + var cts = _cancellationTokenSource; + if (cts == null || cts.IsCancellationRequested) + { + return; + } + + cts.Cancel(false); + cts.Dispose(); + _cancellationTokenSource = null; + + try + { + await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false); + MqttNetTrace.Information(nameof(MqttClient), "Disconnected from adapter."); + } + catch (Exception exception) + { + MqttNetTrace.Warning(nameof(MqttClient), exception, "Error while disconnecting from adapter."); + } + finally + { + Disconnected?.Invoke(this, EventArgs.Empty); + } + } + + private void StartSendKeepAliveMessages(CancellationToken cancellationToken) + { +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Factory.StartNew(() => SendKeepAliveMessagesAsync(cancellationToken), cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default).ConfigureAwait(false); +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + } + + private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken) + { + MqttNetTrace.Information(nameof(MqttClient), "Start sending keep alive packets."); + + try + { + while (!cancellationToken.IsCancellationRequested) + { + await Task.Delay(_options.KeepAlivePeriod, cancellationToken).ConfigureAwait(false); + if (cancellationToken.IsCancellationRequested) + { + return; + } + + await SendAndReceiveAsync(new MqttPingReqPacket()).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + } + catch (MqttCommunicationException exception) + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + MqttNetTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while sending/receiving keep alive packets."); + await DisconnectInternalAsync().ConfigureAwait(false); + } + catch (Exception exception) + { + MqttNetTrace.Warning(nameof(MqttClient), exception, "Unhandled exception while sending/receiving keep alive packets."); + await DisconnectInternalAsync().ConfigureAwait(false); + } + finally + { + MqttNetTrace.Information(nameof(MqttClient), "Stopped sending keep alive packets."); + } + } + + private async Task SendAndReceiveAsync(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket + { + var packetAwaiter = _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.DefaultCommunicationTimeout); + await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, requestPacket).ConfigureAwait(false); + return (TResponsePacket)await packetAwaiter.ConfigureAwait(false); + } + + private Task SendAsync(MqttBasePacket packet) + { + return _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, packet); + } + + private ushort GetNewPacketIdentifier() + { + return (ushort)Interlocked.Increment(ref _latestPacketIdentifier); + } + + private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket) + { + try + { + var applicationMessage = publishPacket.ToApplicationMessage(); + ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage)); + } + catch (Exception exception) + { + MqttNetTrace.Error(nameof(MqttClient), exception, "Unhandled exception while handling application message."); + } + } + #endregion + #endregion + + #region MqttClientQueued Methos + private void SetupOutgoingPacketProcessingAsync() + { +#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Factory.StartNew( + () => SendPackets(_cancellationTokenSource.Token), + _cancellationTokenSource.Token, + TaskCreationOptions.LongRunning, + TaskScheduler.Default).ConfigureAwait(false); +#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + + //while (_cancellationTokenSource != null && !_cancellationTokenSource.IsCancellationRequested) + //{ + // await Task.Delay(TimeSpan.FromMilliseconds(100)); + //} + } + + private async Task SendPackets(CancellationToken cancellationToken) + { + MqttNetTrace.Information(nameof(MqttClient), "Start sending packets."); + MqttApplicationMessage messageToSend = null; + try + { + while (!cancellationToken.IsCancellationRequested) + { + while (_inflightQueue.TryDequeue(out messageToSend)) + { + MqttQualityOfServiceLevel qosGroup = messageToSend.QualityOfServiceLevel; + MqttPublishPacket publishPacket = messageToSend.ToPublishPacket(); + switch (qosGroup) + { + case MqttQualityOfServiceLevel.AtMostOnce: + { + // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] + await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, publishPacket); + break; + } + + case MqttQualityOfServiceLevel.AtLeastOnce: + { + publishPacket.PacketIdentifier = GetNewPacketIdentifier(); + await SendAndReceiveAsync(publishPacket); + break; + } + case MqttQualityOfServiceLevel.ExactlyOnce: + { + publishPacket.PacketIdentifier = GetNewPacketIdentifier(); + var pubRecPacket = await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); + await SendAndReceiveAsync(pubRecPacket.CreateResponse()).ConfigureAwait(false); + break; + } + default: + { + throw new InvalidOperationException(); + } + } + } + }; + } + catch (OperationCanceledException) + { + } + catch (MqttCommunicationException exception) + { + MqttNetTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while sending packets."); + //message not send, equeued again + if (messageToSend != null) + _inflightQueue.Enqueue(messageToSend); + } + catch (Exception exception) + { + MqttNetTrace.Error(nameof(MqttClient), exception, "Unhandled exception while sending packets."); + await DisconnectInternalAsync().ConfigureAwait(false); + } + finally + { + MqttNetTrace.Information(nameof(MqttClient), "Stopped sending packets."); + } + } + #endregion + } +} diff --git a/MQTTnet.Core/MQTTnet.Core.csproj b/MQTTnet.Core/MQTTnet.Core.csproj index 0e9b965c..1e507cb7 100644 --- a/MQTTnet.Core/MQTTnet.Core.csproj +++ b/MQTTnet.Core/MQTTnet.Core.csproj @@ -32,4 +32,10 @@ + + + + + + \ No newline at end of file diff --git a/Tests/MQTTnet.TestApp.NetCore/Program.cs b/Tests/MQTTnet.TestApp.NetCore/Program.cs index 821d8610..f2a44e0c 100644 --- a/Tests/MQTTnet.TestApp.NetCore/Program.cs +++ b/Tests/MQTTnet.TestApp.NetCore/Program.cs @@ -18,26 +18,127 @@ namespace MQTTnet.TestApp.NetCore { Console.WriteLine("MQTTnet - TestApp.NetFramework"); Console.WriteLine("1 = Start client"); - Console.WriteLine("2 = Start server"); - Console.WriteLine("3 = Start performance test"); + Console.WriteLine("2 = Start queued client"); + Console.WriteLine("3 = Start server"); + Console.WriteLine("4 = Start performance test"); var pressedKey = Console.ReadKey(true); if (pressedKey.Key == ConsoleKey.D1) { Task.Run(() => RunClientAsync(args)); Thread.Sleep(Timeout.Infinite); } - else if (pressedKey.Key == ConsoleKey.D2) + if (pressedKey.Key == ConsoleKey.D2) { - Task.Run(() => RunServerAsync(args)); + Task.Run(() => RunClientQueuedAsync(args)); Thread.Sleep(Timeout.Infinite); } else if (pressedKey.Key == ConsoleKey.D3) + { + Task.Run(() => RunServerAsync(args)); + Thread.Sleep(Timeout.Infinite); + } + else if (pressedKey.Key == ConsoleKey.D4) { Task.Run(PerformanceTest.RunAsync); Thread.Sleep(Timeout.Infinite); } } + private static async Task RunClientQueuedAsync(string[] arguments) + { + MqttNetTrace.TraceMessagePublished += (s, e) => + { + Console.WriteLine($">> [{e.ThreadId}] [{e.Source}] [{e.Level}]: {e.Message}"); + if (e.Exception != null) + { + Console.WriteLine(e.Exception); + } + }; + + try + { + var options = new MqttClientTcpOptions + { + Server = "192.168.0.14", + ClientId = "XYZ", + CleanSession = true, + UserName = "lobu", + Password = "passworda", + KeepAlivePeriod = TimeSpan.FromSeconds(31), + DefaultCommunicationTimeout = TimeSpan.FromSeconds(20), + + }; + + var client = new MqttClientFactory().CreateMqttQueuedClient(); + client.ApplicationMessageReceived += (s, e) => + { + Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###"); + Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}"); + Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}"); + Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}"); + Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}"); + Console.WriteLine(); + }; + + client.Connected += async (s, e) => + { + Console.WriteLine("### CONNECTED WITH SERVER ###"); + + await client.SubscribeAsync(new List + { + new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce) + }); + + Console.WriteLine("### SUBSCRIBED ###"); + }; + + client.Disconnected += async (s, e) => + { + Console.WriteLine("### DISCONNECTED FROM SERVER ###"); + await Task.Delay(TimeSpan.FromSeconds(5)); + + try + { + await client.ConnectAsync(options); + } + catch + { + Console.WriteLine("### RECONNECTING FAILED ###"); + } + }; + + try + { + await client.ConnectAsync(options); + } + catch (Exception exception) + { + Console.WriteLine("### CONNECTING FAILED ###" + Environment.NewLine + exception); + } + + Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###"); + + int i = 0; + while (true) + { + Console.ReadLine(); + i++; + var applicationMessage = new MqttApplicationMessage( + "PLNMAIN", + Encoding.UTF8.GetBytes(string.Format("Hello World {0}", i)), + MqttQualityOfServiceLevel.ExactlyOnce, + false + ); + + await client.PublishAsync(applicationMessage); + } + } + catch (Exception exception) + { + Console.WriteLine(exception); + } + } + private static async Task RunClientAsync(string[] arguments) { MqttNetTrace.TraceMessagePublished += (s, e) => @@ -55,7 +156,7 @@ namespace MQTTnet.TestApp.NetCore { Uri = "localhost", ClientId = "XYZ", - CleanSession = true + CleanSession = true, }; var client = new MqttClientFactory().CreateMqttClient();