mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-17 01:33:03 +08:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d003ae1f34 | ||
|
|
301c4411e5 |
@@ -243,5 +243,6 @@ See the LICENSE file in the project root for more information.</s:String>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGREQ/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGRESP/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=PUBCOMP/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=PUBREC/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=Tnet/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/UserDictionary/Words/=unsub/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
|
||||
@@ -26,7 +26,7 @@ public static class PackageInspection_Samples
|
||||
.WithTcpServer("broker.hivemq.com")
|
||||
.Build();
|
||||
|
||||
mqttClient.InspectPackage += OnInspectPackage;
|
||||
mqttClient.InspectPackageAsync += OnInspectPackage;
|
||||
|
||||
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ namespace MQTTnet.Tests.Diagnostics
|
||||
|
||||
var packets = new List<string>();
|
||||
|
||||
mqttClient.InspectPackage += eventArgs =>
|
||||
mqttClient.InspectPackageAsync += eventArgs =>
|
||||
{
|
||||
packets.Add(eventArgs.Direction + ":" + Convert.ToBase64String(eventArgs.Buffer));
|
||||
return CompletedTask.Instance;
|
||||
|
||||
@@ -954,7 +954,7 @@ namespace MQTTnet.Tests.Server
|
||||
|
||||
var client = testEnvironment.CreateClient();
|
||||
|
||||
client.InspectPackage += e =>
|
||||
client.InspectPackageAsync += e =>
|
||||
{
|
||||
if (e.Buffer.Length > 0)
|
||||
{
|
||||
|
||||
@@ -8,14 +8,16 @@ namespace MQTTnet.Client
|
||||
public interface IMqttClient : IDisposable
|
||||
{
|
||||
event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync;
|
||||
|
||||
|
||||
event Func<MqttClientBeforeReleasePublishPacketEventArgs, Task> BeforeReleasePublishPacketAsync;
|
||||
|
||||
event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync;
|
||||
|
||||
event Func<MqttClientConnectingEventArgs, Task> ConnectingAsync;
|
||||
|
||||
event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync;
|
||||
|
||||
event Func<InspectMqttPacketEventArgs, Task> InspectPackage;
|
||||
event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync;
|
||||
|
||||
bool IsConnected { get; }
|
||||
|
||||
@@ -29,6 +31,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);
|
||||
|
||||
@@ -21,6 +21,7 @@ namespace MQTTnet.Client
|
||||
readonly IMqttClientAdapterFactory _adapterFactory;
|
||||
readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
|
||||
readonly AsyncEvent<MqttApplicationMessageReceivedEventArgs> _applicationMessageReceivedEvent = new AsyncEvent<MqttApplicationMessageReceivedEventArgs>();
|
||||
readonly AsyncEvent<MqttClientBeforeReleasePublishPacketEventArgs> _beforeReleasePublishPacketEvent = new AsyncEvent<MqttClientBeforeReleasePublishPacketEventArgs>();
|
||||
|
||||
readonly MqttClientPublishResultFactory _clientPublishResultFactory = new MqttClientPublishResultFactory();
|
||||
readonly MqttClientSubscribeResultFactory _clientSubscribeResultFactory = new MqttClientSubscribeResultFactory();
|
||||
@@ -63,6 +64,12 @@ namespace MQTTnet.Client
|
||||
remove => _applicationMessageReceivedEvent.RemoveHandler(value);
|
||||
}
|
||||
|
||||
public event Func<MqttClientBeforeReleasePublishPacketEventArgs, Task> BeforeReleasePublishPacketAsync
|
||||
{
|
||||
add => _beforeReleasePublishPacketEvent.AddHandler(value);
|
||||
remove => _beforeReleasePublishPacketEvent.RemoveHandler(value);
|
||||
}
|
||||
|
||||
public event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync
|
||||
{
|
||||
add => _connectedEvent.AddHandler(value);
|
||||
@@ -81,12 +88,12 @@ namespace MQTTnet.Client
|
||||
remove => _disconnectedEvent.RemoveHandler(value);
|
||||
}
|
||||
|
||||
public event Func<InspectMqttPacketEventArgs, Task> InspectPackage
|
||||
public event Func<InspectMqttPacketEventArgs, Task> InspectPackageAsync
|
||||
{
|
||||
add => _inspectPacketEvent.AddHandler(value);
|
||||
remove => _inspectPacketEvent.RemoveHandler(value);
|
||||
}
|
||||
|
||||
|
||||
public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected;
|
||||
|
||||
public MqttClientOptions Options { get; private set; }
|
||||
@@ -149,17 +156,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 +231,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 +266,51 @@ 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 eventArgs = new MqttClientBeforeReleasePublishPacketEventArgs(options.PacketIdentifier);
|
||||
await _beforeReleasePublishPacketEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
|
||||
|
||||
var pubRelPacket = _packetFactories.PubRel.Create(options);
|
||||
|
||||
MqttPubCompPacket pubCompPacket;
|
||||
if (cancellationToken.CanBeCanceled)
|
||||
{
|
||||
pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
using (var timeout = new CancellationTokenSource(Options.Timeout))
|
||||
{
|
||||
pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, timeout.Token).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
return new MqttClientReleasePublishPacketResult(pubCompPacket.PacketIdentifier, pubCompPacket.ReasonString, pubCompPacket.UserProperties);
|
||||
}
|
||||
|
||||
public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (data == null)
|
||||
@@ -411,7 +458,7 @@ namespace MQTTnet.Client
|
||||
|
||||
_adapter?.Dispose();
|
||||
_adapter = null;
|
||||
|
||||
|
||||
_packetDispatcher?.Dispose();
|
||||
_packetDispatcher = null;
|
||||
}
|
||||
@@ -448,7 +495,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 +551,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 +567,7 @@ namespace MQTTnet.Client
|
||||
|
||||
connectionStatus = curStatus;
|
||||
break;
|
||||
}
|
||||
}
|
||||
} while (true);
|
||||
}
|
||||
@@ -601,7 +652,7 @@ namespace MQTTnet.Client
|
||||
{
|
||||
// The packet is unknown. Probably due to a restart of the client.
|
||||
// So wen send this to the server to trigger a full resend of the message.
|
||||
var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
|
||||
var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttPubRelReasonCode.PacketIdentifierNotFound);
|
||||
return SendAsync(pubRelPacket, cancellationToken);
|
||||
}
|
||||
|
||||
@@ -628,7 +679,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 +688,16 @@ namespace MQTTnet.Client
|
||||
|
||||
var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
|
||||
if (!Options.AutoReleasePublishPackets)
|
||||
{
|
||||
return _clientPublishResultFactory.Create(pubRecPacket);
|
||||
}
|
||||
|
||||
var eventArgs = new MqttClientBeforeReleasePublishPacketEventArgs(pubRecPacket.PacketIdentifier);
|
||||
await _beforeReleasePublishPacketEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
|
||||
|
||||
var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttPubRelReasonCode.Success);
|
||||
var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
return _clientPublishResultFactory.Create(pubRecPacket, pubCompPacket);
|
||||
}
|
||||
|
||||
@@ -691,6 +748,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 +916,7 @@ namespace MQTTnet.Client
|
||||
{
|
||||
exception = aggregateException.GetBaseException();
|
||||
}
|
||||
|
||||
|
||||
if (exception is OperationCanceledException)
|
||||
{
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
// 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;
|
||||
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public sealed class MqttClientBeforeReleasePublishPacketEventArgs : EventArgs
|
||||
{
|
||||
public MqttClientBeforeReleasePublishPacketEventArgs(ushort packetIdentifier)
|
||||
{
|
||||
PacketIdentifier = packetIdentifier;
|
||||
}
|
||||
|
||||
public ushort PacketIdentifier { get; }
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,14 @@ namespace MQTTnet.Client
|
||||
{
|
||||
public sealed class MqttClientPublishResult
|
||||
{
|
||||
public MqttClientPublishResult(ushort? packetIdentifier, MqttClientPublishReasonCode reasonCode, string reasonString, IReadOnlyCollection<MqttUserProperty> userProperties)
|
||||
{
|
||||
PacketIdentifier = packetIdentifier;
|
||||
ReasonCode = reasonCode;
|
||||
ReasonString = reasonString;
|
||||
UserProperties = userProperties;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns if the overall status of the publish is a success. This can be the reason code _Success_ or
|
||||
/// _NoMatchingSubscribers_. _NoMatchingSubscribers_ usually indicates only that no other client is interested in the topic but overall transmit
|
||||
@@ -19,19 +27,19 @@ namespace MQTTnet.Client
|
||||
/// <summary>
|
||||
/// Gets the packet identifier which was used for this publish.
|
||||
/// </summary>
|
||||
public ushort? PacketIdentifier { get; set; }
|
||||
public ushort? PacketIdentifier { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the reason code.
|
||||
/// Hint: MQTT 5 feature only.
|
||||
/// </summary>
|
||||
public MqttClientPublishReasonCode ReasonCode { get; set; } = MqttClientPublishReasonCode.Success;
|
||||
public MqttClientPublishReasonCode ReasonCode { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the reason string.
|
||||
/// Hint: MQTT 5 feature only.
|
||||
/// </summary>
|
||||
public string ReasonString { get; set; }
|
||||
public string ReasonString { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the user properties.
|
||||
@@ -42,6 +50,6 @@ namespace MQTTnet.Client
|
||||
/// The feature is very similar to the HTTP header concept.
|
||||
/// Hint: MQTT 5 feature only.
|
||||
/// </summary>
|
||||
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; internal set; }
|
||||
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; }
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@
|
||||
// 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;
|
||||
using System.Collections.Generic;
|
||||
using MQTTnet.Packets;
|
||||
using MQTTnet.Protocol;
|
||||
@@ -10,8 +11,8 @@ namespace MQTTnet.Client
|
||||
{
|
||||
public sealed class MqttClientPublishResultFactory
|
||||
{
|
||||
static readonly MqttClientPublishResult EmptySuccessResult = new MqttClientPublishResult();
|
||||
static readonly IReadOnlyCollection<MqttUserProperty> EmptyUserProperties = new List<MqttUserProperty>();
|
||||
static readonly MqttClientPublishResult EmptySuccessResult = new MqttClientPublishResult(null, MqttClientPublishReasonCode.Success, null, EmptyUserProperties);
|
||||
|
||||
public MqttClientPublishResult Create(MqttPubAckPacket pubAckPacket)
|
||||
{
|
||||
@@ -21,59 +22,55 @@ namespace MQTTnet.Client
|
||||
return EmptySuccessResult;
|
||||
}
|
||||
|
||||
var result = new MqttClientPublishResult
|
||||
{
|
||||
// Both enums have the same values. So it can be easily converted.
|
||||
ReasonCode = (MqttClientPublishReasonCode)(int)pubAckPacket.ReasonCode,
|
||||
PacketIdentifier = pubAckPacket.PacketIdentifier,
|
||||
ReasonString = pubAckPacket.ReasonString,
|
||||
UserProperties = pubAckPacket.UserProperties ?? EmptyUserProperties
|
||||
};
|
||||
var result = new MqttClientPublishResult(
|
||||
pubAckPacket.PacketIdentifier,
|
||||
(MqttClientPublishReasonCode)(int)pubAckPacket.ReasonCode,
|
||||
pubAckPacket.ReasonString,
|
||||
pubAckPacket.UserProperties ?? EmptyUserProperties);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket)
|
||||
{
|
||||
if (pubRecPacket == null)
|
||||
{
|
||||
throw new ArgumentNullException(nameof(pubRecPacket));
|
||||
}
|
||||
|
||||
return new MqttClientPublishResult(
|
||||
pubRecPacket.PacketIdentifier,
|
||||
(MqttClientPublishReasonCode)(int)pubRecPacket.ReasonCode,
|
||||
pubRecPacket.ReasonString,
|
||||
pubRecPacket.UserProperties ?? EmptyUserProperties);
|
||||
}
|
||||
|
||||
public MqttClientPublishResult Create(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
|
||||
{
|
||||
if (pubRecPacket == null || pubCompPacket == null)
|
||||
{
|
||||
return new MqttClientPublishResult
|
||||
{
|
||||
ReasonCode = MqttClientPublishReasonCode.UnspecifiedError
|
||||
};
|
||||
return new MqttClientPublishResult(null, MqttClientPublishReasonCode.UnspecifiedError, null, EmptyUserProperties);
|
||||
}
|
||||
|
||||
MqttClientPublishResult result;
|
||||
|
||||
// The PUBCOMP is the last packet in QoS 2. So we use the results from that instead of PUBREC.
|
||||
if (pubCompPacket.ReasonCode == MqttPubCompReasonCode.PacketIdentifierNotFound)
|
||||
{
|
||||
result = new MqttClientPublishResult
|
||||
{
|
||||
PacketIdentifier = pubCompPacket.PacketIdentifier,
|
||||
ReasonCode = MqttClientPublishReasonCode.UnspecifiedError,
|
||||
ReasonString = pubCompPacket.ReasonString,
|
||||
UserProperties = pubCompPacket.UserProperties ?? EmptyUserProperties
|
||||
};
|
||||
|
||||
return result;
|
||||
return new MqttClientPublishResult(
|
||||
pubCompPacket.PacketIdentifier,
|
||||
MqttClientPublishReasonCode.UnspecifiedError,
|
||||
pubCompPacket.ReasonString,
|
||||
pubCompPacket.UserProperties ?? EmptyUserProperties);
|
||||
}
|
||||
|
||||
result = new MqttClientPublishResult
|
||||
{
|
||||
PacketIdentifier = pubCompPacket.PacketIdentifier,
|
||||
ReasonCode = MqttClientPublishReasonCode.Success,
|
||||
ReasonString = pubCompPacket.ReasonString,
|
||||
UserProperties = pubCompPacket.UserProperties ?? EmptyUserProperties
|
||||
};
|
||||
var reasonCode = MqttClientPublishReasonCode.Success;
|
||||
|
||||
if (pubRecPacket.ReasonCode != MqttPubRecReasonCode.Success)
|
||||
{
|
||||
// Both enums share the same values.
|
||||
result.ReasonCode = (MqttClientPublishReasonCode)pubRecPacket.ReasonCode;
|
||||
reasonCode = (MqttClientPublishReasonCode)pubRecPacket.ReasonCode;
|
||||
}
|
||||
|
||||
return result;
|
||||
return new MqttClientPublishResult(pubCompPacket.PacketIdentifier, reasonCode, pubCompPacket.ReasonString, pubCompPacket.UserProperties ?? EmptyUserProperties);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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; }
|
||||
}
|
||||
}
|
||||
@@ -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; }
|
||||
}
|
||||
}
|
||||
@@ -11,7 +11,7 @@ namespace MQTTnet.Formatter
|
||||
{
|
||||
public sealed class MqttPubRelPacketFactory
|
||||
{
|
||||
public MqttPubRelPacket Create(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
|
||||
public MqttPubRelPacket Create(MqttPubRecPacket pubRecPacket, MqttPubRelReasonCode reasonCode)
|
||||
{
|
||||
if (pubRecPacket == null)
|
||||
{
|
||||
@@ -21,7 +21,23 @@ namespace MQTTnet.Formatter
|
||||
return new MqttPubRelPacket
|
||||
{
|
||||
PacketIdentifier = pubRecPacket.PacketIdentifier,
|
||||
ReasonCode = (MqttPubRelReasonCode)(int)reasonCode
|
||||
ReasonCode = 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
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@@ -273,7 +273,7 @@ namespace MQTTnet.Server
|
||||
{
|
||||
// Do not fire the event _ClientAcknowledgedPublishPacket_ here because the QoS 2 process is only finished
|
||||
// properly when the client has sent the PUBCOMP packet.
|
||||
var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
|
||||
var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttPubRelReasonCode.Success);
|
||||
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
|
||||
|
||||
return CompletedTask.Instance;
|
||||
|
||||
Reference in New Issue
Block a user