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