Compare commits

...
Author SHA1 Message Date
Christian 05dcb250e7 Add README to nugets 2024-04-20 14:23:15 +02:00
christian 7f5c437c46 Fix unit tests 2024-04-11 22:26:39 +02:00
christian ae5e038548 Merge remote-tracking branch 'origin/master' 2024-04-09 21:34:31 +02:00
christian 67e0fb8b0b Fix build errors by no longer using not supported language features for certain TFMs. 2024-04-09 21:34:15 +02:00
Marc PiulachsandChristian b763542329 Move "Client is not connected" error to it's own exception (#1954)
* Move "Client is not connected" error to it's own exception

* Update MqttClientNotConnectedException.cs

* Update release notes

---------

Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
2024-04-09 21:26:59 +02:00
christian d061da6077 Update release notes 2024-04-09 21:24:39 +02:00
Andriy Tolstoy 82a235aef4 remove select method (#1959) 2024-04-09 21:18:09 +02:00
James Thompson fd55982d98 #1957 Invert the logic to improve maintainance (#1958)
#1957 Invert the logic to improve maintaince
2024-04-01 22:13:48 +02:00
aabf180b23 1862 Add subscribe and unsubscribe results event for Managed Mqtt Client. (#1894)
* See the issue #1862

Adding an event when ManagedMqttClient actually subscribe/unsubscribe to topics.
This event has to be used in parallel with connected event to know the exact state.

* Rename SubscriptionsResultAsync to SubscriptionsChangedAsync.

* Rename internal field for event

* Update ReleaseNotes.md

---------

Co-authored-by: Pierre Dufrene <pierre.dufrene@se.com>
Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
2024-03-02 09:58:42 +01:00
Christian 1245d8299a Merge branch 'master' of https://github.com/dotnet/MQTTnet 2024-03-02 08:45:03 +01:00
yztandChristian 1ffffaff47 Fix minor bug in MqttV5PacketEncoder.cs (#1939)
* Fix minor bug in MqttV5PacketEncoder.cs

* Update ReleaseNotes.md

---------

Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
2024-03-02 09:44:14 +01:00
Christian c204f6d8b2 Make packet inspector fully async (#1941)
* Make packet inspector methods async.

* Update release notes
2024-03-02 09:28:56 +01:00
Christian b775fb256b Update version 2024-03-02 08:44:57 +01:00
20 changed files with 270 additions and 79 deletions
+4 -5
View File
@@ -1,5 +1,4 @@
* [Client] Exposed the _EndPoint_ type to support other endpoint types (like Unix Domain Sockets) in client options (#1919).
* [ManagedClient] Added support for intercepting (and cancelling) publish messages (#1915, thanks to @hannasm).
* [Server] Added new events for delivered and dropped messages (#1866, thanks to @kallayj).
* [Server] The server will no longer treat a client which is receiving a large payload as alive. The packet must be received completely within the keep alive boundaries (BREAKING CHANGE!, #1883).
* [Server] Fixed "service not registered" exception in ASP.NET integration (#1889).
* [Core] Optimized packet serialization of PUBACK and PUBREC packets for protocol version 5.0.0 (#1939, thanks to @Y-Sindo).
* [Core] The package inspector is now fully async (#1941).
* [Client] Added a dedicated exception when the client is not connected (#1954, thanks to @marcpiulachs).
* [ManagedClient] Added a new event (SubscriptionsChangedAsync) which is fired when a subscription or unsubscription was made (#1894, thanks to @pdufrene).
+1 -1
View File
@@ -3,7 +3,7 @@ name: CI
on: [push, pull_request]
env:
VERSION: "4.3.3.${{github.run_number}}"
VERSION: "4.3.4.${{github.run_number}}"
jobs:
build:
@@ -0,0 +1,87 @@
// 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.
// ReSharper disable UnusedType.Global
// ReSharper disable UnusedMember.Global
// ReSharper disable InconsistentNaming
using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
namespace MQTTnet.Samples.ManagedClient;
public sealed class Managed_Client_Subscribe_Samples
{
public static async Task Connect_Client()
{
/*
* This sample creates a simple managed MQTT client and connects to a public broker, subscribe to a topic and verifies subscription result.
*
* The managed client extends the existing _MqttClient_. It adds the following features.
* - Reconnecting when connection is lost.
* - Storing pending messages in an internal queue so that an enqueue is possible while the client remains not connected.
*/
var mqttFactory = new MqttFactory();
var subscribed = false;
using (var managedMqttClient = mqttFactory.CreateManagedMqttClient())
{
var mqttClientOptions = new MqttClientOptionsBuilder()
.WithTcpServer("broker.hivemq.com")
.Build();
var managedMqttClientOptions = new ManagedMqttClientOptionsBuilder()
.WithClientOptions(mqttClientOptions)
.Build();
await managedMqttClient.StartAsync(managedMqttClientOptions);
// The application message is not sent. It is stored in an internal queue and
// will be sent when the client is connected.
await managedMqttClient.EnqueueAsync("Topic", "Payload");
Console.WriteLine("The managed MQTT client is connected.");
// Wait until the queue is fully processed.
SpinWait.SpinUntil(() => managedMqttClient.PendingApplicationMessagesCount == 0, 10000);
Console.WriteLine($"Pending messages = {managedMqttClient.PendingApplicationMessagesCount}");
managedMqttClient.SubscriptionsChangedAsync += args => SubscriptionsResultAsync(args, subscribed);
await managedMqttClient.SubscribeAsync("Topic").ConfigureAwait(false);
SpinWait.SpinUntil(() => subscribed, 1000);
Console.WriteLine("Subscription properly done");
}
}
private static Task SubscriptionsResultAsync(SubscriptionsChangedEventArgs arg, bool subscribed)
{
foreach (var mqttClientSubscribeResult in arg.SubscribeResult)
{
Console.WriteLine($"Subscription reason {mqttClientSubscribeResult.ReasonString}");
foreach (var item in mqttClientSubscribeResult.Items)
{
Console.WriteLine($"For topic filter {item.TopicFilter}, result code: {item.ResultCode}");
if (item.TopicFilter.Topic == "Topic" && item.ResultCode == MqttClientSubscribeResultCode.GrantedQoS0 && !subscribed)
{
subscribed = true;
}
}
}
foreach (var mqttClientUnsubscribeResult in arg.UnsubscribeResult)
{
Console.WriteLine($"Unsubscription reason {mqttClientUnsubscribeResult.ReasonString}");
foreach (var item in mqttClientUnsubscribeResult.Items)
{
Console.WriteLine($"For topic filter {item.TopicFilter}, result code: {item.ResultCode}");
}
}
return Task.CompletedTask;
}
}
@@ -11,6 +11,7 @@
<Description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.AspNetCore</PackageId>
<PackageReadmeFile>README.md</PackageReadmeFile>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -37,12 +38,10 @@
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
<None Include="..\..\README.md" Pack="true" PackagePath="\" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
</ItemGroup>
@@ -51,11 +50,11 @@
<DefineConstants>RELEASE;NETSTANDARD2_0</DefineConstants>
</PropertyGroup>
<ItemGroup Condition="'$(TargetFramework)' == 'netcoreapp3.1' or '$(TargetFramework)' == 'net5.0' or '$(TargetFramework)' == 'net6.0' or '$(TargetFramework)' == 'net7.0' ">
<ItemGroup Condition="'$(TargetFramework)' != 'netstandard2.0' and '$(TargetFramework)' != 'netcoreapp2.1' ">
<FrameworkReference Include="Microsoft.AspNetCore.App" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)' != 'netcoreapp3.1' and '$(TargetFramework)' != 'net5.0' and '$(TargetFramework)' != 'net6.0' and '$(TargetFramework)' != 'net7.0' ">
<ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0' or '$(TargetFramework)' == 'netcoreapp2.1' ">
<PackageReference Include="Microsoft.AspNetCore.Http.Connections" Version="1.1.0" />
</ItemGroup>
@@ -25,6 +25,8 @@ namespace MQTTnet.Extensions.ManagedClient
event Func<ManagedProcessFailedEventArgs, Task> SynchronizingSubscriptionsFailedAsync;
event Func<SubscriptionsChangedEventArgs, Task> SubscriptionsChangedAsync;
IMqttClient InternalClient { get; }
bool IsConnected { get; }
@@ -13,6 +13,7 @@
<Description>This is an extension library which provides a managed MQTT client with additional features using MQTTnet.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.ManagedClient</PackageId>
<PackageReadmeFile>README.md</PackageReadmeFile>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -50,10 +51,8 @@
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
<None Include="..\..\README.md" Pack="true" PackagePath="\" />
</ItemGroup>
<ItemGroup>
@@ -25,7 +25,8 @@ namespace MQTTnet.Extensions.ManagedClient
readonly AsyncEvent<ConnectingFailedEventArgs> _connectingFailedEvent = new AsyncEvent<ConnectingFailedEventArgs>();
readonly AsyncEvent<EventArgs> _connectionStateChangedEvent = new AsyncEvent<EventArgs>();
readonly AsyncEvent<ManagedProcessFailedEventArgs> _synchronizingSubscriptionsFailedEvent = new AsyncEvent<ManagedProcessFailedEventArgs>();
readonly AsyncEvent<SubscriptionsChangedEventArgs> _subscriptionsChangedEvent = new AsyncEvent<SubscriptionsChangedEventArgs>();
readonly MqttNetSourceLogger _logger;
readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
@@ -133,6 +134,12 @@ namespace MQTTnet.Extensions.ManagedClient
remove => _synchronizingSubscriptionsFailedEvent.RemoveHandler(value);
}
public event Func<SubscriptionsChangedEventArgs, Task> SubscriptionsChangedAsync
{
add => _subscriptionsChangedEvent.AddHandler(value);
remove => _subscriptionsChangedEvent.RemoveHandler(value);
}
public IMqttClient InternalClient { get; }
public bool IsConnected => InternalClient.IsConnected;
@@ -372,6 +379,14 @@ namespace MQTTnet.Extensions.ManagedClient
}
}
async Task HandleSubscriptionsResultAsync(SendSubscribeUnsubscribeResult subscribeUnsubscribeResult)
{
if (_subscriptionsChangedEvent.HasHandlers)
{
await _subscriptionsChangedEvent.InvokeAsync(new SubscriptionsChangedEventArgs(subscribeUnsubscribeResult.SubscribeResults, subscribeUnsubscribeResult.UnsubscribeResults)).ConfigureAwait(false);
}
}
async Task MaintainConnectionAsync(CancellationToken cancellationToken)
{
try
@@ -473,19 +488,22 @@ namespace MQTTnet.Extensions.ManagedClient
if (_reconnectSubscriptions.Any())
{
topicFilters = new List<MqttTopicFilter>();
SendSubscribeUnsubscribeResult subscribeUnsubscribeResult;
foreach (var sub in _reconnectSubscriptions)
{
topicFilters.Add(sub.Value);
if (topicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
{
await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
topicFilters.Clear();
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
}
}
await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false);
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
}
}
catch (Exception exception)
@@ -501,6 +519,7 @@ namespace MQTTnet.Extensions.ManagedClient
while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false))
{
List<MqttTopicFilter> subscriptions;
SendSubscribeUnsubscribeResult subscribeUnsubscribeResult;
HashSet<string> unsubscriptions;
lock (_subscriptions)
@@ -536,12 +555,14 @@ namespace MQTTnet.Extensions.ManagedClient
if (addedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
{
await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
addedTopicFilters.Clear();
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
}
}
await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false);
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
var removedTopicFilters = new List<string>();
foreach (var unSub in unsubscriptions)
@@ -550,12 +571,14 @@ namespace MQTTnet.Extensions.ManagedClient
if (removedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets)
{
await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
removedTopicFilters.Clear();
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
}
}
await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
subscribeUnsubscribeResult = await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false);
await HandleSubscriptionsResultAsync(subscribeUnsubscribeResult).ConfigureAwait(false);
}
}
@@ -588,8 +611,10 @@ namespace MQTTnet.Extensions.ManagedClient
}
}
async Task SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
async Task<SendSubscribeUnsubscribeResult> SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
{
var subscribeResults = new List<MqttClientSubscribeResult>();
var unsubscribeResults = new List<MqttClientUnsubscribeResult>();
try
{
if (removedSubscriptions != null && removedSubscriptions.Any())
@@ -601,7 +626,8 @@ namespace MQTTnet.Extensions.ManagedClient
unsubscribeOptionsBuilder.WithTopicFilter(removedSubscription);
}
await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false);
var unsubscribeResult = await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false);
unsubscribeResults.Add(unsubscribeResult);
//clear because these worked, maybe the subscribe below will fail, only report those
removedSubscriptions.Clear();
@@ -616,13 +642,16 @@ namespace MQTTnet.Extensions.ManagedClient
subscribeOptionsBuilder.WithTopicFilter(addedSubscription);
}
await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false);
var subscribeResult = await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false);
subscribeResults.Add(subscribeResult);
}
}
catch (Exception exception)
{
await HandleSubscriptionExceptionAsync(exception, addedSubscriptions, removedSubscriptions).ConfigureAwait(false);
}
return new SendSubscribeUnsubscribeResult(subscribeResults, unsubscribeResults);
}
void StartPublishing()
@@ -0,0 +1,23 @@
// 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;
using System.Collections.Generic;
using MQTTnet.Client;
namespace MQTTnet.Extensions.ManagedClient
{
public sealed class SendSubscribeUnsubscribeResult
{
public SendSubscribeUnsubscribeResult(List<MqttClientSubscribeResult> subscribeResults, List<MqttClientUnsubscribeResult> unsubscribeResults)
{
SubscribeResults = subscribeResults ?? throw new ArgumentNullException(nameof(subscribeResults));
UnsubscribeResults = unsubscribeResults ?? throw new ArgumentNullException(nameof(unsubscribeResults));
}
public List<MqttClientSubscribeResult> SubscribeResults { get; private set; }
public List<MqttClientUnsubscribeResult> UnsubscribeResults { get; private set; }
}
}
@@ -0,0 +1,23 @@
// 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 MQTTnet.Client;
using System;
using System.Collections.Generic;
namespace MQTTnet.Extensions.ManagedClient
{
public sealed class SubscriptionsChangedEventArgs : EventArgs
{
public SubscriptionsChangedEventArgs(List<MqttClientSubscribeResult> subscribeResult, List<MqttClientUnsubscribeResult> unsubscribeResult)
{
SubscribeResult = subscribeResult ?? throw new ArgumentNullException(nameof(subscribeResult));
UnsubscribeResult = unsubscribeResult ?? throw new ArgumentNullException(nameof(unsubscribeResult));
}
public List<MqttClientSubscribeResult> SubscribeResult { get; }
public List<MqttClientUnsubscribeResult> UnsubscribeResult { get; }
}
}
@@ -13,6 +13,7 @@
<Description>This is an extension library which allows executing synchronous device calls including a response using MQTTnet.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.Rpc</PackageId>
<PackageReadmeFile>README.md</PackageReadmeFile>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -51,10 +52,8 @@
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
<None Include="..\..\README.md" Pack="true" PackagePath="\" />
</ItemGroup>
<ItemGroup>
@@ -13,6 +13,7 @@
<Description>This is an extension library which allows using _WebSocket4Net_ as transport for MQTTnet clients.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.WebSocket4Net</PackageId>
<PackageReadmeFile>README.md</PackageReadmeFile>
<SignAssembly>false</SignAssembly>
<DelaySign>false</DelaySign>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -51,10 +52,8 @@
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
<None Include="..\..\README.md" Pack="true" PackagePath="\" />
</ItemGroup>
<ItemGroup>
@@ -76,7 +76,7 @@ namespace MQTTnet.Extensions.WebSocket4Net
List<KeyValuePair<string, string>> customHeaders = null;
if (_webSocketOptions.RequestHeaders != null)
{
customHeaders = _webSocketOptions.RequestHeaders.Select(i => new KeyValuePair<string, string>(i.Key, i.Value)).ToList();
customHeaders = _webSocketOptions.RequestHeaders.ToList();
}
EndPoint proxy = null;
@@ -65,7 +65,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
return CompletedTask.Instance;
};
await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").Build());
await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").Build());
Assert.Fail("Must fail!");
}
@@ -175,7 +175,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
testEnvironment.IgnoreClientLogErrors = true;
var client = testEnvironment.CreateClient();
var options = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").WithTimeout(TimeSpan.FromSeconds(2)).Build();
var options = new MqttClientOptionsBuilder().WithTcpServer("1.2.3.4").WithTimeout(TimeSpan.FromSeconds(2)).Build();
try
{
@@ -240,9 +240,9 @@ namespace MQTTnet.Tests.Clients.MqttClient
await mqttClient.SubscribeAsync("test", MqttQualityOfServiceLevel.AtLeastOnce);
}
}
catch (MqttCommunicationException exception)
catch (MqttClientNotConnectedException exception)
{
if (exception.Message == "The client is not connected.")
if (exception.Message == "The MQTT client is not connected.")
{
return;
}
+12 -3
View File
@@ -141,7 +141,8 @@ namespace MQTTnet.Adapter
try
{
PacketInspector?.BeginReceivePacket();
var localPacketInspector = PacketInspector;
localPacketInspector?.BeginReceivePacket();
ReceivedMqttPacket receivedPacket;
var receivedPacketTask = ReceiveAsync(cancellationToken);
@@ -159,7 +160,10 @@ namespace MQTTnet.Adapter
return null;
}
PacketInspector?.EndReceivePacket();
if (localPacketInspector != null)
{
await localPacketInspector.EndReceivePacket().ConfigureAwait(false);
}
Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength);
@@ -215,8 +219,13 @@ namespace MQTTnet.Adapter
try
{
var packetBuffer = PacketFormatterAdapter.Encode(packet);
PacketInspector?.BeginSendPacket(packetBuffer);
var localPacketInspector = PacketInspector;
if (localPacketInspector != null)
{
await localPacketInspector.BeginSendPacket(packetBuffer).ConfigureAwait(false);
}
_logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet);
if (packetBuffer.Payload.Count == 0 || !AllowPacketFragmentation)
+30 -25
View File
@@ -4,6 +4,7 @@
using System;
using System.IO;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
using MQTTnet.Internal;
@@ -12,16 +13,20 @@ namespace MQTTnet.Adapter
{
public sealed class MqttPacketInspector
{
readonly MqttNetSourceLogger _logger;
readonly AsyncEvent<InspectMqttPacketEventArgs> _asyncEvent;
readonly MqttNetSourceLogger _logger;
MemoryStream _receivedPacketBuffer;
public MqttPacketInspector(AsyncEvent<InspectMqttPacketEventArgs> asyncEvent, IMqttNetLogger logger)
{
_asyncEvent = asyncEvent ?? throw new ArgumentNullException(nameof(asyncEvent));
if (logger == null) throw new ArgumentNullException(nameof(logger));
if (logger == null)
{
throw new ArgumentNullException(nameof(logger));
}
_logger = logger.WithSource(nameof(MqttPacketInspector));
}
@@ -36,28 +41,15 @@ namespace MQTTnet.Adapter
{
_receivedPacketBuffer = new MemoryStream();
}
_receivedPacketBuffer?.SetLength(0);
}
public void EndReceivePacket()
public Task BeginSendPacket(MqttPacketBuffer buffer)
{
if (!_asyncEvent.HasHandlers)
{
return;
}
var buffer = _receivedPacketBuffer.ToArray();
_receivedPacketBuffer.SetLength(0);
InspectPacket(buffer, MqttPacketFlowDirection.Inbound);
}
public void BeginSendPacket(MqttPacketBuffer buffer)
{
if (!_asyncEvent.HasHandlers)
{
return;
return CompletedTask.Instance;
}
// Create a copy of the actual packet so that the inspector gets no access
@@ -65,7 +57,20 @@ namespace MQTTnet.Adapter
// intended for debugging etc. so that this is OK.
var bufferCopy = buffer.ToArray();
InspectPacket(bufferCopy, MqttPacketFlowDirection.Outbound);
return InspectPacket(bufferCopy, MqttPacketFlowDirection.Outbound);
}
public Task EndReceivePacket()
{
if (!_asyncEvent.HasHandlers)
{
return CompletedTask.Instance;
}
var buffer = _receivedPacketBuffer.ToArray();
_receivedPacketBuffer.SetLength(0);
return InspectPacket(buffer, MqttPacketFlowDirection.Inbound);
}
public void FillReceiveBuffer(byte[] buffer)
@@ -74,16 +79,16 @@ namespace MQTTnet.Adapter
{
return;
}
_receivedPacketBuffer?.Write(buffer, 0, buffer.Length);
}
void InspectPacket(byte[] buffer, MqttPacketFlowDirection direction)
async Task InspectPacket(byte[] buffer, MqttPacketFlowDirection direction)
{
try
{
var eventArgs = new InspectMqttPacketEventArgs(direction, buffer);
_asyncEvent.InvokeAsync(eventArgs).GetAwaiter().GetResult();
await _asyncEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
catch (Exception exception)
{
+1 -1
View File
@@ -930,7 +930,7 @@ namespace MQTTnet.Client
static void ThrowNotConnected()
{
throw new MqttCommunicationException("The client is not connected.");
throw new MqttClientNotConnectedException();
}
void TryInitiateDisconnect()
@@ -0,0 +1,19 @@
// 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.Exceptions
{
public class MqttClientNotConnectedException : MqttCommunicationException
{
public MqttClientNotConnectedException() : base("The MQTT client is not connected.")
{
}
public MqttClientNotConnectedException(Exception innerException) : base("The MQTT client is not connected.", innerException)
{
}
}
}
@@ -279,7 +279,7 @@ namespace MQTTnet.Formatter.V5
_propertiesWriter.WriteReasonString(packet.ReasonString);
_propertiesWriter.WriteUserProperties(packet.UserProperties);
if (_bufferWriter.Length > 0 || packet.ReasonCode != MqttPubAckReasonCode.Success)
if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubAckReasonCode.Success)
{
_bufferWriter.WriteByte((byte)packet.ReasonCode);
_propertiesWriter.WriteTo(_bufferWriter);
@@ -375,7 +375,7 @@ namespace MQTTnet.Formatter.V5
_bufferWriter.WriteTwoByteInteger(packet.PacketIdentifier);
if (_bufferWriter.Length > 0 || packet.ReasonCode != MqttPubRecReasonCode.Success)
if (_propertiesWriter.Length > 0 || packet.ReasonCode != MqttPubRecReasonCode.Success)
{
_bufferWriter.WriteByte((byte)packet.ReasonCode);
_propertiesWriter.WriteTo(_bufferWriter);
@@ -534,4 +534,4 @@ namespace MQTTnet.Formatter.V5
}
}
}
}
}
+3 -4
View File
@@ -23,6 +23,7 @@
<Description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</Description>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet</PackageId>
<PackageReadmeFile>README.md</PackageReadmeFile>
<DelaySign>false</DelaySign>
<SignAssembly>false</SignAssembly>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
@@ -69,10 +70,8 @@
</PropertyGroup>
<ItemGroup>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png" Pack="True" PackagePath="\" />
<None Include="..\..\README.md" Pack="true" PackagePath="\" />
</ItemGroup>
<ItemGroup>
@@ -57,7 +57,7 @@ namespace MQTTnet.Server
while (!cancellationToken.IsCancellationRequested)
{
TryProcessClients();
Sleep(_options.KeepAliveMonitorInterval);
Sleep(_options.KeepAliveOptions.MonitorInterval);
}
}
catch (OperationCanceledException)