Compare commits

...
11 Commits
Author SHA1 Message Date
christian 740d605fd2 Fix Unit Tests 2024-09-07 11:32:14 +02:00
christian 9177b6ae68 Fix Unit Tests 2024-09-07 11:22:43 +02:00
Christian f0ea5c9ecd Fix encoding and decoding of AUTH paket according to RFC. (#2076) 2024-09-07 10:52:53 +02:00
Christian 76479ebabe Setup nuget and code signing for version 4.x.x (#2069)
* Update CI

* Update CI

* Update release notes
2024-08-24 14:58:13 +02:00
pigwing e18a91a4b5 fix LoadingRetainedMessageAsync not exec (#2025)
because IHostedService StartAsync before startup Configure
see https://learn.microsoft.com/en-us/aspnet/core/fundamentals/host/generic-host?view=aspnetcore-8.0#ihostapplicationlifetime
2024-06-22 15:24:25 +02:00
drelidan7 b95d352236 Switch SubscribeAsync/UnsubscribeAsync to IEnumerable<string> (#2026)
The function currently requires an ICollection<string>, but doesn't actually depend on the ICollection<string> interface.  Relax the requirement to be IEnumerable<string> for better flexibility.
2024-06-22 14:55:20 +02:00
simonthum c5f4cf1f9b Topic template extension maintenance (#2022)
* Topic templates: Ensure parameter values are checked for wildcard

* Topic templates: Complement extension methods

- add template.BuildMessage()
- add subscribeOptionsBuilder.WithTopicTemplate()

* Topic templates: switch some samples to use a topic template

The changes should be self-explanatory and are
intended to provide more examples.

* Topic templates: add README
2024-06-22 14:54:11 +02:00
Josua Jäger 8650002523 fix Managed_Client_Subscribe_Samples.Connect_Client sample: (#2023)
- Result was never logged (`bool subscribed`)
- `subscribed` was never set, as it was not passed as a reference but as a value.
2024-06-21 19:05:12 +02:00
Christian 25c1da6a21 Restore behavior of _Server_ and _Port_ in client options (#2009)
* Restore old server and port behavior

* Fix unit test
2024-05-23 19:32:22 +02:00
christian 4e3ede42c6 Increase version 2024-05-21 22:36:34 +02:00
Christian 918a3c8d8b Restore old signing to get out hotfix release (#2004) 2024-05-19 16:39:49 +02:00
24 changed files with 491 additions and 137 deletions
+5 -4
View File
@@ -1,4 +1,5 @@
* [Client] Fix _None of the discovered or specified addresses match the socket address family._ (#1997).
* [Client] Remove the obsolete attribute from the _WithConnectionUri_ methods (#1979).
* All nuget packages are now signed.
* Core: Fixed issue when parsing AUTH packet with 0 length body (#2039).
* nuget: Changed code signing and nuget certificate (**BREAKING CHANGE**).
* TopicTemplates: Updated samples, parameter validation (#2022).
* ManagedClient: Switch SubscribeAsync/UnsubscribeAsync to IEnumerable<string> (#2026).
* Server: Fix _LoadingRetainedMessageAsync_ not executed (#2025).
+9 -3
View File
@@ -3,7 +3,7 @@ name: CI
on: [push, pull_request]
env:
VERSION: "4.3.5.${{github.run_number}}"
VERSION: "4.3.7.${{github.run_number}}"
jobs:
build:
@@ -30,11 +30,17 @@ jobs:
- name: Checkout Code
uses: actions/checkout@v4
- name: Setup Signing Certificate
run: |
$secret = '${{ secrets.SNC_BASE64 }}'
$decoded = [System.Convert]::FromBase64CharArray($secret, 0, $secret.Length)
Set-Content -Path ${{ github.workspace }}\certificate.snk -Value $decoded -AsByteStream
- name: Restore nuget packages
run: msbuild MQTTnet.sln /t:Restore /p:Configuration="Release" /verbosity:m
- name: Build solution
run: msbuild MQTTnet.sln /t:Build /p:Configuration="Release" /verbosity:m /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }}
run: msbuild MQTTnet.sln /t:Build /p:Configuration="Release" /verbosity:m /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}\certificate.snk
- name: Setup VSTest
uses: darenm/Setup-VSTest@v1
@@ -62,7 +68,7 @@ jobs:
--azure-key-vault-client-secret "${{ secrets.AZURE_CLIENT_SECRET }}" `
--azure-key-vault-certificate "${{ secrets.KEY_VAULT_CERTIFICATE_ID }}" `
--azure-key-vault-url "${{ secrets.KEY_VAULT_URL }}"
- name: Upload signed nuget packages
uses: actions/upload-artifact@v4
with:
+2
View File
@@ -10,6 +10,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md
LICENSE = LICENSE
README.md = README.md
.github\workflows\ReleaseNotes.md = .github\workflows\ReleaseNotes.md
.github\workflows\ci.yml = .github\workflows\ci.yml
EndProjectSection
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}"
+13 -30
View File
@@ -8,6 +8,7 @@
// ReSharper disable UnusedMember.Local
using MQTTnet.Client;
using MQTTnet.Extensions.TopicTemplate;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Samples.Helpers;
@@ -16,6 +17,8 @@ namespace MQTTnet.Samples.Client;
public static class Client_Subscribe_Samples
{
static MqttTopicTemplate sampleTemplate = new MqttTopicTemplate("mqttnet/samples/topic/{id}");
public static async Task Handle_Received_Application_Message()
{
/*
@@ -42,11 +45,7 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter(
f =>
{
f.WithTopic("mqttnet/samples/topic/2");
})
.WithTopicTemplate(sampleTemplate.WithParameter("id", "2"))
.Build();
await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -88,11 +87,8 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter(
f =>
{
f.WithTopic("mqttnet/samples/topic/1");
})
.WithTopicTemplate(
sampleTemplate.WithParameter("id", "1"))
.Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -121,21 +117,12 @@ public static class Client_Subscribe_Samples
// Create the subscribe options including several topics with different options.
// It is also possible to all of these topics using a dedicated call of _SubscribeAsync_ per topic.
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter(
f =>
{
f.WithTopic("mqttnet/samples/topic/1");
})
.WithTopicFilter(
f =>
{
f.WithTopic("mqttnet/samples/topic/2").WithNoLocal();
})
.WithTopicFilter(
f =>
{
f.WithTopic("mqttnet/samples/topic/3").WithRetainHandling(MqttRetainHandling.SendAtSubscribe);
})
.WithTopicTemplate(
sampleTemplate.WithParameter("id", "1"))
.WithTopicTemplate(
sampleTemplate.WithParameter("id", "2"), noLocal: true)
.WithTopicTemplate(
sampleTemplate.WithParameter("id", "3"), retainHandling: MqttRetainHandling.SendAtSubscribe)
.Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -162,11 +149,7 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter(
f =>
{
f.WithTopic("mqttnet/samples/topic/1");
})
.WithTopicTemplate(sampleTemplate.WithParameter("id", "1"))
.Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
+1
View File
@@ -17,6 +17,7 @@
<ProjectReference Include="..\Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj" />
<ProjectReference Include="..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj" />
<ProjectReference Include="..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj" />
<ProjectReference Include="..\Source\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj" />
<ProjectReference Include="..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj" />
<ProjectReference Include="..\Source\MQTTnet\MQTTnet.csproj" />
</ItemGroup>
@@ -22,7 +22,7 @@ public sealed class Managed_Client_Subscribe_Samples
* - 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;
@@ -43,21 +43,21 @@ public sealed class Managed_Client_Subscribe_Samples
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);
managedMqttClient.SubscriptionsChangedAsync += args => SubscriptionsResultAsync(args, ref subscribed);
await managedMqttClient.SubscribeAsync("Topic").ConfigureAwait(false);
SpinWait.SpinUntil(() => subscribed, 1000);
Console.WriteLine("Subscription properly done");
Console.WriteLine($"Subscription properly done {subscribed}");
}
}
private static Task SubscriptionsResultAsync(SubscriptionsChangedEventArgs arg, bool subscribed)
private static Task SubscriptionsResultAsync(SubscriptionsChangedEventArgs arg, ref bool subscribed)
{
foreach (var mqttClientSubscribeResult in arg.SubscribeResult)
{
+28 -3
View File
@@ -16,26 +16,51 @@ namespace MQTTnet.AspNetCore
public sealed class MqttHostedServer : MqttServer, IHostedService
{
readonly MqttFactory _mqttFactory;
public MqttHostedServer(MqttFactory mqttFactory, MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(
#if NETCOREAPP3_1_OR_GREATER
readonly IHostApplicationLifetime _hostApplicationLifetime;
public MqttHostedServer(IHostApplicationLifetime hostApplicationLifetime, MqttFactory mqttFactory,
MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(
options,
adapters,
logger)
{
_mqttFactory = mqttFactory ?? throw new ArgumentNullException(nameof(mqttFactory));
_hostApplicationLifetime = hostApplicationLifetime;
}
#else
public MqttHostedServer(MqttFactory mqttFactory,
MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(
options,
adapters,
logger)
{
_mqttFactory = mqttFactory ?? throw new ArgumentNullException(nameof(mqttFactory));
}
#endif
public async Task StartAsync(CancellationToken cancellationToken)
{
// The yield makes sure that the hosted service is considered up and running.
await Task.Yield();
#if NETCOREAPP3_1_OR_GREATER
_hostApplicationLifetime.ApplicationStarted.Register(OnStarted);
#else
_ = StartAsync();
#endif
}
public Task StopAsync(CancellationToken cancellationToken)
{
return StopAsync(_mqttFactory.CreateMqttServerStopOptionsBuilder().Build());
}
#if NETCOREAPP3_1_OR_GREATER
private void OnStarted()
{
_ = StartAsync();
}
#endif
}
}
@@ -47,8 +47,8 @@ namespace MQTTnet.Extensions.ManagedClient
Task StopAsync(bool cleanDisconnect = true);
Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters);
Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters);
Task UnsubscribeAsync(ICollection<string> topics);
Task UnsubscribeAsync(IEnumerable<string> topics);
}
}
@@ -288,7 +288,7 @@ namespace MQTTnet.Extensions.ManagedClient
}
}
public Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters)
public Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters)
{
ThrowIfDisposed();
@@ -316,7 +316,7 @@ namespace MQTTnet.Extensions.ManagedClient
return CompletedTask.Instance;
}
public Task UnsubscribeAsync(ICollection<string> topics)
public Task UnsubscribeAsync(IEnumerable<string> topics)
{
ThrowIfDisposed();
@@ -11,10 +11,10 @@
<Company>The contributors of MQTTnet</Company>
<Product>MQTTnet</Product>
<Description>
This is an extension library which provides mqtt topic templating logic to support dispatch,
routing and similar functionality based on the well known moustache syntax, aka
AsyncAPI dynamic channel address.
Provides mqtt topic templating logic to support dispatch,
routing and similar functionality based on the well known moustache syntax (AsyncAPI compatible).
</Description>
<PackageReadmeFile>README.md</PackageReadmeFile>
<Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.TopicTemplate</PackageId>
<SignAssembly>false</SignAssembly>
@@ -51,6 +51,10 @@
</PropertyGroup>
<ItemGroup>
<None Include="README.md">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
@@ -16,8 +16,8 @@ namespace MQTTnet.Extensions.TopicTemplate
/// segments in curly braces called parameters. This well-known
/// 'moustache' syntax also matches AsyncAPI Channel Address Expressions.
/// The topic template is designed to support dynamic subscription/publication,
/// message-topic matching and routing. It is intended to be more convenient
/// than String.Format() for aforementioned purposes.
/// message-topic matching and routing. It is intended to be more safe and
/// convenient than String.Format() for aforementioned purposes.
/// </summary>
/// <example>
/// topic/subtopic/{parameter}/{otherParameter}
@@ -282,12 +282,17 @@ namespace MQTTnet.Extensions.TopicTemplate
/// <returns>the topic template (without the parameter)</returns>
public MqttTopicTemplate WithoutParameter(string parameter)
{
return WithParameter(parameter, MqttTopicFilterComparer.SingleLevelWildcard.ToString());
if (string.IsNullOrEmpty(parameter) || !_parameterSegments.Contains(parameter))
{
throw new ArgumentException("topic template parameter must exist.");
}
return ReplaceInternal(parameter, MqttTopicFilterComparer.SingleLevelWildcard.ToString());
}
/// <summary>
/// Substitute a parameter with a given value, thus removing the parameter. If the parameter is not present,
/// the method trows. The value must not contain slashes.
/// the method trows. The value must not contain slashes or wildcards.
/// </summary>
/// <param
/// name="parameter">
@@ -304,12 +309,19 @@ namespace MQTTnet.Extensions.TopicTemplate
/// <returns>the topic template (without the parameter)</returns>
public MqttTopicTemplate WithParameter(string parameter, string value)
{
if (value == null || string.IsNullOrEmpty(parameter) || !_parameterSegments.Contains(parameter) || value.Contains(MqttTopicFilterComparer.LevelSeparator) ||
if (value == null || string.IsNullOrEmpty(parameter) || !_parameterSegments.Contains(parameter) ||
value.Contains(MqttTopicFilterComparer.LevelSeparator) ||
value.Contains(MqttTopicFilterComparer.SingleLevelWildcard) ||
value.Contains(MqttTopicFilterComparer.MultiLevelWildcard))
{
throw new ArgumentException("parameter must exist and value must not contain slashes.");
throw new ArgumentException("parameter must exist and value must not contain slashes or wildcard.");
}
return ReplaceInternal(parameter, value);
}
private MqttTopicTemplate ReplaceInternal(string parameter, string value)
{
var moustache = "{" + parameter + "}";
return new MqttTopicTemplate(Template.Replace(moustache, value));
}
@@ -0,0 +1,79 @@
# MQTTNet.Extensions.TopicTemplate
Provides mqtt topic templating logic to support dispatch, routing and
similar functionality based on the well known moustache syntax, aka
AsyncAPI dynamic channel address.
Generating and parsing MQTT topics and topic filters is often done in
an ad-hoc fashion, leading to error-prone code and subtly introduced
showstoppers like forgetting to strip a slash from a parameter. To
remedy, this extension aids you write safe, maintainable logic
for topic generation, routing and dispatch.
## How it works
The package lets you write MQTTNet code like:
```csharp
var template = new MqttTopicTemplate("App/v1/{sender}/message");
var filter = new MqttTopicFilterBuilder()
.WithTopicTemplate(template) // (1)
.WithAtLeastOnceQoS()
.WithNoLocal()
.Build();
// filter.Topic == "App/v1/+/message"
var myTopic = template
.WithParameter("sender", "me, myself & i"); // (2)
var message = new MqttApplicationMessageBuilder()
.WithTopicTemplate( // (3)
myTopic)
.WithPayload("Hello!")
.Build();
// message.Topic == "App/v1/me, myself & i/message"
Assert.IsTrue(message.MatchesTopicTemplate(template)); // (4)
```
At (1), the parameter "sender" was automatically replaced by a plus sign
(single level wildcard) because the caller is creating a topic filter.
Using the same template (they are immutable), you can send
a message. At (2), we replace the parameter with a known value.
If that value contained a level separator or topic filter only
characters, we'd get an ArgumentException. If we forgot to replace
the parameter "sender" (e.g. when placing template not myTopic),
we would be prevented from sending at (3).
On the receiving side, a message can be matched to a template
using code like at (4).
These daily tasks become much safer and easier to code than
string.Format() and Split() allow for. More elaborate computation is supported
too. You could extract the sender using the template or route
messages based on topic content.
## Features
- Safe handling of topic parameters in moustache syntax
- Simplifies topic generation and parsing logic
- Match topics and extract the values of their parameters
- Translate between topic templates to ease dynamic routing
- Derive a canonical topic filter from a bunch of templates
- Integrates with MQTTNet message, subscription and filter builders
- Extension method friendly (sealed class with one public ctor)
## Additional documentation
- Some of the [unit tests](https://github.com/dotnet/MQTTnet/blob/master/Source/MQTTnet.Tests/Extensions/MqttTopicTemplate_Tests.cs) are instructive
- Some examples feature the extension, e.g. TopicGenerator, Client_Subscribe_Samples
## Feedback
Report any issues with the main project on https://github.com/dotnet/MQTTnet/issues
@@ -4,6 +4,8 @@
using System;
using System.Linq;
using MQTTnet.Client;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Extensions.TopicTemplate
@@ -55,6 +57,24 @@ namespace MQTTnet.Extensions.TopicTemplate
return new MqttTopicFilterBuilder().WithTopicTemplate(topicTemplate, subscribeTreeRoot);
}
/// <summary>
/// Create a message builder from this template. The template must not have
/// remaining parameters.
/// </summary>
/// <param
/// name="topicTemplate">
/// a parameterless topic template
/// </param>
/// <returns>a new message builder</returns>
/// <exception
/// cref="ArgumentException">
/// if the topic template has parameters
/// </exception>
public static MqttApplicationMessageBuilder BuildMessage(this MqttTopicTemplate topicTemplate)
{
return new MqttApplicationMessageBuilder().WithTopicTemplate(topicTemplate);
}
/// <summary>
/// Return a message builder to respond to this message. The
/// message's response topic and correlation data are included
@@ -116,6 +136,29 @@ namespace MQTTnet.Extensions.TopicTemplate
return builder.WithTopic(subscribeTreeRoot ? topicTemplate.TopicTreeRootFilter : topicTemplate.TopicFilter);
}
/// <summary>
/// Set the subscription to the template's topic filter.
/// </summary>
/// <returns>the builder</returns>
public static MqttClientSubscribeOptionsBuilder WithTopicTemplate(
this MqttClientSubscribeOptionsBuilder builder,
MqttTopicTemplate topicTemplate,
MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
bool noLocal = false,
bool retainAsPublished = false,
MqttRetainHandling retainHandling = MqttRetainHandling.SendAtSubscribe)
{
return builder.WithTopicFilter(
new MqttTopicFilter
{
Topic = topicTemplate.TopicFilter,
QualityOfServiceLevel = qualityOfServiceLevel,
NoLocal = noLocal,
RetainAsPublished = retainAsPublished,
RetainHandling = retainHandling
});
}
/// <summary>
/// Set the publication topic according to the topic template. The template
/// must not have remaining (unset) parameters or contain wildcards.
@@ -21,6 +21,7 @@
<ProjectReference Include="..\..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj" />
<ProjectReference Include="..\..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj" />
<ProjectReference Include="..\..\Source\MQTTnet\MQTTnet.csproj" />
<ProjectReference Include="..\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj" />
</ItemGroup>
</Project>
+18 -4
View File
@@ -1,6 +1,6 @@
using System;
using System.Collections.Generic;
using System.Text;
using MQTTnet.Extensions.TopicTemplate;
namespace MQTTnet.TestApp
{
@@ -17,6 +17,8 @@ namespace MQTTnet.TestApp
singleWildcardTopicsByPublisher = new Dictionary<string, List<string>>();
multiWildcardTopicsByPublisher = new Dictionary<string, List<string>>();
MqttTopicTemplate baseTemplate = new MqttTopicTemplate("{publisher}/{building}/{level}/{sensor}");
// Find some reasonable distribution across three topic levels
var topicsPerLevel = (int)Math.Pow(numTopicsPerPublisher, (1.0 / 3.0));
@@ -40,25 +42,37 @@ namespace MQTTnet.TestApp
int publisherTopicCount = 0;
var publisherName = "pub" + p;
var publisherTemplate = baseTemplate
.WithParameter("publisher", publisherName);
for (var l1 = 0; l1 < numLevel1Topics; ++l1)
{
var l1Template = publisherTemplate
.WithParameter("building", "building" + (l1 + 1));
for (var l2 = 0; l2 < numLevel2Topics; ++l2)
{
var l2Template = l1Template
.WithParameter("level", "level" + (l2 + 1));
for (var l3 = 0; l3 < maxNumLevel3Topics; ++l3)
{
if (publisherTopicCount >= numTopicsPerPublisher)
break;
var topic = string.Format("{0}/building{1}/level{2}/sensor{3}", publisherName, l1 + 1, l2 + 1, l3 + 1);
var topic = l2Template
.WithParameter("sensor", "sensor" + (l3 + 1))
.TopicFilter;
AddPublisherTopic(publisherName, topic, topicsByPublisher);
if (l2 == 0)
{
var singleWildcardTopic = string.Format("{0}/building{1}/+/sensor{2}", publisherName, l1 + 1, l3 + 1);
var singleWildcardTopic = l1Template
.WithParameter("sensor", "sensor" + (l3 + 1))
.TopicFilter;
AddPublisherTopic(publisherName, singleWildcardTopic, singleWildcardTopicsByPublisher);
if (l1 == 0)
{
var multiWildcardTopic = string.Format("{0}/+/level{1}/+", publisherName, l3 + 1);
var multiWildcardTopic = publisherTemplate
.WithParameter("sensor", "sensor" + (l3 + 1))
.TopicFilter;
AddPublisherTopic(publisherName, multiWildcardTopic, multiWildcardTopicsByPublisher);
}
}
@@ -6,6 +6,7 @@ using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
@@ -86,7 +87,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
await client.ConnectAsync(clientOptions);
}
}
[TestMethod]
[ExpectedException(typeof(InvalidOperationException))]
public async Task Connect_Multiple_Times_Should_Fail()
@@ -748,8 +749,8 @@ namespace MQTTnet.Tests.Clients.MqttClient
{
if (e.ApplicationMessage.Topic == "request")
{
// Use AtMostOnce here because with QoS 1 or even QoS 2 the process waits for
// the ACK etc. The problem is that the SpinUntil below only waits until the
// Use AtMostOnce here because with QoS 1 or even QoS 2 the process waits for
// the ACK etc. The problem is that the SpinUntil below only waits until the
// flag is set. It does not wait until the client has sent the ACK
await client2.PublishStringAsync("reply");
}
@@ -925,5 +926,54 @@ namespace MQTTnet.Tests.Clients.MqttClient
Assert.IsFalse(disconnectedFired);
}
}
[TestMethod]
public void Backward_compatible_TCP_options()
{
var options = new MqttClientOptionsBuilder().WithTcpServer("host", 3).Build();
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
Assert.AreEqual(3, ((MqttClientTcpOptions)options.ChannelOptions).Port);
options = new MqttClientOptions
{
ChannelOptions = new MqttClientTcpOptions
{
Server = "host",
Port = 3
}
};
Assert.AreEqual("host:3", options.ChannelOptions.ToString());
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
Assert.AreEqual(3, ((MqttClientTcpOptions)options.ChannelOptions).Port);
options = new MqttClientOptionsBuilder().WithEndPoint(new DnsEndPoint("host", 3)).Build();
Assert.AreEqual("Unspecified/host:3", options.ChannelOptions.ToString());
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
Assert.AreEqual(3, ((MqttClientTcpOptions)options.ChannelOptions).Port);
options = new MqttClientOptionsBuilder().WithTcpServer("host").Build();
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
Assert.AreEqual(1883, ((MqttClientTcpOptions)options.ChannelOptions).Port);
Assert.AreEqual("Unspecified/host:1883", options.ChannelOptions.ToString());
options = new MqttClientOptionsBuilder().WithTlsOptions(o => o.UseTls()).WithTcpServer("host").Build();
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
Assert.AreEqual(8883, ((MqttClientTcpOptions)options.ChannelOptions).Port);
options = new MqttClientOptions
{
ChannelOptions = new MqttClientTcpOptions
{
Server = "host"
}
};
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
Assert.AreEqual(null, ((MqttClientTcpOptions)options.ChannelOptions).Port);
}
}
}
@@ -45,6 +45,29 @@ namespace MQTTnet.Tests.Extensions
var template = new MqttTopicTemplate("A/B/{foo}/D");
template.WithParameter("foo", "a#");
}
[TestMethod]
[ExpectedException(typeof(ArgumentException))]
public void RejectsReservedChars2()
{
var template = new MqttTopicTemplate("A/B/{foo}/D");
template.WithParameter("foo", "a+b");
}
[TestMethod]
[ExpectedException(typeof(ArgumentException))]
public void RejectsReservedChars3()
{
var template = new MqttTopicTemplate("A/B/{foo}/D");
template.WithParameter("foo", "a/b");
}
[TestMethod]
public void AcceptsEmptyValue()
{
var template = new MqttTopicTemplate("A/B/{foo}/D");
template.WithParameter("foo", "");
}
[TestMethod]
[ExpectedException(typeof(MqttProtocolViolationException))]
@@ -60,13 +83,6 @@ namespace MQTTnet.Tests.Extensions
var _ = new MqttTopicTemplate(null);
}
[TestMethod]
[ExpectedException(typeof(ArgumentException))]
public void RejectsReservedChars2()
{
var template = new MqttTopicTemplate("A/B/{foo}/D");
template.WithParameter("foo", "e/f");
}
[TestMethod]
public void IgnoresEmptyParameters()
@@ -114,6 +130,50 @@ namespace MQTTnet.Tests.Extensions
Assert.AreEqual("A/v1/+/F", filter.Topic);
}
[TestMethod]
public void SubscriptionSupport2()
{
var template = new MqttTopicTemplate("A/v1/{param}/F");
var subscribeOptions = new MqttFactory().CreateSubscribeOptionsBuilder()
.WithTopicTemplate(template)
.WithSubscriptionIdentifier(5)
.Build();
Assert.AreEqual("A/v1/+/F", subscribeOptions.TopicFilters[0].Topic);
}
[TestMethod]
public void SendAndSubscribeSupport()
{
var template = new MqttTopicTemplate("App/v1/{sender}/message");
var filter = new MqttTopicFilterBuilder()
.WithTopicTemplate(template)
.WithAtLeastOnceQoS()
.WithNoLocal()
.Build();
Assert.AreEqual("App/v1/+/message", filter.Topic);
var myTopic = template.WithParameter("sender", "me, myself & i");
var message = new MqttApplicationMessageBuilder()
.WithTopicTemplate(
myTopic)
.WithPayload("Hello!")
.Build();
Assert.IsTrue(message.MatchesTopicTemplate(template));
}
[TestMethod]
public void SendAndSubscribeSupport2()
{
var template = new MqttTopicTemplate("App/v1/{sender}/message");
Assert.ThrowsException<ArgumentException>(() =>
template.BuildMessage());
}
[TestMethod]
public void CanonicalPrefixFilter()
{
+56 -50
View File
@@ -2,13 +2,13 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Implementations;
using System;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Implementations;
namespace MQTTnet.Tests
{
@@ -18,62 +18,68 @@ namespace MQTTnet.Tests
[TestMethod]
public async Task Dispose_Channel_While_Used()
{
var ct = new CancellationTokenSource();
var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
try
using (var ct = new CancellationTokenSource())
{
serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001));
serverSocket.Listen(0);
#pragma warning disable 4014
Task.Run(async () =>
#pragma warning restore 4014
using (var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp))
{
while (!ct.IsCancellationRequested)
try
{
var client = await serverSocket.AcceptAsync();
var data = new byte[] { 128 };
await client.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
}
}, ct.Token);
serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001));
serverSocket.Listen(0);
var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
await clientSocket.ConnectAsync("localhost", 50001, CancellationToken.None);
var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), "test", null);
await Task.Delay(100, ct.Token);
var buffer = new byte[1];
await tcpChannel.ReadAsync(buffer, 0, 1, ct.Token);
Assert.AreEqual(128, buffer[0]);
// This block should fail after dispose.
#pragma warning disable 4014
Task.Run(() =>
Task.Run(
async () =>
#pragma warning restore 4014
{
Task.Delay(200, ct.Token);
tcpChannel.Dispose();
}, ct.Token);
{
while (!ct.IsCancellationRequested)
{
var client = await serverSocket.AcceptAsync(ct.Token);
var data = new byte[] { 128 };
await client.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
}
},
ct.Token);
try
{
await tcpChannel.ReadAsync(buffer, 0, 1, CancellationToken.None);
var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
await clientSocket.ConnectAsync("localhost", 50001, CancellationToken.None);
var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), "test", null);
await Task.Delay(100, ct.Token);
var buffer = new byte[1];
await tcpChannel.ReadAsync(buffer, 0, 1, ct.Token);
Assert.AreEqual(128, buffer[0]);
// This block should fail after dispose.
#pragma warning disable 4014
Task.Run(
() =>
#pragma warning restore 4014
{
Task.Delay(200, ct.Token);
tcpChannel.Dispose();
},
ct.Token);
try
{
await tcpChannel.ReadAsync(buffer, 0, 1, CancellationToken.None);
}
catch (Exception exception)
{
Assert.IsInstanceOfType(exception, typeof(SocketException));
Assert.AreEqual(SocketError.OperationAborted, ((SocketException)exception).SocketErrorCode);
}
}
finally
{
ct.Cancel(false);
}
}
catch (Exception exception)
{
Assert.IsInstanceOfType(exception, typeof(SocketException));
Assert.AreEqual(SocketError.OperationAborted, ((SocketException)exception).SocketErrorCode);
}
}
finally
{
ct.Cancel(false);
serverSocket.Dispose();
}
}
}
}
}
+17 -4
View File
@@ -1,5 +1,6 @@
using System;
using System.Collections.Generic;
using MQTTnet.Extensions.TopicTemplate;
namespace MQTTnet.Tests
{
@@ -31,31 +32,43 @@ namespace MQTTnet.Tests
{
maxNumLevel3Topics = 1;
}
MqttTopicTemplate baseTemplate = new MqttTopicTemplate("{publisher}/{building}/{level}/{sensor}");
for (var p = 0; p < numPublishers; ++p)
{
int publisherTopicCount = 0;
var publisherName = "pub" + p;
var publisherTemplate = baseTemplate
.WithParameter("publisher", publisherName);
for (var l1 = 0; l1 < numLevel1Topics; ++l1)
{
var l1Template = publisherTemplate.WithParameter("building", "building" + (l1 + 1));
for (var l2 = 0; l2 < numLevel2Topics; ++l2)
{
var l2Template = l1Template.WithParameter("level", "level" + (l2 + 1));
for (var l3 = 0; l3 < maxNumLevel3Topics; ++l3)
{
if (publisherTopicCount >= numTopicsPerPublisher)
break;
var topic = string.Format("{0}/building{1}/level{2}/sensor{3}", publisherName, l1 + 1, l2 + 1, l3 + 1);
AddPublisherTopic(publisherName, topic, topicsByPublisher);
var l3Template = l2Template
.WithParameter("sensor", "sensor" + (l3 + 1));
AddPublisherTopic(publisherName, l3Template.TopicFilter, topicsByPublisher);
if (l2 == 0)
{
var singleWildcardTopic = string.Format("{0}/building{1}/+/sensor{2}", publisherName, l1 + 1, l3 + 1);
var singleWildcardTopic = l1Template
.WithParameter("sensor", "sensor" + (l3 + 1))
.TopicFilter;
AddPublisherTopic(publisherName, singleWildcardTopic, singleWildcardTopicsByPublisher);
}
if ((l1 == 0) && (l3 == 0))
{
var multiWildcardTopic = string.Format("{0}/+/level{1}/+", publisherName, l2 + 1);
var multiWildcardTopic = publisherTemplate
.WithParameter("level", "level" + (l2 + 1))
.TopicFilter;
AddPublisherTopic(publisherName, multiWildcardTopic, multiWildcardTopicsByPublisher);
}
@@ -5,11 +5,13 @@
using System;
using System.Net;
using System.Net.Sockets;
using MQTTnet.Protocol;
namespace MQTTnet.Client
{
public sealed class MqttClientTcpOptions : IMqttClientChannelOptions
{
EndPoint _remoteEndpoint;
public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
public int BufferSize { get; set; } = 8192;
@@ -49,13 +51,56 @@ namespace MQTTnet.Client
/// </summary>
public ProtocolType ProtocolType { get; set; } = ProtocolType.Tcp;
public EndPoint RemoteEndpoint { get; set; }
public EndPoint RemoteEndpoint
{
get => _remoteEndpoint;
set
{
_remoteEndpoint = value;
if (_remoteEndpoint is DnsEndPoint dnsEndPoint)
{
Server = dnsEndPoint.Host;
Port = dnsEndPoint.Port;
}
else if (_remoteEndpoint is IPEndPoint ipEndPoint)
{
Server = ipEndPoint.Address.ToString();
Port = ipEndPoint.Port;
}
}
}
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
public override string ToString()
{
return RemoteEndpoint?.ToString() ?? string.Empty;
if (RemoteEndpoint != null)
{
return RemoteEndpoint.ToString();
}
if (!string.IsNullOrEmpty(Server))
{
return $"{Server}:{GetPort()}";
}
return string.Empty;
}
int GetPort()
{
if (Port.HasValue)
{
return Port.Value;
}
if (TlsOptions?.UseTls == true)
{
return MqttPorts.Secure;
}
return MqttPorts.Default;
}
}
}
@@ -68,12 +68,12 @@ namespace MQTTnet.Formatter.V5
MqttPacket DecodeAuthPacket(ArraySegment<byte> body)
{
ThrowIfBodyIsEmpty(body);
_bufferReader.SetBuffer(body.Array, body.Offset, body.Count);
var packet = new MqttAuthPacket();
// MQTT spec: The Reason Code and Property Length can be omitted if the Reason Code is 0x00 (Success) and there are no Properties.
// In this case the AUTH has a Remaining Length of 0.
if (_bufferReader.EndOfStream)
{
packet.ReasonCode = MqttAuthenticateReasonCode.Success;
@@ -13,7 +13,7 @@ namespace MQTTnet.Formatter.V5
public sealed class MqttV5PacketEncoder
{
const int FixedHeaderSize = 1;
readonly MqttBufferWriter _bufferWriter;
readonly MqttV5PropertiesWriter _propertiesWriter = new MqttV5PropertiesWriter(new MqttBufferWriter(1024, 4096));
@@ -64,13 +64,20 @@ namespace MQTTnet.Formatter.V5
byte EncodeAuthPacket(MqttAuthPacket packet)
{
_bufferWriter.WriteByte((byte)packet.ReasonCode);
_propertiesWriter.WriteAuthenticationMethod(packet.AuthenticationMethod);
_propertiesWriter.WriteAuthenticationData(packet.AuthenticationData);
_propertiesWriter.WriteReasonString(packet.ReasonString);
_propertiesWriter.WriteUserProperties(packet.UserProperties);
// MQTT spec: The Reason Code and Property Length can be omitted if the Reason Code is 0x00 (Success) and there are no Properties.
// In this case the AUTH has a Remaining Length of 0.
if (packet.ReasonCode == MqttAuthenticateReasonCode.Success && _propertiesWriter.Length == 0)
{
return MqttBufferWriter.BuildFixedHeader(MqttControlPacketType.Auth);
}
_bufferWriter.WriteByte((byte)packet.ReasonCode);
_propertiesWriter.WriteTo(_bufferWriter);
_propertiesWriter.Reset();
@@ -526,7 +533,7 @@ namespace MQTTnet.Formatter.V5
static void ThrowIfPacketIdentifierIsInvalid(ushort packetIdentifier, MqttPacket packet)
{
// SUBSCRIBE, UNSUBSCRIBE, and PUBLISH(in cases where QoS > 0) Control Packets MUST contain a non-zero 16 - bit Packet Identifier[MQTT - 2.3.1 - 1].
// SUBSCRIBE, UNSUBSCRIBE, and PUBLISH(in cases where QoS > 0) Control Packets MUST contain a non-zero 16 - bit Packet Identifier[MQTT - 2.3.1 - 1].
if (packetIdentifier == 0)
{
@@ -142,12 +142,14 @@ namespace MQTTnet.Implementations
set => _socket.SendTimeout = value;
}
public async Task<CrossPlatformSocket> AcceptAsync()
public async Task<CrossPlatformSocket> AcceptAsync(CancellationToken cancellationToken)
{
try
{
#if NET452 || NET461
var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false);
#elif NET7_0_OR_GREATER
var clientSocket = await _socket.AcceptAsync(cancellationToken).ConfigureAwait(false);
#else
var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
#endif
@@ -141,7 +141,7 @@ namespace MQTTnet.Implementations
{
try
{
var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
var clientSocket = await _socket.AcceptAsync(cancellationToken).ConfigureAwait(false);
if (clientSocket == null)
{
continue;