Compare 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
22 changed files with 412 additions and 138 deletions
+5 -1
View File
@@ -1 +1,5 @@
* [Client] Restored _Server_ and _Port_ behavior of client options (#2005). * 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).
+30 -12
View File
@@ -3,7 +3,7 @@ name: CI
on: [push, pull_request] on: [push, pull_request]
env: env:
VERSION: "4.3.6.${{github.run_number}}" VERSION: "4.3.7.${{github.run_number}}"
jobs: jobs:
build: build:
@@ -17,7 +17,7 @@ jobs:
sdk-version: 18362 sdk-version: 18362
- name: Setup .NET SDK - name: Setup .NET SDK
uses: actions/setup-dotnet@v1.9.0 uses: actions/setup-dotnet@v4
with: with:
dotnet-version: | dotnet-version: |
3.1.x 3.1.x
@@ -28,7 +28,7 @@ jobs:
uses: microsoft/setup-msbuild@v1.1 uses: microsoft/setup-msbuild@v1.1
- name: Checkout Code - name: Checkout Code
uses: actions/checkout@v2 uses: actions/checkout@v4
- name: Setup Signing Certificate - name: Setup Signing Certificate
run: | run: |
@@ -42,14 +42,6 @@ jobs:
- name: Build solution - 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 }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}\certificate.snk 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: Collect nuget Packages
uses: actions/upload-artifact@v2
with:
name: nuget Packages
path: |
**\*.nupkg
**\*.snupkg
- name: Setup VSTest - name: Setup VSTest
uses: darenm/Setup-VSTest@v1 uses: darenm/Setup-VSTest@v1
@@ -59,6 +51,32 @@ jobs:
- name: ASP.NET Tests - name: ASP.NET Tests
run: vstest.console.exe Source\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\MQTTnet.AspNetCore.Tests.dll run: vstest.console.exe Source\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp3.1\MQTTnet.AspNetCore.Tests.dll
- name: Install sign CLI tool
run: dotnet tool install --tool-path . sign --version 0.9.0-beta.23127.3
- name: Sign nugets
shell: pwsh
run: >
./sign code azure-key-vault `
"**/*.nupkg" `
--base-directory "${{ github.workspace }}/Source" `
--publisher-name "MQTTnet" `
--description "MQTTnet" `
--description-url "https://github.com/dotnet/MQTTnet" `
--azure-key-vault-tenant-id "${{ secrets.AZURE_TENANT_ID }}" `
--azure-key-vault-client-id "${{ secrets.AZURE_CLIENT_ID }}" `
--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:
name: signed-nugets
path: |
${{ github.workspace }}/Source/**/*.nupkg
${{ github.workspace }}/Source/**/*.snupkg
- name: Publish MyGet nugets - name: Publish MyGet nugets
if: ${{ github.event_name == 'push' }} if: ${{ github.event_name == 'push' }}
run: dotnet nuget push **/*.nupkg -k ${{ secrets.MYGET_API_KEY }} -s https://www.myget.org/F/mqttnet/api/v3/index.json --skip-duplicate run: dotnet nuget push ${{ github.workspace }}/Source/**/*.nupkg -k ${{ secrets.MYGET_API_KEY }} -s https://www.myget.org/F/mqttnet/api/v3/index.json --skip-duplicate
+2
View File
@@ -10,6 +10,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md
LICENSE = LICENSE LICENSE = LICENSE
README.md = README.md README.md = README.md
.github\workflows\ReleaseNotes.md = .github\workflows\ReleaseNotes.md
.github\workflows\ci.yml = .github\workflows\ci.yml
EndProjectSection EndProjectSection
EndProject EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}" 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 // ReSharper disable UnusedMember.Local
using MQTTnet.Client; using MQTTnet.Client;
using MQTTnet.Extensions.TopicTemplate;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Samples.Helpers; using MQTTnet.Samples.Helpers;
@@ -16,6 +17,8 @@ namespace MQTTnet.Samples.Client;
public static class Client_Subscribe_Samples public static class Client_Subscribe_Samples
{ {
static MqttTopicTemplate sampleTemplate = new MqttTopicTemplate("mqttnet/samples/topic/{id}");
public static async Task Handle_Received_Application_Message() public static async Task Handle_Received_Application_Message()
{ {
/* /*
@@ -42,11 +45,7 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder() var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter( .WithTopicTemplate(sampleTemplate.WithParameter("id", "2"))
f =>
{
f.WithTopic("mqttnet/samples/topic/2");
})
.Build(); .Build();
await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None); await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -88,11 +87,8 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder() var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter( .WithTopicTemplate(
f => sampleTemplate.WithParameter("id", "1"))
{
f.WithTopic("mqttnet/samples/topic/1");
})
.Build(); .Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None); 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. // 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. // It is also possible to all of these topics using a dedicated call of _SubscribeAsync_ per topic.
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder() var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter( .WithTopicTemplate(
f => sampleTemplate.WithParameter("id", "1"))
{ .WithTopicTemplate(
f.WithTopic("mqttnet/samples/topic/1"); sampleTemplate.WithParameter("id", "2"), noLocal: true)
}) .WithTopicTemplate(
.WithTopicFilter( sampleTemplate.WithParameter("id", "3"), retainHandling: MqttRetainHandling.SendAtSubscribe)
f =>
{
f.WithTopic("mqttnet/samples/topic/2").WithNoLocal();
})
.WithTopicFilter(
f =>
{
f.WithTopic("mqttnet/samples/topic/3").WithRetainHandling(MqttRetainHandling.SendAtSubscribe);
})
.Build(); .Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None); var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -162,11 +149,7 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None); await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder() var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
.WithTopicFilter( .WithTopicTemplate(sampleTemplate.WithParameter("id", "1"))
f =>
{
f.WithTopic("mqttnet/samples/topic/1");
})
.Build(); .Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None); 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.AspnetCore\MQTTnet.AspNetCore.csproj" />
<ProjectReference Include="..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.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.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.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj" />
<ProjectReference Include="..\Source\MQTTnet\MQTTnet.csproj" /> <ProjectReference Include="..\Source\MQTTnet\MQTTnet.csproj" />
</ItemGroup> </ItemGroup>
@@ -49,15 +49,15 @@ public sealed class Managed_Client_Subscribe_Samples
Console.WriteLine($"Pending messages = {managedMqttClient.PendingApplicationMessagesCount}"); 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); await managedMqttClient.SubscribeAsync("Topic").ConfigureAwait(false);
SpinWait.SpinUntil(() => subscribed, 1000); 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) foreach (var mqttClientSubscribeResult in arg.SubscribeResult)
{ {
+28 -3
View File
@@ -16,26 +16,51 @@ namespace MQTTnet.AspNetCore
public sealed class MqttHostedServer : MqttServer, IHostedService public sealed class MqttHostedServer : MqttServer, IHostedService
{ {
readonly MqttFactory _mqttFactory; readonly MqttFactory _mqttFactory;
#if NETCOREAPP3_1_OR_GREATER
public MqttHostedServer(MqttFactory mqttFactory, MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base( 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, options,
adapters, adapters,
logger) logger)
{ {
_mqttFactory = mqttFactory ?? throw new ArgumentNullException(nameof(mqttFactory)); _mqttFactory = mqttFactory ?? throw new ArgumentNullException(nameof(mqttFactory));
} }
#endif
public async Task StartAsync(CancellationToken cancellationToken) public async Task StartAsync(CancellationToken cancellationToken)
{ {
// The yield makes sure that the hosted service is considered up and running. // The yield makes sure that the hosted service is considered up and running.
await Task.Yield(); await Task.Yield();
#if NETCOREAPP3_1_OR_GREATER
_hostApplicationLifetime.ApplicationStarted.Register(OnStarted);
#else
_ = StartAsync(); _ = StartAsync();
#endif
} }
public Task StopAsync(CancellationToken cancellationToken) public Task StopAsync(CancellationToken cancellationToken)
{ {
return StopAsync(_mqttFactory.CreateMqttServerStopOptionsBuilder().Build()); 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 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(); ThrowIfDisposed();
@@ -316,7 +316,7 @@ namespace MQTTnet.Extensions.ManagedClient
return CompletedTask.Instance; return CompletedTask.Instance;
} }
public Task UnsubscribeAsync(ICollection<string> topics) public Task UnsubscribeAsync(IEnumerable<string> topics)
{ {
ThrowIfDisposed(); ThrowIfDisposed();
@@ -11,10 +11,10 @@
<Company>The contributors of MQTTnet</Company> <Company>The contributors of MQTTnet</Company>
<Product>MQTTnet</Product> <Product>MQTTnet</Product>
<Description> <Description>
This is an extension library which provides mqtt topic templating logic to support dispatch, Provides mqtt topic templating logic to support dispatch,
routing and similar functionality based on the well known moustache syntax, aka routing and similar functionality based on the well known moustache syntax (AsyncAPI compatible).
AsyncAPI dynamic channel address.
</Description> </Description>
<PackageReadmeFile>README.md</PackageReadmeFile>
<Authors>The contributors of MQTTnet</Authors> <Authors>The contributors of MQTTnet</Authors>
<PackageId>MQTTnet.Extensions.TopicTemplate</PackageId> <PackageId>MQTTnet.Extensions.TopicTemplate</PackageId>
<SignAssembly>false</SignAssembly> <SignAssembly>false</SignAssembly>
@@ -51,6 +51,10 @@
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
<None Include="README.md">
<Pack>True</Pack>
<PackagePath>\</PackagePath>
</None>
<None Include="..\..\Images\nuget.png"> <None Include="..\..\Images\nuget.png">
<Pack>True</Pack> <Pack>True</Pack>
<PackagePath>\</PackagePath> <PackagePath>\</PackagePath>
@@ -16,8 +16,8 @@ namespace MQTTnet.Extensions.TopicTemplate
/// segments in curly braces called parameters. This well-known /// segments in curly braces called parameters. This well-known
/// 'moustache' syntax also matches AsyncAPI Channel Address Expressions. /// 'moustache' syntax also matches AsyncAPI Channel Address Expressions.
/// The topic template is designed to support dynamic subscription/publication, /// The topic template is designed to support dynamic subscription/publication,
/// message-topic matching and routing. It is intended to be more convenient /// message-topic matching and routing. It is intended to be more safe and
/// than String.Format() for aforementioned purposes. /// convenient than String.Format() for aforementioned purposes.
/// </summary> /// </summary>
/// <example> /// <example>
/// topic/subtopic/{parameter}/{otherParameter} /// topic/subtopic/{parameter}/{otherParameter}
@@ -282,12 +282,17 @@ namespace MQTTnet.Extensions.TopicTemplate
/// <returns>the topic template (without the parameter)</returns> /// <returns>the topic template (without the parameter)</returns>
public MqttTopicTemplate WithoutParameter(string parameter) 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> /// <summary>
/// Substitute a parameter with a given value, thus removing the parameter. If the parameter is not present, /// 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> /// </summary>
/// <param /// <param
/// name="parameter"> /// name="parameter">
@@ -304,12 +309,19 @@ namespace MQTTnet.Extensions.TopicTemplate
/// <returns>the topic template (without the parameter)</returns> /// <returns>the topic template (without the parameter)</returns>
public MqttTopicTemplate WithParameter(string parameter, string value) 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)) 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 + "}"; var moustache = "{" + parameter + "}";
return new MqttTopicTemplate(Template.Replace(moustache, value)); 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;
using System.Linq; using System.Linq;
using MQTTnet.Client;
using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
namespace MQTTnet.Extensions.TopicTemplate namespace MQTTnet.Extensions.TopicTemplate
@@ -55,6 +57,24 @@ namespace MQTTnet.Extensions.TopicTemplate
return new MqttTopicFilterBuilder().WithTopicTemplate(topicTemplate, subscribeTreeRoot); 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> /// <summary>
/// Return a message builder to respond to this message. The /// Return a message builder to respond to this message. The
/// message's response topic and correlation data are included /// 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); 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> /// <summary>
/// Set the publication topic according to the topic template. The template /// Set the publication topic according to the topic template. The template
/// must not have remaining (unset) parameters or contain wildcards. /// 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.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj" />
<ProjectReference Include="..\..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj" /> <ProjectReference Include="..\..\Source\MQTTnet.Extensions.WebSocket4Net\MQTTnet.Extensions.WebSocket4Net.csproj" />
<ProjectReference Include="..\..\Source\MQTTnet\MQTTnet.csproj" /> <ProjectReference Include="..\..\Source\MQTTnet\MQTTnet.csproj" />
<ProjectReference Include="..\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj" />
</ItemGroup> </ItemGroup>
</Project> </Project>
+18 -4
View File
@@ -1,6 +1,6 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Text; using MQTTnet.Extensions.TopicTemplate;
namespace MQTTnet.TestApp namespace MQTTnet.TestApp
{ {
@@ -17,6 +17,8 @@ namespace MQTTnet.TestApp
singleWildcardTopicsByPublisher = new Dictionary<string, List<string>>(); singleWildcardTopicsByPublisher = new Dictionary<string, List<string>>();
multiWildcardTopicsByPublisher = 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 // Find some reasonable distribution across three topic levels
var topicsPerLevel = (int)Math.Pow(numTopicsPerPublisher, (1.0 / 3.0)); var topicsPerLevel = (int)Math.Pow(numTopicsPerPublisher, (1.0 / 3.0));
@@ -40,25 +42,37 @@ namespace MQTTnet.TestApp
int publisherTopicCount = 0; int publisherTopicCount = 0;
var publisherName = "pub" + p; var publisherName = "pub" + p;
var publisherTemplate = baseTemplate
.WithParameter("publisher", publisherName);
for (var l1 = 0; l1 < numLevel1Topics; ++l1) for (var l1 = 0; l1 < numLevel1Topics; ++l1)
{ {
var l1Template = publisherTemplate
.WithParameter("building", "building" + (l1 + 1));
for (var l2 = 0; l2 < numLevel2Topics; ++l2) for (var l2 = 0; l2 < numLevel2Topics; ++l2)
{ {
var l2Template = l1Template
.WithParameter("level", "level" + (l2 + 1));
for (var l3 = 0; l3 < maxNumLevel3Topics; ++l3) for (var l3 = 0; l3 < maxNumLevel3Topics; ++l3)
{ {
if (publisherTopicCount >= numTopicsPerPublisher) if (publisherTopicCount >= numTopicsPerPublisher)
break; 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); AddPublisherTopic(publisherName, topic, topicsByPublisher);
if (l2 == 0) 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); AddPublisherTopic(publisherName, singleWildcardTopic, singleWildcardTopicsByPublisher);
if (l1 == 0) 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); AddPublisherTopic(publisherName, multiWildcardTopic, multiWildcardTopicsByPublisher);
} }
} }
@@ -46,6 +46,29 @@ namespace MQTTnet.Tests.Extensions
template.WithParameter("foo", "a#"); 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] [TestMethod]
[ExpectedException(typeof(MqttProtocolViolationException))] [ExpectedException(typeof(MqttProtocolViolationException))]
public void RejectsEmptyTemplate() public void RejectsEmptyTemplate()
@@ -60,13 +83,6 @@ namespace MQTTnet.Tests.Extensions
var _ = new MqttTopicTemplate(null); 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] [TestMethod]
public void IgnoresEmptyParameters() public void IgnoresEmptyParameters()
@@ -114,6 +130,50 @@ namespace MQTTnet.Tests.Extensions
Assert.AreEqual("A/v1/+/F", filter.Topic); 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] [TestMethod]
public void CanonicalPrefixFilter() public void CanonicalPrefixFilter()
{ {
+55 -49
View File
@@ -2,13 +2,13 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Implementations;
using System; using System;
using System.Net; using System.Net;
using System.Net.Sockets; using System.Net.Sockets;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Implementations;
namespace MQTTnet.Tests namespace MQTTnet.Tests
{ {
@@ -18,61 +18,67 @@ namespace MQTTnet.Tests
[TestMethod] [TestMethod]
public async Task Dispose_Channel_While_Used() public async Task Dispose_Channel_While_Used()
{ {
var ct = new CancellationTokenSource(); using (var ct = new CancellationTokenSource())
var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
try
{ {
serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001)); using (var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp))
serverSocket.Listen(0);
#pragma warning disable 4014
Task.Run(async () =>
#pragma warning restore 4014
{ {
while (!ct.IsCancellationRequested) try
{ {
var client = await serverSocket.AcceptAsync(); serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001));
var data = new byte[] { 128 }; serverSocket.Listen(0);
await client.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
}
}, ct.Token);
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 #pragma warning disable 4014
Task.Run(() => Task.Run(
async () =>
#pragma warning restore 4014 #pragma warning restore 4014
{ {
Task.Delay(200, ct.Token); while (!ct.IsCancellationRequested)
tcpChannel.Dispose(); {
}, ct.Token); var client = await serverSocket.AcceptAsync(ct.Token);
var data = new byte[] { 128 };
await client.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
}
},
ct.Token);
try var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
{ await clientSocket.ConnectAsync("localhost", 50001, CancellationToken.None);
await tcpChannel.ReadAsync(buffer, 0, 1, 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;
using System.Collections.Generic; using System.Collections.Generic;
using MQTTnet.Extensions.TopicTemplate;
namespace MQTTnet.Tests namespace MQTTnet.Tests
{ {
@@ -32,30 +33,42 @@ namespace MQTTnet.Tests
maxNumLevel3Topics = 1; maxNumLevel3Topics = 1;
} }
MqttTopicTemplate baseTemplate = new MqttTopicTemplate("{publisher}/{building}/{level}/{sensor}");
for (var p = 0; p < numPublishers; ++p) for (var p = 0; p < numPublishers; ++p)
{ {
int publisherTopicCount = 0; int publisherTopicCount = 0;
var publisherName = "pub" + p; var publisherName = "pub" + p;
var publisherTemplate = baseTemplate
.WithParameter("publisher", publisherName);
for (var l1 = 0; l1 < numLevel1Topics; ++l1) for (var l1 = 0; l1 < numLevel1Topics; ++l1)
{ {
var l1Template = publisherTemplate.WithParameter("building", "building" + (l1 + 1));
for (var l2 = 0; l2 < numLevel2Topics; ++l2) for (var l2 = 0; l2 < numLevel2Topics; ++l2)
{ {
var l2Template = l1Template.WithParameter("level", "level" + (l2 + 1));
for (var l3 = 0; l3 < maxNumLevel3Topics; ++l3) for (var l3 = 0; l3 < maxNumLevel3Topics; ++l3)
{ {
if (publisherTopicCount >= numTopicsPerPublisher) if (publisherTopicCount >= numTopicsPerPublisher)
break; break;
var topic = string.Format("{0}/building{1}/level{2}/sensor{3}", publisherName, l1 + 1, l2 + 1, l3 + 1); var l3Template = l2Template
AddPublisherTopic(publisherName, topic, topicsByPublisher); .WithParameter("sensor", "sensor" + (l3 + 1));
AddPublisherTopic(publisherName, l3Template.TopicFilter, topicsByPublisher);
if (l2 == 0) 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); AddPublisherTopic(publisherName, singleWildcardTopic, singleWildcardTopicsByPublisher);
} }
if ((l1 == 0) && (l3 == 0)) 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); AddPublisherTopic(publisherName, multiWildcardTopic, multiWildcardTopicsByPublisher);
} }
@@ -68,12 +68,12 @@ namespace MQTTnet.Formatter.V5
MqttPacket DecodeAuthPacket(ArraySegment<byte> body) MqttPacket DecodeAuthPacket(ArraySegment<byte> body)
{ {
ThrowIfBodyIsEmpty(body);
_bufferReader.SetBuffer(body.Array, body.Offset, body.Count); _bufferReader.SetBuffer(body.Array, body.Offset, body.Count);
var packet = new MqttAuthPacket(); 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) if (_bufferReader.EndOfStream)
{ {
packet.ReasonCode = MqttAuthenticateReasonCode.Success; packet.ReasonCode = MqttAuthenticateReasonCode.Success;
@@ -64,13 +64,20 @@ namespace MQTTnet.Formatter.V5
byte EncodeAuthPacket(MqttAuthPacket packet) byte EncodeAuthPacket(MqttAuthPacket packet)
{ {
_bufferWriter.WriteByte((byte)packet.ReasonCode);
_propertiesWriter.WriteAuthenticationMethod(packet.AuthenticationMethod); _propertiesWriter.WriteAuthenticationMethod(packet.AuthenticationMethod);
_propertiesWriter.WriteAuthenticationData(packet.AuthenticationData); _propertiesWriter.WriteAuthenticationData(packet.AuthenticationData);
_propertiesWriter.WriteReasonString(packet.ReasonString); _propertiesWriter.WriteReasonString(packet.ReasonString);
_propertiesWriter.WriteUserProperties(packet.UserProperties); _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.WriteTo(_bufferWriter);
_propertiesWriter.Reset(); _propertiesWriter.Reset();
@@ -142,12 +142,14 @@ namespace MQTTnet.Implementations
set => _socket.SendTimeout = value; set => _socket.SendTimeout = value;
} }
public async Task<CrossPlatformSocket> AcceptAsync() public async Task<CrossPlatformSocket> AcceptAsync(CancellationToken cancellationToken)
{ {
try try
{ {
#if NET452 || NET461 #if NET452 || NET461
var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false); 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 #else
var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false); var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
#endif #endif
@@ -141,7 +141,7 @@ namespace MQTTnet.Implementations
{ {
try try
{ {
var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false); var clientSocket = await _socket.AcceptAsync(cancellationToken).ConfigureAwait(false);
if (clientSocket == null) if (clientSocket == null)
{ {
continue; continue;