Compare commits

...
14 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
Christian 290419b3b6 [Client] Fix endpoint handling (#1999)
* Fix endpoint handling

* Fix different behavior for Mono framework.

* Restore old properties to avoid breaking changes
2024-05-19 12:18:12 +02:00
Christian 32b6146532 Setup nuget signing (#2001)
* Setup signing

* Update release notes
2024-05-18 20:21:09 +02:00
Christian 3a73ec21f7 [Client] Restore _WithConnectionUri_ methods (#1998)
Remove obsolete attributes
2024-05-17 17:49:33 +02:00
27 changed files with 595 additions and 169 deletions
+5 -13
View File
@@ -1,13 +1,5 @@
* [Core] Optimized packet serialization of PUBACK and PUBREC packets for protocol version 5.0.0 (#1939, thanks to @Y-Sindo). * Core: Fixed issue when parsing AUTH packet with 0 length body (#2039).
* [Core] The package inspector is now fully async (#1941). * nuget: Changed code signing and nuget certificate (**BREAKING CHANGE**).
* [Core] Fixed decoding of DISCONNECT packet with empty body (#1994, thanks to @Y-Sindo). * TopicTemplates: Updated samples, parameter validation (#2022).
* [Client] Exposed the _EndPoint_ type to support other endpoint types (like Unix Domain Sockets) in client options (#1919). * ManagedClient: Switch SubscribeAsync/UnsubscribeAsync to IEnumerable<string> (#2026).
* [Client] Fixed support for unix sockets by exposing more options (#1995). * Server: Fix _LoadingRetainedMessageAsync_ not executed (#2025).
* [Client] Added a dedicated exception when the client is not connected (#1954, thanks to @marcpiulachs).
* [Client] The client will now throw a _MqttClientUnexpectedDisconnectReceivedException_ when publishing a QoS 0 message which leads to a server disconnect (BREAKING CHANGE!, #1974, thanks to @fazho).
* [Client] Exposed the certificate selection event handler in client options (#1984).
* [Server] The server will no longer send _NoMatchingSubscribers_ when the actual subscription was non success (#1965, BREAKING CHANGE!).
* [Server] Fixed broken support for _null_ in _AddServer_ method in ASP.NET integration (#1981).
* [ManagedClient] Added a new event (SubscriptionsChangedAsync) which is fired when a subscription or unsubscription was made (#1894, thanks to @pdufrene).
* [ManagedClient] Fixed race condition when server shuts down while subscribing (#1987, thanks to @marve).
* [TopicTemplate] Added new extension which provides a template engine for topics (#1932, thanks to @simonthum).
+30 -12
View File
@@ -3,7 +3,7 @@ name: CI
on: [push, pull_request] on: [push, pull_request]
env: env:
VERSION: "4.3.4.${{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>
@@ -22,7 +22,7 @@ public sealed class Managed_Client_Subscribe_Samples
* - Reconnecting when connection is lost. * - Reconnecting when connection is lost.
* - Storing pending messages in an internal queue so that an enqueue is possible while the client remains not connected. * - Storing pending messages in an internal queue so that an enqueue is possible while the client remains not connected.
*/ */
var mqttFactory = new MqttFactory(); var mqttFactory = new MqttFactory();
var subscribed = false; var subscribed = false;
@@ -43,21 +43,21 @@ public sealed class Managed_Client_Subscribe_Samples
await managedMqttClient.EnqueueAsync("Topic", "Payload"); await managedMqttClient.EnqueueAsync("Topic", "Payload");
Console.WriteLine("The managed MQTT client is connected."); Console.WriteLine("The managed MQTT client is connected.");
// Wait until the queue is fully processed. // Wait until the queue is fully processed.
SpinWait.SpinUntil(() => managedMqttClient.PendingApplicationMessagesCount == 0, 10000); SpinWait.SpinUntil(() => managedMqttClient.PendingApplicationMessagesCount == 0, 10000);
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);
} }
} }
@@ -6,6 +6,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Diagnostics; using System.Diagnostics;
using System.Linq; using System.Linq;
using System.Net;
using System.Net.Sockets; using System.Net.Sockets;
using System.Text; using System.Text;
using System.Threading; using System.Threading;
@@ -86,7 +87,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
await client.ConnectAsync(clientOptions); await client.ConnectAsync(clientOptions);
} }
} }
[TestMethod] [TestMethod]
[ExpectedException(typeof(InvalidOperationException))] [ExpectedException(typeof(InvalidOperationException))]
public async Task Connect_Multiple_Times_Should_Fail() public async Task Connect_Multiple_Times_Should_Fail()
@@ -748,8 +749,8 @@ namespace MQTTnet.Tests.Clients.MqttClient
{ {
if (e.ApplicationMessage.Topic == "request") if (e.ApplicationMessage.Topic == "request")
{ {
// Use AtMostOnce here because with QoS 1 or even QoS 2 the process waits for // 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 // 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 // flag is set. It does not wait until the client has sent the ACK
await client2.PublishStringAsync("reply"); await client2.PublishStringAsync("reply");
} }
@@ -925,5 +926,54 @@ namespace MQTTnet.Tests.Clients.MqttClient
Assert.IsFalse(disconnectedFired); 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"); var template = new MqttTopicTemplate("A/B/{foo}/D");
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))]
@@ -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()
{ {
@@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System; using System;
using System.Net.Sockets;
using System.Text; using System.Text;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -17,7 +18,7 @@ namespace MQTTnet.Tests.Internal
[TestMethod] [TestMethod]
public async Task Connect_Send_Receive() public async Task Connect_Send_Receive()
{ {
var crossPlatformSocket = new CrossPlatformSocket(); var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
await crossPlatformSocket.ConnectAsync("www.google.de", 80, CancellationToken.None); await crossPlatformSocket.ConnectAsync("www.google.de", 80, CancellationToken.None);
var requestBuffer = Encoding.UTF8.GetBytes("GET / HTTP/1.1\r\nHost: www.google.de\r\n\r\n"); var requestBuffer = Encoding.UTF8.GetBytes("GET / HTTP/1.1\r\nHost: www.google.de\r\n\r\n");
@@ -36,7 +37,7 @@ namespace MQTTnet.Tests.Internal
[ExpectedException(typeof(OperationCanceledException))] [ExpectedException(typeof(OperationCanceledException))]
public async Task Try_Connect_Invalid_Host() public async Task Try_Connect_Invalid_Host()
{ {
var crossPlatformSocket = new CrossPlatformSocket(); var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
var cancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var cancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(5));
cancellationToken.Token.Register(() => crossPlatformSocket.Dispose()); cancellationToken.Token.Register(() => crossPlatformSocket.Dispose());
@@ -65,7 +66,7 @@ namespace MQTTnet.Tests.Internal
[TestMethod] [TestMethod]
public void Set_Options() public void Set_Options()
{ {
var crossPlatformSocket = new CrossPlatformSocket(); var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
Assert.IsFalse(crossPlatformSocket.ReuseAddress); Assert.IsFalse(crossPlatformSocket.ReuseAddress);
crossPlatformSocket.ReuseAddress = true; crossPlatformSocket.ReuseAddress = true;
+56 -50
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,62 +18,68 @@ 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
{ {
@@ -31,31 +32,43 @@ 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);
} }
@@ -79,11 +79,26 @@ namespace MQTTnet.Client
{ {
if (_port.HasValue) if (_port.HasValue)
{ {
_remoteEndPoint = new DnsEndPoint(dns.Host, _port.Value); _remoteEndPoint = new DnsEndPoint(dns.Host, _port.Value, dns.AddressFamily);
} }
else else
{ {
_remoteEndPoint = new DnsEndPoint(dns.Host, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure); _remoteEndPoint = new DnsEndPoint(dns.Host, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure, dns.AddressFamily);
}
}
}
if (_remoteEndPoint is IPEndPoint ip)
{
if (ip.Port == 0)
{
if (_port.HasValue)
{
_remoteEndPoint = new IPEndPoint(ip.Address, _port.Value);
}
else
{
_remoteEndPoint = new IPEndPoint(ip.Address, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure);
} }
} }
} }
@@ -152,7 +167,6 @@ namespace MQTTnet.Client
return this; return this;
} }
[Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
public MqttClientOptionsBuilder WithConnectionUri(Uri uri) public MqttClientOptionsBuilder WithConnectionUri(Uri uri)
{ {
if (uri == null) if (uri == null)
@@ -196,7 +210,6 @@ namespace MQTTnet.Client
return this; return this;
} }
[Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
public MqttClientOptionsBuilder WithConnectionUri(string uri) public MqttClientOptionsBuilder WithConnectionUri(string uri)
{ {
return WithConnectionUri(new Uri(uri, UriKind.Absolute)); return WithConnectionUri(new Uri(uri, UriKind.Absolute));
@@ -354,13 +367,18 @@ namespace MQTTnet.Client
return this; return this;
} }
public MqttClientOptionsBuilder WithTcpServer(string server, int? port = null) public MqttClientOptionsBuilder WithTcpServer(string host, int? port = null, AddressFamily addressFamily = AddressFamily.Unspecified)
{ {
if (host == null)
{
throw new ArgumentNullException(nameof(host));
}
_tcpOptions = new MqttClientTcpOptions(); _tcpOptions = new MqttClientTcpOptions();
// The value 0 will be updated when building the options. // The value 0 will be updated when building the options.
// This a backward compatibility feature. // This a backward compatibility feature.
_remoteEndPoint = new DnsEndPoint(server, port ?? 0); _remoteEndPoint = new DnsEndPoint(host, port ?? 0, addressFamily);
_port = port; _port = port;
return this; return this;
@@ -2,13 +2,16 @@
// 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 System;
using System.Net; using System.Net;
using System.Net.Sockets; using System.Net.Sockets;
using MQTTnet.Protocol;
namespace MQTTnet.Client namespace MQTTnet.Client
{ {
public sealed class MqttClientTcpOptions : IMqttClientChannelOptions public sealed class MqttClientTcpOptions : IMqttClientChannelOptions
{ {
EndPoint _remoteEndpoint;
public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified; public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
public int BufferSize { get; set; } = 8192; public int BufferSize { get; set; } = 8192;
@@ -20,6 +23,12 @@ namespace MQTTnet.Client
/// </summary> /// </summary>
public bool? DualMode { get; set; } public bool? DualMode { get; set; }
[Obsolete("Use RemoteEndpoint or MqttClientOptionsBuilder instead.")]
public int? Port { get; set; }
[Obsolete("Use RemoteEndpoint or MqttClientOptionsBuilder instead.")]
public string Server { get; set; }
public LingerOption LingerState { get; set; } = new LingerOption(true, 0); public LingerOption LingerState { get; set; } = new LingerOption(true, 0);
/// <summary> /// <summary>
@@ -42,13 +51,56 @@ namespace MQTTnet.Client
/// </summary> /// </summary>
public ProtocolType ProtocolType { get; set; } = ProtocolType.Tcp; 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 MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
public override string ToString() 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) 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;
@@ -13,7 +13,7 @@ namespace MQTTnet.Formatter.V5
public sealed class MqttV5PacketEncoder public sealed class MqttV5PacketEncoder
{ {
const int FixedHeaderSize = 1; const int FixedHeaderSize = 1;
readonly MqttBufferWriter _bufferWriter; readonly MqttBufferWriter _bufferWriter;
readonly MqttV5PropertiesWriter _propertiesWriter = new MqttV5PropertiesWriter(new MqttBufferWriter(1024, 4096)); readonly MqttV5PropertiesWriter _propertiesWriter = new MqttV5PropertiesWriter(new MqttBufferWriter(1024, 4096));
@@ -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();
@@ -526,7 +533,7 @@ namespace MQTTnet.Formatter.V5
static void ThrowIfPacketIdentifierIsInvalid(ushort packetIdentifier, MqttPacket packet) 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) if (packetIdentifier == 0)
{ {
@@ -31,11 +31,11 @@ namespace MQTTnet.Implementations
#endif #endif
} }
public CrossPlatformSocket() public CrossPlatformSocket(ProtocolType protocolType)
{ {
// Having this constructor is important because avoiding the address family as parameter // Having this constructor is important because avoiding the address family as parameter
// will make use of dual mode in the .net framework. // will make use of dual mode in the .net framework.
_socket = new Socket(SocketType.Stream, ProtocolType.Tcp); _socket = new Socket(SocketType.Stream, protocolType);
#if !NET5_0_OR_GREATER #if !NET5_0_OR_GREATER
_socketDisposeAction = _socket.Dispose; _socketDisposeAction = _socket.Dispose;
@@ -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
@@ -202,9 +204,28 @@ namespace MQTTnet.Implementations
using (cancellationToken.Register(_socketDisposeAction)) using (cancellationToken.Register(_socketDisposeAction))
{ {
#if NET452 || NET461 #if NET452 || NET461
await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, endPoint, null).ConfigureAwait(false); // This is a fix for Mono which behaves differently than dotnet.
// The connection will not be established when the DNS endpoint is used.
if (endPoint is DnsEndPoint dns && dns.AddressFamily == AddressFamily.Unspecified)
{
await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, dns.Host, dns.Port, null).ConfigureAwait(false);
}
else
{
await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, endPoint, null).ConfigureAwait(false);
}
#else #else
await _socket.ConnectAsync(endPoint).ConfigureAwait(false);
// This is a fix for Mono which behaves differently than dotnet.
// The connection will not be established when the DNS endpoint is used.
if (endPoint is DnsEndPoint dns && dns.AddressFamily == AddressFamily.Unspecified)
{
await _socket.ConnectAsync(dns.Host, dns.Port).ConfigureAwait(false);
}
else
{
await _socket.ConnectAsync(endPoint).ConfigureAwait(false);
}
#endif #endif
} }
#endif #endif
@@ -16,6 +16,7 @@ using MQTTnet.Channel;
using MQTTnet.Client; using MQTTnet.Client;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Protocol;
namespace MQTTnet.Implementations namespace MQTTnet.Implementations
{ {
@@ -63,7 +64,7 @@ namespace MQTTnet.Implementations
{ {
if (_tcpOptions.AddressFamily == AddressFamily.Unspecified) if (_tcpOptions.AddressFamily == AddressFamily.Unspecified)
{ {
socket = new CrossPlatformSocket(); socket = new CrossPlatformSocket(_tcpOptions.ProtocolType);
} }
else else
{ {
@@ -98,6 +99,29 @@ namespace MQTTnet.Implementations
socket.DualMode = _tcpOptions.DualMode.Value; socket.DualMode = _tcpOptions.DualMode.Value;
} }
// This block is only for backward compatibility.
if (_tcpOptions.RemoteEndpoint == null && !string.IsNullOrEmpty(_tcpOptions.Server))
{
int port;
if (_tcpOptions.Port.HasValue)
{
port = _tcpOptions.Port.Value;
}
else
{
if (_tcpOptions.TlsOptions?.UseTls == true)
{
port = MqttPorts.Secure;
}
else
{
port = MqttPorts.Default;
}
}
_tcpOptions.RemoteEndpoint = new DnsEndPoint(_tcpOptions.Server, port, AddressFamily.Unspecified);
}
await socket.ConnectAsync(_tcpOptions.RemoteEndpoint, cancellationToken).ConfigureAwait(false); await socket.ConnectAsync(_tcpOptions.RemoteEndpoint, cancellationToken).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested(); cancellationToken.ThrowIfCancellationRequested();
@@ -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;