Add support to not persistent Queue

Added new class MqttClientQueued

Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs

@@ -9,5 +9,10 @@ namespace MQTTnet
         {
             return new MqttClient(new MqttCommunicationAdapterFactory());
         }
+
+        public IMqttClient CreateMqttQueuedClient()
+        {
+            return new MqttClientQueued(new MqttCommunicationAdapterFactory());
+        }
     }
 }

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<ushort> _unacknowledgedPublishPackets;
+
+        private MqttClientOptions _options;
+        private bool _isReceivingPackets;
+        private int _latestPacketIdentifier;
+        private CancellationTokenSource _cancellationTokenSource;
+        private IMqttCommunicationAdapter _adapter;
+
+        //added
+        private readonly ConcurrentQueue<MqttApplicationMessage> _inflightQueue;
+        private bool _usePersistance = false;
+        private ConcurrentQueue<MqttApplicationMessage> _persistentQueue;
+        #endregion
+
+        #region Ctrs
+        public MqttClientQueued(IMqttCommunicationAdapterFactory communicationChannelFactory)
+        {
+            _communicationChannelFactory = communicationChannelFactory ?? throw new ArgumentNullException(nameof(communicationChannelFactory));
+            _packetDispatcher = new MqttPacketDispatcher();
+            _unacknowledgedPublishPackets = new HashSet<ushort>();
+            _inflightQueue = new ConcurrentQueue<MqttApplicationMessage>();
+            _persistentQueue = new ConcurrentQueue<MqttApplicationMessage>();
+        }
+        #endregion
+
+        #region Events
+        public event EventHandler Connected;
+        public event EventHandler Disconnected;
+        public event EventHandler<MqttApplicationMessageReceivedEventArgs> 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<MqttApplicationMessage> 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<IList<MqttSubscribeResult>> SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
+        {
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            ThrowIfNotConnected();
+
+            var subscribePacket = new MqttSubscribePacket
+            {
+                PacketIdentifier = GetNewPacketIdentifier(),
+                TopicFilters = topicFilters.ToList()
+            };
+
+            var response = await SendAndReceiveAsync<MqttSubAckPacket>(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<string> topicFilters)
+        {
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            ThrowIfNotConnected();
+
+            var unsubscribePacket = new MqttUnsubscribePacket
+            {
+                PacketIdentifier = GetNewPacketIdentifier(),
+                TopicFilters = topicFilters.ToList()
+            };
+
+            await SendAndReceiveAsync<MqttUnsubAckPacket>(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<MqttConnAckPacket>(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<MqttPubCompPacket>());
+        }
+
+        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<MqttPingRespPacket>(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<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(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<MqttPubAckPacket>(publishPacket);
+                                    break;
+                                }
+                            case MqttQualityOfServiceLevel.ExactlyOnce:
+                                {
+                                    publishPacket.PacketIdentifier = GetNewPacketIdentifier();
+                                    var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
+                                    await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>()).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
+    }
+}

MQTTnet.Core/MQTTnet.Core.csproj

@@ -32,4 +32,10 @@
 
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Any CPU|x86'" />
 
+  <ItemGroup>
+    <Compile Remove="DiskQueue\**" />
+    <EmbeddedResource Remove="DiskQueue\**" />
+    <None Remove="DiskQueue\**" />
+  </ItemGroup>
+
 </Project>

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("2 = Start queued client");
-            Console.WriteLine("3 = Start performance test");
+            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<TopicFilter>
+                    {
+                        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();