mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-19 19:13:11 +08:00
Compare commits
14
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
740d605fd2 | ||
|
|
9177b6ae68 | ||
|
|
f0ea5c9ecd | ||
|
|
76479ebabe | ||
|
|
e18a91a4b5 | ||
|
|
b95d352236 | ||
|
|
c5f4cf1f9b | ||
|
|
8650002523 | ||
|
|
25c1da6a21 | ||
|
|
4e3ede42c6 | ||
|
|
918a3c8d8b | ||
|
|
290419b3b6 | ||
|
|
32b6146532 | ||
|
|
3a73ec21f7 |
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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}"
|
||||||
|
|||||||
@@ -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);
|
||||||
|
|||||||
@@ -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)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -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>
|
||||||
|
|||||||
@@ -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;
|
||||||
|
|||||||
@@ -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();
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -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;
|
||||||
|
|||||||
Reference in New Issue
Block a user