YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-15 00:15:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into Managed-Client-V2

Browse Source
This commit is contained in:
Christian
2023-01-19 20:50:34 +01:00
parent 7bb29d3c92 2aa4219838
commit f8d614e71f
20 changed files with 246 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ReleaseNotes.md
+1
View File
@@ -5,3 +5,4 @@
* [Server] Exposed MQTT v5 sent properties from the affected client in _ClientDisconnectedAsync_ event.
* [Server] Fixed wrong client ID passed to _InterceptingUnsubscriptionEventArgs_ (#1631, thanks to @ghord).
* [Server] Exposed socket settings for TCP keep alive in TCP options (#1544).
* [Server] Exposed server session items at server level and allow custom session items for injected application messages (#1638).
Source/MQTTnet/Adapter/MqttChannelAdapter.cs
+18 -11
View File
@@ -33,8 +33,7 @@ namespace MQTTnet.Adapter
readonly AsyncLock _syncRoot = new AsyncLock();
long _bytesReceived;
long _bytesSent;
Statistics _statistics; // mutable struct, don't make readonly!
public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, MqttPacketInspector packetInspector, IMqttNetLogger logger)
{
@@ -51,9 +50,9 @@ namespace MQTTnet.Adapter
_logger = logger.WithSource(nameof(MqttChannelAdapter));
}
public long BytesReceived => Interlocked.Read(ref _bytesReceived);
public long BytesReceived => Volatile.Read(ref _statistics._bytesReceived);
public long BytesSent => Interlocked.Read(ref _bytesSent);
public long BytesSent => Volatile.Read(ref _statistics._bytesSent);
public X509Certificate2 ClientCertificate => _channel.ClientCertificate;
@@ -147,7 +146,7 @@ namespace MQTTnet.Adapter
_packetInspector?.EndReceivePacket();
Interlocked.Add(ref _bytesSent, receivedPacket.TotalLength);
Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength);
if (PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.Unknown)
{
@@ -181,11 +180,7 @@ namespace MQTTnet.Adapter
return null;
}
public void ResetStatistics()
{
Interlocked.Exchange(ref _bytesReceived, 0L);
Interlocked.Exchange(ref _bytesSent, 0L);
}
public void ResetStatistics() => _statistics.Reset();
public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
{
@@ -216,7 +211,7 @@ namespace MQTTnet.Adapter
await _channel.WriteAsync(packetBuffer.Packet, true, cancellationToken).ConfigureAwait(false);
}
Interlocked.Add(ref _bytesReceived, packetBuffer.Length);
Interlocked.Add(ref _statistics._bytesReceived, packetBuffer.Length);
}
catch (Exception exception)
{
@@ -459,5 +454,17 @@ namespace MQTTnet.Adapter
throw new MqttCommunicationException(exception);
}
private struct Statistics
{
public long _bytesReceived;
public long _bytesSent;
public void Reset()
{
Volatile.Write(ref _bytesReceived, 0);
Volatile.Write(ref _bytesSent, 0);
}
}
}
}
Source/MQTTnet/Adapter/MqttPacketInspector.cs
+1 -6
View File
@@ -82,12 +82,7 @@ namespace MQTTnet.Adapter
{
try
{
var eventArgs = new InspectMqttPacketEventArgs
{
Buffer = buffer,
Direction = direction
};
var eventArgs = new InspectMqttPacketEventArgs(direction, buffer);
_asyncEvent.InvokeAsync(eventArgs).GetAwaiter().GetResult();
}
catch (Exception exception)
Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs
+17 -8
View File
@@ -9,9 +9,18 @@ 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
/// _NoMatchingSubscribers_. _NoMatchingSubscribers_ usually indicates only that no other client is interested in the
/// topic but overall transmit
/// to the server etc. was a success.
/// </summary>
public bool IsSuccess => ReasonCode == MqttClientPublishReasonCode.Success || ReasonCode == MqttClientPublishReasonCode.NoMatchingSubscribers;
@@ -19,19 +28,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.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public MqttClientPublishReasonCode ReasonCode { get; set; } = MqttClientPublishReasonCode.Success;
public MqttClientPublishReasonCode ReasonCode { get; } = MqttClientPublishReasonCode.Success;
/// <summary>
/// Gets or sets the reason string.
/// Hint: MQTT 5 feature only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public string ReasonString { get; set; }
public string ReasonString { get; }
/// <summary>
/// Gets or sets the user properties.
@@ -40,8 +49,8 @@ namespace MQTTnet.Client
/// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add
/// metadata to MQTT messages and pass information between publisher, broker, and subscriber.
/// The feature is very similar to the HTTP header concept.
/// Hint: MQTT 5 feature only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; internal set; }
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; }
}
}
Source/MQTTnet/Client/Publishing/MqttClientPublishResultFactory.cs
+17 -34
View File
@@ -10,25 +10,23 @@ namespace MQTTnet.Client
{
public sealed class MqttClientPublishResultFactory
{
static readonly MqttClientPublishResult EmptySuccessResult = new MqttClientPublishResult();
static readonly IReadOnlyCollection<MqttUserProperty> EmptyUserProperties = new List<MqttUserProperty>();
static readonly MqttClientPublishResult AtMostOnceSuccessResult = new MqttClientPublishResult(null, MqttClientPublishReasonCode.Success, null, EmptyUserProperties);
public MqttClientPublishResult Create(MqttPubAckPacket pubAckPacket)
{
// QoS 0 has no response. So we treat it as a success always.
if (pubAckPacket == null)
{
return EmptySuccessResult;
return AtMostOnceSuccessResult;
}
var result = new MqttClientPublishResult
{
var result = new MqttClientPublishResult(
pubAckPacket.PacketIdentifier,
// 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
};
(MqttClientPublishReasonCode)(int)pubAckPacket.ReasonCode,
pubAckPacket.ReasonString,
pubAckPacket.UserProperties ?? EmptyUserProperties);
return result;
}
@@ -37,43 +35,28 @@ namespace MQTTnet.Client
{
if (pubRecPacket == null || pubCompPacket == null)
{
return new MqttClientPublishResult
{
ReasonCode = MqttClientPublishReasonCode.UnspecifiedError
};
var packetIdentifier = pubRecPacket?.PacketIdentifier ?? pubCompPacket?.PacketIdentifier;
return new MqttClientPublishResult(packetIdentifier, MqttClientPublishReasonCode.ImplementationSpecificError, 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, null, pubCompPacket.UserProperties ?? EmptyUserProperties);
}
}
}
Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResult.cs
+21 -8
View File
@@ -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;
@@ -9,23 +10,35 @@ namespace MQTTnet.Client
{
public sealed class MqttClientSubscribeResult
{
public IReadOnlyCollection<MqttClientSubscribeResultItem> Items { get; internal set; }
public MqttClientSubscribeResult(ushort packetIdentifier, IReadOnlyCollection<MqttClientSubscribeResultItem> items, string reasonString, IReadOnlyCollection<MqttUserProperty> userProperties)
{
PacketIdentifier = packetIdentifier;
Items = items ?? throw new ArgumentNullException(nameof(items));
ReasonString = reasonString;
UserProperties = userProperties ?? throw new ArgumentNullException(nameof(userProperties));
}
/// <summary>
/// Gets the packet identifier which was used.
/// Gets the result for every topic filter item.
/// </summary>
public ushort PacketIdentifier { get; internal set; }
public IReadOnlyCollection<MqttClientSubscribeResultItem> Items { get; }
/// <summary>
/// Gets the reason string.
/// MQTTv5 only.
/// Gets the user properties which were part of the SUBACK packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public string ReasonString { get; internal set; }
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; }
/// <summary>
/// Gets the reason string.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public string ReasonString { get; }
/// <summary>
/// Gets the user properties which were part of the SUBACK packet.
/// MQTTv5 only.
/// </summary>
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; internal set; }
public ushort PacketIdentifier { get; }
}
}
Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs
+16 -23
View File
@@ -16,42 +16,35 @@ namespace MQTTnet.Client
public MqttClientSubscribeResult Create(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket)
{
if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
if (subAckPacket == null) throw new ArgumentNullException(nameof(subAckPacket));
if (subscribePacket == null)
{
throw new ArgumentNullException(nameof(subscribePacket));
}
if (subAckPacket == null)
{
throw new ArgumentNullException(nameof(subAckPacket));
}
// MQTTv5.0.0 handling.
if (subAckPacket.ReasonCodes.Any() && subAckPacket.ReasonCodes.Count != subscribePacket.TopicFilters.Count)
{
throw new MqttProtocolViolationException(
"The reason codes are not matching the topic filters [MQTT-3.9.3-1].");
throw new MqttProtocolViolationException("The reason codes are not matching the topic filters [MQTT-3.9.3-1].");
}
var items = new List<MqttClientSubscribeResultItem>();
for (var i = 0; i < subscribePacket.TopicFilters.Count; i++)
{
items.Add(CreateSubscribeResultItem(i, subscribePacket, subAckPacket));
}
var result = new MqttClientSubscribeResult
{
PacketIdentifier = subAckPacket.PacketIdentifier,
ReasonString = subAckPacket.ReasonString,
UserProperties = subAckPacket.UserProperties ?? EmptyUserProperties,
Items = items
};
return result;
return new MqttClientSubscribeResult(subAckPacket.PacketIdentifier, items, subAckPacket.ReasonString, subAckPacket.UserProperties ?? EmptyUserProperties);
}
static MqttClientSubscribeResultItem CreateSubscribeResultItem(int index, MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket)
{
var resultCode = (MqttClientSubscribeResultCode) subAckPacket.ReasonCodes[index];
return new MqttClientSubscribeResultItem
{
TopicFilter = subscribePacket.TopicFilters[index],
ResultCode = resultCode,
};
var resultCode = (MqttClientSubscribeResultCode)subAckPacket.ReasonCodes[index];
return new MqttClientSubscribeResultItem(subscribePacket.TopicFilters[index], resultCode);
}
}
}
Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultItem.cs
+15 -8
View File
@@ -2,22 +2,29 @@
// 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 MQTTnet.Packets;
namespace MQTTnet.Client
{
public sealed class MqttClientSubscribeResultItem
{
/// <summary>
/// Gets or sets the result code.
/// Hint: MQTT 5 feature only.
/// </summary>
public MqttClientSubscribeResultCode ResultCode { get; internal set; }
public MqttClientSubscribeResultItem(MqttTopicFilter topicFilter, MqttClientSubscribeResultCode resultCode)
{
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
ResultCode = resultCode;
}
/// <summary>
/// Gets or sets the topic filter.
/// The topic filter can contain topics and wildcards.
/// Gets or sets the topic filter.
/// The topic filter can contain topics and wildcards.
/// </summary>
public MqttTopicFilter TopicFilter { get; internal set; }
public MqttTopicFilter TopicFilter { get; }
/// <summary>
/// Gets or sets the result code.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public MqttClientSubscribeResultCode ResultCode { get; }
}
}
Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs
+22 -6
View File
@@ -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;
@@ -9,23 +10,38 @@ namespace MQTTnet.Client
{
public sealed class MqttClientUnsubscribeResult
{
public IReadOnlyCollection<MqttClientUnsubscribeResultItem> Items { get; internal set; }
public MqttClientUnsubscribeResult(
ushort packetIdentifier,
IReadOnlyCollection<MqttClientUnsubscribeResultItem> items,
string reasonString,
IReadOnlyCollection<MqttUserProperty> userProperties)
{
PacketIdentifier = packetIdentifier;
Items = items ?? throw new ArgumentNullException(nameof(items));
ReasonString = reasonString;
UserProperties = userProperties ?? throw new ArgumentNullException(nameof(userProperties));
}
/// <summary>
/// Gets the result for every topic filter item.
/// </summary>
public IReadOnlyCollection<MqttClientUnsubscribeResultItem> Items { get; }
/// <summary>
/// Gets the packet identifier which was used.
/// </summary>
public ushort PacketIdentifier { get; internal set; }
public ushort PacketIdentifier { get; }
/// <summary>
/// Gets the reason string.
/// MQTTv5 only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public string ReasonString { get; internal set; }
public string ReasonString { get; }
/// <summary>
/// Gets the user properties which were part of the UNSUBACK packet.
/// MQTTv5 only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; internal set; }
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; }
}
}
Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultFactory.cs
+18 -24
View File
@@ -12,17 +12,23 @@ namespace MQTTnet.Client
public sealed class MqttClientUnsubscribeResultFactory
{
static readonly IReadOnlyCollection<MqttUserProperty> EmptyUserProperties = new List<MqttUserProperty>();
public MqttClientUnsubscribeResult Create(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket)
{
if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
if (unsubAckPacket == null) throw new ArgumentNullException(nameof(unsubAckPacket));
if (unsubscribePacket == null)
{
throw new ArgumentNullException(nameof(unsubscribePacket));
}
if (unsubAckPacket == null)
{
throw new ArgumentNullException(nameof(unsubAckPacket));
}
// MQTTv3.1.1 has no reason code at all!
if (unsubAckPacket.ReasonCodes != null && unsubAckPacket.ReasonCodes.Count != unsubscribePacket.TopicFilters.Count)
{
throw new MqttProtocolViolationException(
"The return codes are not matching the topic filters [MQTT-3.9.3-1].");
throw new MqttProtocolViolationException("The return codes are not matching the topic filters [MQTT-3.9.3-1].");
}
var items = new List<MqttClientUnsubscribeResultItem>();
@@ -30,33 +36,21 @@ namespace MQTTnet.Client
{
items.Add(CreateUnsubscribeResultItem(i, unsubscribePacket, unsubAckPacket));
}
var result = new MqttClientUnsubscribeResult
{
PacketIdentifier = unsubAckPacket.PacketIdentifier,
ReasonString = unsubAckPacket.ReasonString,
UserProperties = unsubAckPacket.UserProperties ?? EmptyUserProperties,
Items = items
};
return result;
return new MqttClientUnsubscribeResult(unsubscribePacket.PacketIdentifier, items, unsubAckPacket.ReasonString, unsubAckPacket.UserProperties ?? EmptyUserProperties);
}
static MqttClientUnsubscribeResultItem CreateUnsubscribeResultItem(int index, MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket)
{
var resultCode = MqttClientUnsubscribeResultCode.Success;
if (unsubAckPacket.ReasonCodes != null)
{
// MQTTv3.1.1 has no reason code and no return code!.
resultCode = (MqttClientUnsubscribeResultCode) unsubAckPacket.ReasonCodes[index];
resultCode = (MqttClientUnsubscribeResultCode)unsubAckPacket.ReasonCodes[index];
}
return new MqttClientUnsubscribeResultItem
{
TopicFilter = unsubscribePacket.TopicFilters[index],
ResultCode = resultCode
};
return new MqttClientUnsubscribeResultItem(unsubscribePacket.TopicFilters[index], resultCode);
}
}
}
Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResultItem.cs
+11 -3
View File
@@ -2,20 +2,28 @@
// 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 MqttClientUnsubscribeResultItem
{
public MqttClientUnsubscribeResultItem(string topicFilter, MqttClientUnsubscribeResultCode resultCode)
{
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
ResultCode = resultCode;
}
/// <summary>
/// Gets or sets the result code.
/// Hint: MQTT 5 feature only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public MqttClientUnsubscribeResultCode ResultCode { get; internal set; }
public MqttClientUnsubscribeResultCode ResultCode { get; }
/// <summary>
/// Gets or sets the topic filter.
/// The topic filter can contain topics and wildcards.
/// </summary>
public string TopicFilter { get; internal set; }
public string TopicFilter { get; }
}
}
Source/MQTTnet/Diagnostics/PacketInspection/InspectMqttPacketEventArgs.cs
+8 -2
View File
@@ -8,8 +8,14 @@ namespace MQTTnet.Diagnostics
{
public sealed class InspectMqttPacketEventArgs : EventArgs
{
public MqttPacketFlowDirection Direction { get; internal set; }
public InspectMqttPacketEventArgs(MqttPacketFlowDirection direction, byte[] buffer)
{
Direction = direction;
Buffer = buffer ?? throw new ArgumentNullException(nameof(buffer));
}
public byte[] Buffer { get; set; }
public byte[] Buffer { get; }
public MqttPacketFlowDirection Direction { get; }
}
}
Source/MQTTnet/Internal/AsyncQueue.cs
+14
View File
@@ -22,7 +22,11 @@ namespace MQTTnet.Internal
public void Clear()
{
#if NETCOREAPP3_1_OR_GREATER
_queue.Clear();
#else
Interlocked.Exchange(ref _queue, new ConcurrentQueue<TItem>());
#endif
}
public void Dispose()
@@ -32,6 +36,16 @@ namespace MQTTnet.Internal
_signal.Dispose();
_isDisposed = true;
#if !NETSTANDARD1_3
if (typeof(IDisposable).IsAssignableFrom(typeof(TItem)))
{
while (_queue.TryDequeue(out TItem item))
{
(item as IDisposable).Dispose();
}
}
#endif
}
}
Source/MQTTnet/MQTTnet.csproj
+1 -1
View File
@@ -89,4 +89,4 @@
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.10" />
</ItemGroup>
</Project>
</Project>
Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs
+1 -1
View File
@@ -10,7 +10,7 @@ namespace MQTTnet.Server
{
public sealed class PreparingSessionEventArgs : EventArgs
{
public string Id { get; internal set; }
public string Id { get; set; }
// TODO: Allow adding of packets to the queue etc.
Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs
+13 -13
View File
@@ -9,7 +9,6 @@ using System.Security.Cryptography.X509Certificates;
using System.Text;
using MQTTnet.Adapter;
using MQTTnet.Formatter;
using MQTTnet.Implementations;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -20,10 +19,11 @@ namespace MQTTnet.Server
{
readonly MqttConnectPacket _connectPacket;
public ValidatingConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter)
public ValidatingConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter, IDictionary sessionItems)
{
_connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
ChannelAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter));
SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
}
/// <summary>
@@ -34,13 +34,13 @@ namespace MQTTnet.Server
/// <summary>
/// Gets or sets the authentication data.
/// Hint: MQTT 5 feature only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public byte[] AuthenticationData => _connectPacket.AuthenticationData;
/// <summary>
/// Gets or sets the authentication method.
/// Hint: MQTT 5 feature only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public string AuthenticationMethod => _connectPacket.AuthenticationMethod;
@@ -99,7 +99,7 @@ namespace MQTTnet.Server
/// Gets or sets the reason code. When a MQTTv3 client connects the enum value must be one which is
/// also supported in MQTTv3. Otherwise the connection attempt will fail because not all codes can be
/// converted properly.
/// MQTTv5 only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public MqttConnectReasonCode ReasonCode { get; set; } = MqttConnectReasonCode.Success;
@@ -114,19 +114,19 @@ namespace MQTTnet.Server
/// <summary>
/// Gets the request problem information.
/// Hint: MQTT 5 feature only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public bool RequestProblemInformation => _connectPacket.RequestProblemInformation;
/// <summary>
/// Gets the request response information.
/// Hint: MQTT 5 feature only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public bool RequestResponseInformation => _connectPacket.RequestResponseInformation;
/// <summary>
/// Gets or sets the response authentication data.
/// MQTTv5 only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public byte[] ResponseAuthenticationData { get; set; }
@@ -137,14 +137,14 @@ namespace MQTTnet.Server
/// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add
/// metadata to MQTT messages and pass information between publisher, broker, and subscriber.
/// The feature is very similar to the HTTP header concept.
/// Hint: MQTT 5 feature only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public List<MqttUserProperty> ResponseUserProperties { get; set; }
/// <summary>
/// Gets or sets the server reference. This can be used together with i.e. "Server Moved" to send
/// a different server address to the client.
/// MQTTv5 only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public string ServerReference { get; set; }
@@ -158,7 +158,7 @@ namespace MQTTnet.Server
/// <summary>
/// Gets or sets a key/value collection that can be used to share data within the scope of this session.
/// </summary>
public IDictionary SessionItems { get; internal set; }
public IDictionary SessionItems { get; }
/// <summary>
/// Gets or sets the topic alias maximum.
@@ -169,7 +169,7 @@ namespace MQTTnet.Server
[Obsolete("This property name has a typo. Use 'UserName' instead. This one will be removed soon.")]
public string Username => _connectPacket.Username;
public string UserName => _connectPacket.Username;
/// <summary>
@@ -179,7 +179,7 @@ namespace MQTTnet.Server
/// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add
/// metadata to MQTT messages and pass information between publisher, broker, and subscriber.
/// The feature is very similar to the HTTP header concept.
/// Hint: MQTT 5 feature only.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public List<MqttUserProperty> UserProperties => _connectPacket.UserProperties;
Source/MQTTnet/Server/InjectedMqttApplicationMessage.cs
+10 -2
View File
@@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information.
using System;
using System.Collections;
namespace MQTTnet.Server
{
@@ -13,8 +14,15 @@ namespace MQTTnet.Server
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
}
public string SenderClientId { get; set; } = string.Empty;
public MqttApplicationMessage ApplicationMessage { get; }
public MqttApplicationMessage ApplicationMessage { get; set; }
/// <summary>
/// Gets or sets the session items which should be used for all event handlers which are involved in message
/// processing.
/// If _null_ is specified the singleton session items from the server are used instead.
/// </summary>
public IDictionary CustomSessionItems { get; set; }
public string SenderClientId { get; set; } = string.Empty;
}
}
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+3 -5
View File
@@ -669,11 +669,9 @@ namespace MQTTnet.Server
async Task<ValidatingConnectionEventArgs> ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
{
var eventArgs = new ValidatingConnectionEventArgs(connectPacket, channelAdapter)
{
SessionItems = new ConcurrentDictionary<object, object>()
};
// TODO: Load session items from persisted sessions in the future.
var sessionItems = new ConcurrentDictionary<object, object>();
var eventArgs = new ValidatingConnectionEventArgs(connectPacket, channelAdapter, sessionItems);
await _eventContainer.ValidatingConnectionEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
// Check the client ID and set a random one if supported.
Source/MQTTnet/Server/Internal/MqttClientStatistics.cs
+30 -21
View File
@@ -10,14 +10,14 @@ namespace MQTTnet.Server
{
public sealed class MqttClientStatistics
{
long _receivedApplicationMessagesCount;
Statistics _statistics = new Statistics // mutable struct, don't make readonly!
{
// Start with 1 because the CONNACK packet is not counted here.
_receivedPacketsCount = 1,
// Start with 1 because the CONNACK packet is not counted here.
long _receivedPacketsCount = 1;
long _sentApplicationMessagesCount;
// Start with 1 because the CONNECT packet is not counted here.
long _sentPacketsCount = 1;
// Start with 1 because the CONNECT packet is not counted here.
_sentPacketsCount = 1
};
public MqttClientStatistics()
{
@@ -30,7 +30,6 @@ namespace MQTTnet.Server
}
public DateTime ConnectedTimestamp { get; }
public DateTime LastNonKeepAlivePacketReceivedTimestamp { get; private set; }
/// <summary>
@@ -43,13 +42,13 @@ namespace MQTTnet.Server
/// </summary>
public DateTime LastPacketSentTimestamp { get; private set; }
public long ReceivedApplicationMessagesCount => Interlocked.Read(ref _receivedApplicationMessagesCount);
public long SentApplicationMessagesCount => Volatile.Read(ref _statistics._sentApplicationMessagesCount);
public long ReceivedPacketsCount => Interlocked.Read(ref _receivedPacketsCount);
public long ReceivedApplicationMessagesCount => Volatile.Read(ref _statistics._receivedApplicationMessagesCount);
public long SentApplicationMessagesCount => Interlocked.Read(ref _sentApplicationMessagesCount);
public long SentPacketsCount => Volatile.Read(ref _statistics._sentPacketsCount);
public long SentPacketsCount => Interlocked.Read(ref _sentPacketsCount);
public long ReceivedPacketsCount => Volatile.Read(ref _statistics._receivedPacketsCount);
public void HandleReceivedPacket(MqttPacket packet)
{
@@ -61,11 +60,11 @@ namespace MQTTnet.Server
// This class is tracking all values from Clients perspective!
LastPacketSentTimestamp = DateTime.UtcNow;
Interlocked.Increment(ref _sentPacketsCount);
Interlocked.Increment(ref _statistics._sentPacketsCount);
if (packet is MqttPublishPacket)
{
Interlocked.Increment(ref _sentApplicationMessagesCount);
Interlocked.Increment(ref _statistics._sentApplicationMessagesCount);
}
if (!(packet is MqttPingReqPacket || packet is MqttPingRespPacket))
@@ -73,6 +72,7 @@ namespace MQTTnet.Server
LastNonKeepAlivePacketReceivedTimestamp = LastPacketReceivedTimestamp;
}
}
public void ResetStatistics() => _statistics.Reset();
public void HandleSentPacket(MqttPacket packet)
{
@@ -84,20 +84,29 @@ namespace MQTTnet.Server
// This class is tracking all values from Clients perspective!
LastPacketReceivedTimestamp = DateTime.UtcNow;
Interlocked.Increment(ref _receivedPacketsCount);
Interlocked.Increment(ref _statistics._receivedPacketsCount);
if (packet is MqttPublishPacket)
{
Interlocked.Increment(ref _receivedApplicationMessagesCount);
Interlocked.Increment(ref _statistics._receivedApplicationMessagesCount);
}
}
public void ResetStatistics()
private struct Statistics
{
Interlocked.Exchange(ref _sentApplicationMessagesCount, 0);
Interlocked.Exchange(ref _receivedApplicationMessagesCount, 0);
Interlocked.Exchange(ref _sentPacketsCount, 0);
Interlocked.Exchange(ref _receivedPacketsCount, 0);
public long _receivedPacketsCount;
public long _sentPacketsCount;
public long _receivedApplicationMessagesCount;
public long _sentApplicationMessagesCount;
public void Reset()
{
Volatile.Write(ref _receivedPacketsCount, 0);
Volatile.Write(ref _sentPacketsCount, 0);
Volatile.Write(ref _receivedApplicationMessagesCount, 0);
Volatile.Write(ref _sentApplicationMessagesCount, 0);
}
}
}
}
Source/MQTTnet/Server/MqttServer.cs
+9 -3
View File
@@ -28,8 +28,6 @@ namespace MQTTnet.Server
readonly MqttRetainedMessagesManager _retainedMessagesManager;
readonly IMqttNetLogger _rootLogger;
readonly IDictionary _sessionItems = new ConcurrentDictionary<object, object>();
CancellationTokenSource _cancellationTokenSource;
public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
@@ -167,6 +165,12 @@ namespace MQTTnet.Server
public bool IsStarted => _cancellationTokenSource != null;
/// <summary>
/// Gives access to the session items which belong to this server. This session items are passed
/// to several events instead of the client session items if the event is caused by the server instead of a client.
/// </summary>
public IDictionary ServerSessionItems { get; } = new ConcurrentDictionary<object, object>();
public Task DeleteRetainedMessagesAsync()
{
ThrowIfNotStarted();
@@ -228,9 +232,11 @@ namespace MQTTnet.Server
throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported.");
}
var sessionItems = injectedApplicationMessage.CustomSessionItems ?? ServerSessionItems;
return _clientSessionsManager.DispatchApplicationMessage(
injectedApplicationMessage.SenderClientId,
_sessionItems,
sessionItems,
injectedApplicationMessage.ApplicationMessage,
cancellationToken);
}