diff --git a/Source/MQTTnet/Client/IMqttClient.cs b/Source/MQTTnet/Client/IMqttClient.cs index 849bff12..f0787e75 100644 --- a/Source/MQTTnet/Client/IMqttClient.cs +++ b/Source/MQTTnet/Client/IMqttClient.cs @@ -8,7 +8,7 @@ namespace MQTTnet.Client public interface IMqttClient : IDisposable { event Func ApplicationMessageReceivedAsync; - + event Func ConnectedAsync; event Func ConnectingAsync; @@ -29,6 +29,8 @@ namespace MQTTnet.Client Task PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default); + Task ReleasePublishPacketAsync(MqttClientReleasePublishPacketOptions options, CancellationToken cancellationToken = default); + Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default); Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default); diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 50e9223a..4b752194 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/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 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 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(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 PublishExactlyOnceAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) @@ -637,10 +664,13 @@ namespace MQTTnet.Client var pubRecPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); + if (!Options.AutoReleasePublishPackets) + { + return _clientPublishResultFactory.Create(pubRecPacket); + } + var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); - var pubCompPacket = await SendAndReceiveAsync(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) { } diff --git a/Source/MQTTnet/Client/MqttClientExtensions.cs b/Source/MQTTnet/Client/MqttClientExtensions.cs index ba917aad..c32d6858 100644 --- a/Source/MQTTnet/Client/MqttClientExtensions.cs +++ b/Source/MQTTnet/Client/MqttClientExtensions.cs @@ -80,6 +80,23 @@ namespace MQTTnet.Client return client.ConnectAsync(client.Options, cancellationToken); } + public static Task ReleasePublishPacketAsync( + this IMqttClient mqttClient, + ushort packetIdentifier, + string reasonString = null, + List 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) diff --git a/Source/MQTTnet/Client/Options/MqttClientOptions.cs b/Source/MQTTnet/Client/Options/MqttClientOptions.cs index 931ece6f..53f1ae64 100644 --- a/Source/MQTTnet/Client/Options/MqttClientOptions.cs +++ b/Source/MQTTnet/Client/Options/MqttClientOptions.cs @@ -24,6 +24,11 @@ namespace MQTTnet.Client /// public string AuthenticationMethod { get; set; } + /// + /// Gets or sets whether PUBLISH packets with QoS level 2 should be released automatically. + /// + public bool AutoReleasePublishPackets { get; set; } = true; + public IMqttClientChannelOptions ChannelOptions { get; set; } /// @@ -108,6 +113,14 @@ namespace MQTTnet.Client /// public bool TryPrivate { get; set; } = true; + /// + /// 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. + /// + public bool UseExternalKeepAliveManagement { get; set; } + /// /// 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 diff --git a/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs b/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs index ea5bc233..c8550a6b 100644 --- a/Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs +++ b/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) diff --git a/Source/MQTTnet/Client/Publishing/MqttClientReleasePublishPacketOptions.cs b/Source/MQTTnet/Client/Publishing/MqttClientReleasePublishPacketOptions.cs new file mode 100644 index 00000000..ad550ddc --- /dev/null +++ b/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 + { + /// + /// Gets or sets the identifier of the PUBLISH packet which should be released. + /// + public ushort PacketIdentifier { get; set; } + + public MqttPubRelReasonCode ReasonCode { get; set; } = MqttPubRelReasonCode.Success; + + /// + /// MQTT 5.0.0+ feature. + /// + public string ReasonString { get; set; } + + /// + /// MQTT 5.0.0+ feature. + /// + public List UserProperties { get; set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Client/Publishing/MqttClientReleasePublishPacketResult.cs b/Source/MQTTnet/Client/Publishing/MqttClientReleasePublishPacketResult.cs new file mode 100644 index 00000000..9f3dd3db --- /dev/null +++ b/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 userProperties) + { + PacketIdentifier = packetIdentifier; + ReasonString = reasonString; + UserProperties = userProperties; + } + + /// + /// Gets or sets the identifier which was used in the PUBCOMP packet. + /// + public uint PacketIdentifier { get; } + + public string ReasonString { get; } + + /// + /// MQTT 5.0.0+ feature. + /// + public List UserProperties { get; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs index a16790b6..9b999c54 100644 --- a/Source/MQTTnet/Formatter/MqttPubRelPacketFactory.cs +++ b/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 + }; + } } } \ No newline at end of file