Add ID property for the ManagedMqttApplicationMessage.

Extensions/MQTTnet.Extensions.ManagedClient/ApplicationMessageProcessedEventArgs.cs

@@ -4,13 +4,13 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public class ApplicationMessageProcessedEventArgs : EventArgs
     {
-        public ApplicationMessageProcessedEventArgs(MqttApplicationMessage applicationMessage, Exception exception)
+        public ApplicationMessageProcessedEventArgs(ManagedMqttApplicationMessage applicationMessage, Exception exception)
         {
             ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
             Exception = exception;
         }
 
-        public MqttApplicationMessage ApplicationMessage { get; }
+        public ManagedMqttApplicationMessage ApplicationMessage { get; }
         public Exception Exception { get; }
 
         public bool HasFailed => Exception != null;

Extensions/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs

@@ -20,5 +20,7 @@ namespace MQTTnet.Extensions.ManagedClient
 
         Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters);
         Task UnsubscribeAsync(IEnumerable<string> topics);
+
+        Task PublishAsync(IEnumerable<ManagedMqttApplicationMessage> applicationMessages);
     }
 }

Extensions/MQTTnet.Extensions.ManagedClient/IManagedMqttClientStorage.cs

@@ -5,8 +5,8 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public interface IManagedMqttClientStorage
     {
-        Task SaveQueuedMessagesAsync(IList<MqttApplicationMessage> messages);
+        Task SaveQueuedMessagesAsync(IList<ManagedMqttApplicationMessage> messages);
 
-        Task<IList<MqttApplicationMessage>> LoadQueuedMessagesAsync();
+        Task<IList<ManagedMqttApplicationMessage>> LoadQueuedMessagesAsync();
     }
 }

Extensions/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessageBuilder.cs

@@ -0,0 +1,47 @@
+using System;
+
+namespace MQTTnet.Extensions.ManagedClient
+{
+    public class ManagedMqttApplicationMessageBuilder
+    {
+        private Guid _id = Guid.NewGuid();
+        private MqttApplicationMessage _applicationMessage;
+
+        public ManagedMqttApplicationMessageBuilder WithId(Guid id)
+        {
+            _id = id;
+            return this;
+        }
+
+        public ManagedMqttApplicationMessageBuilder WithApplicationMessage(MqttApplicationMessage applicationMessage)
+        {
+            _applicationMessage = applicationMessage;
+            return this;
+        }
+
+        public ManagedMqttApplicationMessageBuilder WithApplicationMessage(Action<MqttApplicationMessageBuilder> builder)
+        {
+            if (builder == null) throw new ArgumentNullException(nameof(builder));
+
+            var internalBuilder = new MqttApplicationMessageBuilder();
+            builder(internalBuilder);
+
+            _applicationMessage = internalBuilder.Build();
+            return this;
+        }
+
+        public ManagedMqttApplicationMessage Build()
+        {
+            if (_applicationMessage == null)
+            {
+                throw new InvalidOperationException("The ApplicationMessage cannot be null.");
+            }
+
+            return new ManagedMqttApplicationMessage
+            {
+                Id = _id,
+                ApplicationMessage = _applicationMessage
+            };
+        }
+    }
+}

Extensions/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -14,7 +14,7 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public class ManagedMqttClient : IManagedMqttClient
     {
-        private readonly BlockingCollection<MqttApplicationMessage> _messageQueue = new BlockingCollection<MqttApplicationMessage>();
+        private readonly BlockingCollection<ManagedMqttApplicationMessage> _messageQueue = new BlockingCollection<ManagedMqttApplicationMessage>();
         private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         private readonly AsyncLock _subscriptionsLock = new AsyncLock();
         private readonly List<string> _unsubscriptions = new List<string>();
@@ -99,7 +99,15 @@ namespace MQTTnet.Extensions.ManagedClient
             return Task.FromResult(0);
         }
 
-        public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
+        public Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
+        {
+            if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
+
+            return PublishAsync(applicationMessages.Select(m =>
+                new ManagedMqttApplicationMessageBuilder().WithApplicationMessage(m).Build()));
+        }
+
+        public async Task PublishAsync(IEnumerable<ManagedMqttApplicationMessage> applicationMessages)
         {
             if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
 
@@ -244,12 +252,12 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        private async Task TryPublishQueuedMessageAsync(MqttApplicationMessage message)
+        private async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
         {
             Exception transmitException = null;
             try
             {
-                await _mqttClient.PublishAsync(message).ConfigureAwait(false);
+                await _mqttClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
 
                 if (_storageManager != null)
                 {
@@ -260,9 +268,9 @@ namespace MQTTnet.Extensions.ManagedClient
             {
                 transmitException = exception;
 
-                _logger.Warning(exception, "Publishing application message failed.");
+                _logger.Warning(exception, $"Publishing application ({message.Id}) message failed.");
 
-                if (message.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
+                if (message.ApplicationMessage.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
                 {
                     _messageQueue.Add(message);
                 }
@@ -270,7 +278,7 @@ namespace MQTTnet.Extensions.ManagedClient
             catch (Exception exception)
             {
                 transmitException = exception;
-                _logger.Error(exception, "Unhandled exception while publishing queued application message.");
+                _logger.Error(exception, $"Unhandled exception while publishing application message ({message.Id}).");
             }
             finally
             {
@@ -294,7 +302,7 @@ namespace MQTTnet.Extensions.ManagedClient
 
                 _subscriptionsNotPushed = false;
             }
-            
+
             if (!subscriptions.Any() && !unsubscriptions.Any())
             {
                 return;

Extensions/MQTTnet.Extensions.ManagedClient/ManagedMqttClientStorageManager.cs

@@ -8,7 +8,7 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public class ManagedMqttClientStorageManager
     {
-        private readonly List<MqttApplicationMessage> _messages = new List<MqttApplicationMessage>();
+        private readonly List<ManagedMqttApplicationMessage> _messages = new List<ManagedMqttApplicationMessage>();
         private readonly AsyncLock _messagesLock = new AsyncLock();
 
         private readonly IManagedMqttClientStorage _storage;
@@ -18,7 +18,7 @@ namespace MQTTnet.Extensions.ManagedClient
             _storage = storage ?? throw new ArgumentNullException(nameof(storage));
         }
 
-        public async Task<List<MqttApplicationMessage>> LoadQueuedMessagesAsync()
+        public async Task<List<ManagedMqttApplicationMessage>> LoadQueuedMessagesAsync()
         {
             var loadedMessages = await _storage.LoadQueuedMessagesAsync().ConfigureAwait(false);
             _messages.AddRange(loadedMessages);
@@ -26,7 +26,7 @@ namespace MQTTnet.Extensions.ManagedClient
             return _messages;
         }
 
-        public async Task AddAsync(MqttApplicationMessage applicationMessage)
+        public async Task AddAsync(ManagedMqttApplicationMessage applicationMessage)
         {
             using (await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
             {
@@ -35,7 +35,7 @@ namespace MQTTnet.Extensions.ManagedClient
             }
         }
 
-        public async Task RemoveAsync(MqttApplicationMessage applicationMessage)
+        public async Task RemoveAsync(ManagedMqttApplicationMessage applicationMessage)
         {
             using (await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
             {