Add new API for releasing publish packets

Source/MQTTnet/Client/IMqttClient.cs

@@ -8,7 +8,7 @@ namespace MQTTnet.Client
     public interface IMqttClient : IDisposable
     {
         event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync;
-        
+
         event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync;
 
         event Func<MqttClientConnectingEventArgs, Task> ConnectingAsync;
@@ -29,6 +29,8 @@ namespace MQTTnet.Client
 
         Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default);
 
+        Task<MqttClientReleasePublishPacketResult> ReleasePublishPacketAsync(MqttClientReleasePublishPacketOptions options, CancellationToken cancellationToken = default);
+
         Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default);
 
         Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default);

Source/MQTTnet/Client/MqttClient.cs

@@ -86,7 +86,7 @@ namespace MQTTnet.Client
             add => _inspectPacketEvent.AddHandler(value);
             remove => _inspectPacketEvent.RemoveHandler(value);
         }
-        
+
         public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected;
 
         public MqttClientOptions Options { get; private set; }
@@ -149,17 +149,7 @@ namespace MQTTnet.Client
 
                 _lastPacketSentTimestamp = DateTime.UtcNow;
 
-                var keepAliveInterval = Options.KeepAlivePeriod;
-                if (connectResult.ServerKeepAlive > 0)
-                {
-                    _logger.Info($"Using keep alive value ({connectResult.ServerKeepAlive}) sent from the server.");
-                    keepAliveInterval = TimeSpan.FromSeconds(connectResult.ServerKeepAlive);
-                }
-
-                if (keepAliveInterval != TimeSpan.Zero)
-                {
-                    _keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessagesAsync(backgroundCancellationToken), backgroundCancellationToken);
-                }
+                SetupKeepAliveHandling(connectResult, _backgroundCancellationTokenSource.Token);
 
                 CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connected, MqttClientConnectionStatus.Connecting);
 
@@ -234,6 +224,11 @@ namespace MQTTnet.Client
 
         public Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default)
         {
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
             cancellationToken.ThrowIfCancellationRequested();
 
             MqttTopicValidator.ThrowIfInvalid(applicationMessage);
@@ -264,6 +259,34 @@ namespace MQTTnet.Client
             }
         }
 
+        public async Task<MqttClientReleasePublishPacketResult> ReleasePublishPacketAsync(MqttClientReleasePublishPacketOptions options, CancellationToken cancellationToken = default)
+        {
+            if (options == null)
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
+
+            cancellationToken.ThrowIfCancellationRequested();
+
+            ThrowIfDisposed();
+            ThrowIfNotConnected();
+
+            if (options.PacketIdentifier == 0)
+            {
+                throw new MqttProtocolViolationException("Releasing a PUBLISH packet requires a valid packet identifier.");
+            }
+
+            if (Options.AutoReleasePublishPackets)
+            {
+                throw new InvalidOperationException("Manually releasing PUBLISH packets requires the feature _AutoReleasePublishPackets_ disabled in the client options.");
+            }
+
+            var pubRelPacket = _packetFactories.PubRel.Create(options);
+            var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
+
+            return new MqttClientReleasePublishPacketResult(pubCompPacket.PacketIdentifier, pubCompPacket.ReasonString, pubCompPacket.UserProperties);
+        }
+
         public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default)
         {
             if (data == null)
@@ -411,7 +434,7 @@ namespace MQTTnet.Client
 
                 _adapter?.Dispose();
                 _adapter = null;
-                
+
                 _packetDispatcher?.Dispose();
                 _packetDispatcher = null;
             }
@@ -448,7 +471,7 @@ namespace MQTTnet.Client
             try
             {
                 _packetDispatcher.Dispose(new MqttClientDisconnectedException(exception));
-                
+
                 var receiverTask = WaitForTaskAsync(_packetReceiverTask, sender);
                 var publishPacketReceiverTask = WaitForTaskAsync(_publishPacketReceiverTask, sender);
                 var keepAliveTask = WaitForTaskAsync(_keepAlivePacketsSenderTask, sender);
@@ -504,9 +527,12 @@ namespace MQTTnet.Client
                 {
                     case MqttClientConnectionStatus.Disconnected:
                     case MqttClientConnectionStatus.Disconnecting:
+                    {
                         return true;
+                    }
                     case MqttClientConnectionStatus.Connected:
                     case MqttClientConnectionStatus.Connecting:
+                    {
                         // This will compare the _connectionStatus to old value and set it to "MqttClientConnectionStatus.Disconnecting" afterwards.
                         // So the first caller will get a "false" and all subsequent ones will get "true".
                         var curStatus = CompareExchangeConnectionStatus(MqttClientConnectionStatus.Disconnecting, connectionStatus);
@@ -517,6 +543,7 @@ namespace MQTTnet.Client
 
                         connectionStatus = curStatus;
                         break;
+                    }
                 }
             } while (true);
         }
@@ -628,7 +655,7 @@ namespace MQTTnet.Client
             // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
             await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false);
 
-            return _clientPublishResultFactory.Create(null);
+            return _clientPublishResultFactory.Create((MqttPubAckPacket)null);
         }
 
         async Task<MqttClientPublishResult> PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
@@ -637,10 +664,13 @@ namespace MQTTnet.Client
 
             var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
 
+            if (!Options.AutoReleasePublishPackets)
+            {
+                return _clientPublishResultFactory.Create(pubRecPacket);
+            }
+
             var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
-
             var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
-
             return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket);
         }
 
@@ -691,6 +721,32 @@ namespace MQTTnet.Client
             return _adapter.SendPacketAsync(packet, cancellationToken);
         }
 
+        void SetupKeepAliveHandling(MqttClientConnectResult connectResult, CancellationToken cancellationToken)
+        {
+            if (Options.UseExternalKeepAliveManagement)
+            {
+                _logger.Info("Using external keep alive management.");
+                return;
+            }
+
+            var keepAliveInterval = Options.KeepAlivePeriod;
+
+            if (connectResult.ServerKeepAlive > 0)
+            {
+                _logger.Info("Using keep alive value {0} sent from the server.", connectResult.ServerKeepAlive);
+                keepAliveInterval = TimeSpan.FromSeconds(connectResult.ServerKeepAlive);
+            }
+
+            if (keepAliveInterval != TimeSpan.Zero)
+            {
+                _keepAlivePacketsSenderTask = Task.Factory.StartNew(
+                    () => TrySendKeepAliveMessagesAsync(cancellationToken),
+                    cancellationToken,
+                    TaskCreationOptions.PreferFairness,
+                    TaskScheduler.Default);
+            }
+        }
+
         void ThrowIfConnected(string message)
         {
             if (IsConnected)
@@ -833,7 +889,7 @@ namespace MQTTnet.Client
                 {
                     exception = aggregateException.GetBaseException();
                 }
-                
+
                 if (exception is OperationCanceledException)
                 {
                 }

Source/MQTTnet/Client/MqttClientExtensions.cs

@@ -80,6 +80,23 @@ namespace MQTTnet.Client
             return client.ConnectAsync(client.Options, cancellationToken);
         }
 
+        public static Task<MqttClientReleasePublishPacketResult> ReleasePublishPacketAsync(
+            this IMqttClient mqttClient,
+            ushort packetIdentifier,
+            string reasonString = null,
+            List<MqttUserProperty> userProperties = null,
+            CancellationToken cancellationToken = default)
+        {
+            var options = new MqttClientReleasePublishPacketOptions
+            {
+                PacketIdentifier = packetIdentifier,
+                ReasonString = reasonString,
+                UserProperties = userProperties
+            };
+
+            return mqttClient.ReleasePublishPacketAsync(options, cancellationToken);
+        }
+
         public static Task SendExtendedAuthenticationExchangeDataAsync(this IMqttClient client, MqttExtendedAuthenticationExchangeData data)
         {
             if (client == null)

Source/MQTTnet/Client/Options/MqttClientOptions.cs

@@ -24,6 +24,11 @@ namespace MQTTnet.Client
         /// </summary>
         public string AuthenticationMethod { get; set; }
 
+        /// <summary>
+        ///     Gets or sets whether PUBLISH packets with QoS level 2 should be released automatically.
+        /// </summary>
+        public bool AutoReleasePublishPackets { get; set; } = true;
+
         public IMqttClientChannelOptions ChannelOptions { get; set; }
 
         /// <summary>
@@ -108,6 +113,14 @@ namespace MQTTnet.Client
         /// </summary>
         public bool TryPrivate { get; set; } = true;
 
+        /// <summary>
+        ///     Gets or sets whether the sending of keep alive packets (PINGREQ, PINGRESP) is handled externally (False by
+        ///     default).
+        ///     If set to true the user of this library must make sure that PINGREQ packets are sent according to the sent keep
+        ///     alive interval.
+        /// </summary>
+        public bool UseExternalKeepAliveManagement { get; set; }
+
         /// <summary>
         ///     Gets or sets the user properties.
         ///     In MQTT 5, user properties are basic UTF-8 string key-value pairs that you can append to almost every type of MQTT

Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs

@@ -33,6 +33,17 @@ namespace MQTTnet.Client
             return result;
         }
 
+        public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket)
+        {
+            return new MqttClientPublishResult
+            {
+                PacketIdentifier = pubRecPacket.PacketIdentifier,
+                ReasonCode = (MqttClientPublishReasonCode)(int)pubRecPacket.ReasonCode,
+                ReasonString = pubRecPacket.ReasonString,
+                UserProperties = pubRecPacket.UserProperties
+            };
+        }
+        
         public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
         {
             if (pubRecPacket == null || pubCompPacket == null)

Source/MQTTnet/Client/Publishing/MqttClientReleasePublishPacketOptions.cs

@@ -0,0 +1,30 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Collections.Generic;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Client
+{
+    public sealed class MqttClientReleasePublishPacketOptions
+    {
+        /// <summary>
+        ///     Gets or sets the identifier of the PUBLISH packet which should be released.
+        /// </summary>
+        public ushort PacketIdentifier { get; set; }
+        
+        public MqttPubRelReasonCode ReasonCode { get; set; } = MqttPubRelReasonCode.Success;
+
+        /// <summary>
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public string ReasonString { get; set; }
+
+        /// <summary>
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public List<MqttUserProperty> UserProperties { get; set; }
+    }
+}

Source/MQTTnet/Client/Publishing/MqttClientReleasePublishPacketResult.cs

@@ -0,0 +1,31 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Collections.Generic;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Client
+{
+    public sealed class MqttClientReleasePublishPacketResult
+    {
+        public MqttClientReleasePublishPacketResult(uint packetIdentifier, string reasonString, List<MqttUserProperty> userProperties)
+        {
+            PacketIdentifier = packetIdentifier;
+            ReasonString = reasonString;
+            UserProperties = userProperties;
+        }
+
+        /// <summary>
+        ///     Gets or sets the identifier which was used in the PUBCOMP packet.
+        /// </summary>
+        public uint PacketIdentifier { get; }
+
+        public string ReasonString { get; }
+
+        /// <summary>
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public List<MqttUserProperty> UserProperties { get; }
+    }
+}

Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs

@@ -24,5 +24,21 @@ namespace MQTTnet.Formatter
                 ReasonCode = (MqttPubRelReasonCode)(int)reasonCode
             };
         }
+
+        public MqttPubRelPacket Create(MqttClientReleasePublishPacketOptions options)
+        {
+            if (options == null)
+            {
+                throw new ArgumentNullException(nameof(options));
+            }
+
+            return new MqttPubRelPacket
+            {
+                PacketIdentifier = options.PacketIdentifier,
+                ReasonCode = options.ReasonCode,
+                ReasonString = options.ReasonString,
+                UserProperties = options.UserProperties
+            };
+        }
     }
 }