Refactor topic filter naming.

This commit is contained in:
Christian Kratky
2020-04-18 01:38:20 +02:00
parent 2e69f8be78
commit 4b78710705
36 changed files with 154 additions and 143 deletions
@@ -5,7 +5,7 @@ namespace MQTTnet.Client.Subscribing
{ {
public class MqttClientSubscribeOptions public class MqttClientSubscribeOptions
{ {
public List<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>(); public List<MqttTopicFilter> TopicFilters { get; set; } = new List<MqttTopicFilter>();
public uint? SubscriptionIdentifier { get; set; } public uint? SubscriptionIdentifier { get; set; }
@@ -1,7 +1,7 @@
using System; using MQTTnet.Packets;
using System.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
namespace MQTTnet.Client.Subscribing namespace MQTTnet.Client.Subscribing
{ {
@@ -38,7 +38,7 @@ namespace MQTTnet.Client.Subscribing
bool? retainAsPublished = null, bool? retainAsPublished = null,
MqttRetainHandling? retainHandling = null) MqttRetainHandling? retainHandling = null)
{ {
return WithTopicFilter(new TopicFilter return WithTopicFilter(new MqttTopicFilter
{ {
Topic = topic, Topic = topic,
QualityOfServiceLevel = qualityOfServiceLevel, QualityOfServiceLevel = qualityOfServiceLevel,
@@ -48,30 +48,30 @@ namespace MQTTnet.Client.Subscribing
}); });
} }
public MqttClientSubscribeOptionsBuilder WithTopicFilter(Action<TopicFilterBuilder> topicFilterBuilder) public MqttClientSubscribeOptionsBuilder WithTopicFilter(Action<MqttTopicFilterBuilder> topicFilterBuilder)
{ {
if (topicFilterBuilder == null) throw new ArgumentNullException(nameof(topicFilterBuilder)); if (topicFilterBuilder == null) throw new ArgumentNullException(nameof(topicFilterBuilder));
var internalTopicFilterBuilder = new TopicFilterBuilder(); var internalTopicFilterBuilder = new MqttTopicFilterBuilder();
topicFilterBuilder(internalTopicFilterBuilder); topicFilterBuilder(internalTopicFilterBuilder);
return WithTopicFilter(internalTopicFilterBuilder); return WithTopicFilter(internalTopicFilterBuilder);
} }
public MqttClientSubscribeOptionsBuilder WithTopicFilter(TopicFilterBuilder topicFilterBuilder) public MqttClientSubscribeOptionsBuilder WithTopicFilter(MqttTopicFilterBuilder topicFilterBuilder)
{ {
if (topicFilterBuilder == null) throw new ArgumentNullException(nameof(topicFilterBuilder)); if (topicFilterBuilder == null) throw new ArgumentNullException(nameof(topicFilterBuilder));
return WithTopicFilter(topicFilterBuilder.Build()); return WithTopicFilter(topicFilterBuilder.Build());
} }
public MqttClientSubscribeOptionsBuilder WithTopicFilter(TopicFilter topicFilter) public MqttClientSubscribeOptionsBuilder WithTopicFilter(MqttTopicFilter topicFilter)
{ {
if (topicFilter == null) throw new ArgumentNullException(nameof(topicFilter)); if (topicFilter == null) throw new ArgumentNullException(nameof(topicFilter));
if (_subscribeOptions.TopicFilters == null) if (_subscribeOptions.TopicFilters == null)
{ {
_subscribeOptions.TopicFilters = new List<TopicFilter>(); _subscribeOptions.TopicFilters = new List<MqttTopicFilter>();
} }
_subscribeOptions.TopicFilters.Add(topicFilter); _subscribeOptions.TopicFilters.Add(topicFilter);
@@ -4,13 +4,13 @@ namespace MQTTnet.Client.Subscribing
{ {
public class MqttClientSubscribeResultItem public class MqttClientSubscribeResultItem
{ {
public MqttClientSubscribeResultItem(TopicFilter topicFilter, MqttClientSubscribeResultCode resultCode) public MqttClientSubscribeResultItem(MqttTopicFilter topicFilter, MqttClientSubscribeResultCode resultCode)
{ {
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter)); TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
ResultCode = resultCode; ResultCode = resultCode;
} }
public TopicFilter TopicFilter { get; } public MqttTopicFilter TopicFilter { get; }
public MqttClientSubscribeResultCode ResultCode { get; } public MqttClientSubscribeResultCode ResultCode { get; }
} }
@@ -44,7 +44,7 @@ namespace MQTTnet.Client.Unsubscribing
return this; return this;
} }
public MqttClientUnsubscribeOptionsBuilder WithTopicFilter(TopicFilter topicFilter) public MqttClientUnsubscribeOptionsBuilder WithTopicFilter(MqttTopicFilter topicFilter)
{ {
if (topicFilter is null) throw new ArgumentNullException(nameof(topicFilter)); if (topicFilter is null) throw new ArgumentNullException(nameof(topicFilter));
@@ -194,7 +194,7 @@ namespace MQTTnet.Formatter.V3
while (!body.EndOfStream) while (!body.EndOfStream)
{ {
var topicFilter = new TopicFilter var topicFilter = new MqttTopicFilter
{ {
Topic = body.ReadStringWithLengthPrefix(), Topic = body.ReadStringWithLengthPrefix(),
QualityOfServiceLevel = (MqttQualityOfServiceLevel)body.ReadByte() QualityOfServiceLevel = (MqttQualityOfServiceLevel)body.ReadByte()
@@ -405,7 +405,7 @@ namespace MQTTnet.Formatter.V5
var retainAsPublished = (options & (1 << 3)) > 0; var retainAsPublished = (options & (1 << 3)) > 0;
var retainHandling = (MqttRetainHandling)((options >> 4) & 3); var retainHandling = (MqttRetainHandling)((options >> 4) & 3);
packet.TopicFilters.Add(new TopicFilter packet.TopicFilters.Add(new MqttTopicFilter
{ {
Topic = topic, Topic = topic,
QualityOfServiceLevel = qos, QualityOfServiceLevel = qos,
@@ -1,9 +1,14 @@
using MQTTnet.Protocol; using MQTTnet.Protocol;
using System;
namespace MQTTnet namespace MQTTnet
{ {
// TODO: Consider renaming to "MqttTopicFilter" [Obsolete("Use MqttTopicFilter instead. It is just a renamed version to align with general namings in this lib.")]
public class TopicFilter public class TopicFilter : MqttTopicFilter
{
}
public class MqttTopicFilter
{ {
public string Topic { get; set; } public string Topic { get; set; }
+57
View File
@@ -0,0 +1,57 @@
using MQTTnet.Exceptions;
using MQTTnet.Protocol;
using System;
namespace MQTTnet
{
[Obsolete("Use MqttTopicFilter instead. It is just a renamed version to align with general namings in this lib.")]
public class TopicFilterBuilder : MqttTopicFilterBuilder
{
}
public class MqttTopicFilterBuilder
{
MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
string _topic;
public MqttTopicFilterBuilder WithTopic(string topic)
{
_topic = topic;
return this;
}
public MqttTopicFilterBuilder WithQualityOfServiceLevel(MqttQualityOfServiceLevel qualityOfServiceLevel)
{
_qualityOfServiceLevel = qualityOfServiceLevel;
return this;
}
public MqttTopicFilterBuilder WithAtLeastOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce;
return this;
}
public MqttTopicFilterBuilder WithAtMostOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
return this;
}
public MqttTopicFilterBuilder WithExactlyOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce;
return this;
}
public MqttTopicFilter Build()
{
if (string.IsNullOrEmpty(_topic))
{
throw new MqttProtocolViolationException("Topic is not set.");
}
return new MqttTopicFilter { Topic = _topic, QualityOfServiceLevel = _qualityOfServiceLevel };
}
}
}
@@ -7,7 +7,7 @@ namespace MQTTnet.Packets
{ {
public ushort? PacketIdentifier { get; set; } public ushort? PacketIdentifier { get; set; }
public List<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>(); public List<MqttTopicFilter> TopicFilters { get; set; } = new List<MqttTopicFilter>();
#region Added in MQTTv5 #region Added in MQTTv5
@@ -16,6 +16,6 @@ namespace MQTTnet.Server
Task<IList<MqttApplicationMessage>> GetMessagesAsync(); Task<IList<MqttApplicationMessage>> GetMessagesAsync();
Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<TopicFilter> topicFilters); Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<MqttTopicFilter> topicFilters);
} }
} }
+1 -1
View File
@@ -23,7 +23,7 @@ namespace MQTTnet.Server
Task<IList<MqttApplicationMessage>> GetRetainedApplicationMessagesAsync(); Task<IList<MqttApplicationMessage>> GetRetainedApplicationMessagesAsync();
Task ClearRetainedApplicationMessagesAsync(); Task ClearRetainedApplicationMessagesAsync();
Task SubscribeAsync(string clientId, ICollection<TopicFilter> topicFilters); Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters);
Task UnsubscribeAsync(string clientId, ICollection<string> topicFilters); Task UnsubscribeAsync(string clientId, ICollection<string> topicFilters);
Task StartAsync(IMqttServerOptions options); Task StartAsync(IMqttServerOptions options);
@@ -9,7 +9,7 @@ namespace MQTTnet.Server
IDictionary<object, object> Items { get; } IDictionary<object, object> Items { get; }
IList<TopicFilter> Subscriptions { get; } IList<MqttTopicFilter> Subscriptions { get; }
MqttApplicationMessage WillMessage { get; } MqttApplicationMessage WillMessage { get; }
@@ -282,7 +282,7 @@ namespace MQTTnet.Server
_cancellationToken.Cancel(false); _cancellationToken.Cancel(false);
} }
async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters) async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<MqttTopicFilter> topicFilters)
{ {
var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false); var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false);
foreach (var applicationMessage in retainedMessages) foreach (var applicationMessage in retainedMessages)
+1 -1
View File
@@ -53,7 +53,7 @@ namespace MQTTnet.Server
ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage); ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage);
} }
public async Task SubscribeAsync(ICollection<TopicFilter> topicFilters) public async Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters)
{ {
if (topicFilters is null) throw new ArgumentNullException(nameof(topicFilters)); if (topicFilters is null) throw new ArgumentNullException(nameof(topicFilters));
@@ -108,7 +108,7 @@ namespace MQTTnet.Server
_messageQueue.Enqueue(new MqttEnqueuedApplicationMessage(applicationMessage, sender)); _messageQueue.Enqueue(new MqttEnqueuedApplicationMessage(applicationMessage, sender));
} }
public Task SubscribeAsync(string clientId, ICollection<TopicFilter> topicFilters) public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
{ {
if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
@@ -9,7 +9,7 @@ namespace MQTTnet.Server
{ {
public class MqttClientSubscriptionsManager public class MqttClientSubscriptionsManager
{ {
private readonly Dictionary<string, TopicFilter> _subscriptions = new Dictionary<string, TopicFilter>(); private readonly Dictionary<string, MqttTopicFilter> _subscriptions = new Dictionary<string, MqttTopicFilter>();
private readonly MqttClientSession _clientSession; private readonly MqttClientSession _clientSession;
private readonly IMqttServerOptions _serverOptions; private readonly IMqttServerOptions _serverOptions;
private readonly MqttServerEventDispatcher _eventDispatcher; private readonly MqttServerEventDispatcher _eventDispatcher;
@@ -74,7 +74,7 @@ namespace MQTTnet.Server
return result; return result;
} }
public async Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters) public async Task SubscribeAsync(IEnumerable<MqttTopicFilter> topicFilters)
{ {
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
@@ -206,7 +206,7 @@ namespace MQTTnet.Server
} }
} }
private async Task<MqttSubscriptionInterceptorContext> InterceptSubscribeAsync(TopicFilter topicFilter) private async Task<MqttSubscriptionInterceptorContext> InterceptSubscribeAsync(MqttTopicFilter topicFilter)
{ {
var context = new MqttSubscriptionInterceptorContext(_clientSession.ClientId, topicFilter, _clientSession.Items); var context = new MqttSubscriptionInterceptorContext(_clientSession.ClientId, topicFilter, _clientSession.Items);
if (_serverOptions.SubscriptionInterceptor != null) if (_serverOptions.SubscriptionInterceptor != null)
@@ -105,7 +105,7 @@ namespace MQTTnet.Server
} }
} }
public async Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<TopicFilter> topicFilters) public async Task<IList<MqttApplicationMessage>> GetSubscribedMessagesAsync(ICollection<MqttTopicFilter> topicFilters)
{ {
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+1 -1
View File
@@ -87,7 +87,7 @@ namespace MQTTnet.Server
return _retainedMessagesManager.GetMessagesAsync(); return _retainedMessagesManager.GetMessagesAsync();
} }
public Task SubscribeAsync(string clientId, ICollection<TopicFilter> topicFilters) public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
{ {
if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
@@ -4,7 +4,7 @@ namespace MQTTnet.Server
{ {
public class MqttServerClientSubscribedTopicEventArgs : EventArgs public class MqttServerClientSubscribedTopicEventArgs : EventArgs
{ {
public MqttServerClientSubscribedTopicEventArgs(string clientId, TopicFilter topicFilter) public MqttServerClientSubscribedTopicEventArgs(string clientId, MqttTopicFilter topicFilter)
{ {
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter)); TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
@@ -12,6 +12,6 @@ namespace MQTTnet.Server
public string ClientId { get; } public string ClientId { get; }
public TopicFilter TopicFilter { get; } public MqttTopicFilter TopicFilter { get; }
} }
} }
@@ -60,7 +60,7 @@ namespace MQTTnet.Server
} }
} }
public async Task SafeNotifyClientSubscribedTopicAsync(string clientId, TopicFilter topicFilter) public async Task SafeNotifyClientSubscribedTopicAsync(string clientId, MqttTopicFilter topicFilter)
{ {
try try
{ {
@@ -1,10 +1,10 @@
using System; using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving;
using MQTTnet.Protocol;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -106,7 +106,7 @@ namespace MQTTnet.Server
return server; return server;
} }
public static Task SubscribeAsync(this IMqttServer server, string clientId, params TopicFilter[] topicFilters) public static Task SubscribeAsync(this IMqttServer server, string clientId, params MqttTopicFilter[] topicFilters)
{ {
if (server == null) throw new ArgumentNullException(nameof(server)); if (server == null) throw new ArgumentNullException(nameof(server));
if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (clientId == null) throw new ArgumentNullException(nameof(clientId));
@@ -121,7 +121,7 @@ namespace MQTTnet.Server
if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topic == null) throw new ArgumentNullException(nameof(topic)); if (topic == null) throw new ArgumentNullException(nameof(topic));
return server.SubscribeAsync(clientId, new TopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
} }
public static Task SubscribeAsync(this IMqttServer server, string clientId, string topic) public static Task SubscribeAsync(this IMqttServer server, string clientId, string topic)
@@ -130,7 +130,7 @@ namespace MQTTnet.Server
if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topic == null) throw new ArgumentNullException(nameof(topic)); if (topic == null) throw new ArgumentNullException(nameof(topic));
return server.SubscribeAsync(clientId, new TopicFilterBuilder().WithTopic(topic).Build()); return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).Build());
} }
public static Task UnsubscribeAsync(this IMqttServer server, string clientId, params string[] topicFilters) public static Task UnsubscribeAsync(this IMqttServer server, string clientId, params string[] topicFilters)
@@ -4,7 +4,7 @@ namespace MQTTnet.Server
{ {
public class MqttSubscriptionInterceptorContext public class MqttSubscriptionInterceptorContext
{ {
public MqttSubscriptionInterceptorContext(string clientId, TopicFilter topicFilter, IDictionary<object, object> sessionItems) public MqttSubscriptionInterceptorContext(string clientId, MqttTopicFilter topicFilter, IDictionary<object, object> sessionItems)
{ {
ClientId = clientId; ClientId = clientId;
TopicFilter = topicFilter; TopicFilter = topicFilter;
@@ -13,7 +13,7 @@ namespace MQTTnet.Server
public string ClientId { get; } public string ClientId { get; }
public TopicFilter TopicFilter { get; set; } public MqttTopicFilter TopicFilter { get; set; }
/// <summary> /// <summary>
/// Gets or sets a key/value collection that can be used to share data within the scope of this session. /// Gets or sets a key/value collection that can be used to share data within the scope of this session.
-51
View File
@@ -1,51 +0,0 @@
using MQTTnet.Exceptions;
using MQTTnet.Protocol;
namespace MQTTnet
{
public class TopicFilterBuilder
{
private MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
private string _topic;
public TopicFilterBuilder WithTopic(string topic)
{
_topic = topic;
return this;
}
public TopicFilterBuilder WithQualityOfServiceLevel(MqttQualityOfServiceLevel qualityOfServiceLevel)
{
_qualityOfServiceLevel = qualityOfServiceLevel;
return this;
}
public TopicFilterBuilder WithAtLeastOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce;
return this;
}
public TopicFilterBuilder WithAtMostOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
return this;
}
public TopicFilterBuilder WithExactlyOnceQoS()
{
_qualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce;
return this;
}
public TopicFilter Build()
{
if (string.IsNullOrEmpty(_topic))
{
throw new MqttProtocolViolationException("Topic is not set.");
}
return new TopicFilter { Topic = _topic, QualityOfServiceLevel = _qualityOfServiceLevel };
}
}
}
@@ -95,9 +95,9 @@ namespace MQTTnet.Tests
await client.SendAsync(new MqttSubscribePacket await client.SendAsync(new MqttSubscribePacket
{ {
PacketIdentifier = 1, PacketIdentifier = 1,
TopicFilters = new List<TopicFilter> TopicFilters = new List<MqttTopicFilter>
{ {
new TopicFilter new MqttTopicFilter
{ {
Topic = topic Topic = topic
} }
@@ -171,9 +171,9 @@ namespace MQTTnet.Tests.MQTTv5
var result = await client.SubscribeAsync(new MqttClientSubscribeOptions() var result = await client.SubscribeAsync(new MqttClientSubscribeOptions()
{ {
SubscriptionIdentifier = 1, SubscriptionIdentifier = 1,
TopicFilters = new List<TopicFilter> TopicFilters = new List<MqttTopicFilter>
{ {
new TopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce} new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce}
} }
}); });
@@ -34,9 +34,9 @@ namespace MQTTnet.Tests.MQTTv5
await receiver.SubscribeAsync(new MqttClientSubscribeOptions await receiver.SubscribeAsync(new MqttClientSubscribeOptions
{ {
TopicFilters = new List<TopicFilter> TopicFilters = new List<MqttTopicFilter>
{ {
new TopicFilter { Topic = "#" } new MqttTopicFilter { Topic = "#" }
} }
}, CancellationToken.None); }, CancellationToken.None);
@@ -1,10 +1,10 @@
using System.Threading; using Microsoft.VisualStudio.TestTools.UnitTesting;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client; using MQTTnet.Client;
using MQTTnet.Client.Options; using MQTTnet.Client.Options;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Tests.Mockups; using MQTTnet.Tests.Mockups;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.Tests.MQTTv5 namespace MQTTnet.Tests.MQTTv5
{ {
@@ -28,7 +28,7 @@ namespace MQTTnet.Tests.MQTTv5
var c1 = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500)); var c1 = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic("#").Build()); await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
var c2 = await testEnvironment.ConnectClientAsync(clientOptions); var c2 = await testEnvironment.ConnectClientAsync(clientOptions);
c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent. c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent.
@@ -123,7 +123,7 @@ namespace MQTTnet.Tests.Mockups
return Implementation.StopAsync(); return Implementation.StopAsync();
} }
public Task SubscribeAsync(string clientId, ICollection<TopicFilter> topicFilters) public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
{ {
return Implementation.SubscribeAsync(clientId, topicFilters); return Implementation.SubscribeAsync(clientId, topicFilters);
} }
+1 -1
View File
@@ -650,7 +650,7 @@ namespace MQTTnet.Tests
await receiver.SubscribeAsync(new MqttClientSubscribeOptions await receiver.SubscribeAsync(new MqttClientSubscribeOptions
{ {
TopicFilters = new List<TopicFilter> { new TopicFilter { Topic = "#" } } TopicFilters = new List<MqttTopicFilter> { new MqttTopicFilter { Topic = "#" } }
}, CancellationToken.None); }, CancellationToken.None);
MqttApplicationMessage receivedMessage = null; MqttApplicationMessage receivedMessage = null;
@@ -100,7 +100,7 @@ namespace MQTTnet.Tests
return Task.FromResult(0); return Task.FromResult(0);
} }
public Task SubscribeAsync(IList<TopicFilter> topicFilters) public Task SubscribeAsync(IList<MqttTopicFilter> topicFilters)
{ {
throw new NotSupportedException(); throw new NotSupportedException();
} }
@@ -449,9 +449,9 @@ namespace MQTTnet.Tests
PacketIdentifier = 123 PacketIdentifier = 123
}; };
p.TopicFilters.Add(new TopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce }); p.TopicFilters.Add(new MqttTopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce });
p.TopicFilters.Add(new TopicFilter { Topic = "1/2/3", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); p.TopicFilters.Add(new MqttTopicFilter { Topic = "1/2/3", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });
p.TopicFilters.Add(new TopicFilter { Topic = "x/y/z", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); p.TopicFilters.Add(new MqttTopicFilter { Topic = "x/y/z", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
SerializeAndCompare(p, "ghoAewAFQS9CL0MCAAUxLzIvMwEABXgveS96AA=="); SerializeAndCompare(p, "ghoAewAFQS9CL0MCAAUxLzIvMwEABXgveS96AA==");
} }
@@ -464,9 +464,9 @@ namespace MQTTnet.Tests
PacketIdentifier = 123 PacketIdentifier = 123
}; };
p.TopicFilters.Add(new TopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce }); p.TopicFilters.Add(new MqttTopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce });
p.TopicFilters.Add(new TopicFilter { Topic = "1/2/3", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); p.TopicFilters.Add(new MqttTopicFilter { Topic = "1/2/3", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });
p.TopicFilters.Add(new TopicFilter { Topic = "x/y/z", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); p.TopicFilters.Add(new MqttTopicFilter { Topic = "x/y/z", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
DeserializeAndCompare(p, "ghoAewAFQS9CL0MCAAUxLzIvMwEABXgveS96AA=="); DeserializeAndCompare(p, "ghoAewAFQS9CL0MCAAUxLzIvMwEABXgveS96AA==");
} }
@@ -19,7 +19,7 @@ namespace MQTTnet.Tests
var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions()); var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());
var sp = new MqttSubscribePacket(); var sp = new MqttSubscribePacket();
sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build()); sp.TopicFilters.Add(new MqttTopicFilterBuilder().WithTopic("A/B/C").Build());
await sm.SubscribeAsync(sp, new MqttConnectPacket()); await sm.SubscribeAsync(sp, new MqttConnectPacket());
@@ -36,7 +36,7 @@ namespace MQTTnet.Tests
var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions()); var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());
var sp = new MqttSubscribePacket(); var sp = new MqttSubscribePacket();
sp.TopicFilters.Add(new TopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); sp.TopicFilters.Add(new MqttTopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
await sm.SubscribeAsync(sp, new MqttConnectPacket()); await sm.SubscribeAsync(sp, new MqttConnectPacket());
@@ -53,8 +53,8 @@ namespace MQTTnet.Tests
var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions()); var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());
var sp = new MqttSubscribePacket(); var sp = new MqttSubscribePacket();
sp.TopicFilters.Add(new TopicFilter { Topic = "#", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); sp.TopicFilters.Add(new MqttTopicFilter { Topic = "#", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
sp.TopicFilters.Add(new TopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); sp.TopicFilters.Add(new MqttTopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });
await sm.SubscribeAsync(sp, new MqttConnectPacket()); await sm.SubscribeAsync(sp, new MqttConnectPacket());
@@ -71,7 +71,7 @@ namespace MQTTnet.Tests
var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions()); var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());
var sp = new MqttSubscribePacket(); var sp = new MqttSubscribePacket();
sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build()); sp.TopicFilters.Add(new MqttTopicFilterBuilder().WithTopic("A/B/C").Build());
await sm.SubscribeAsync(sp, new MqttConnectPacket()); await sm.SubscribeAsync(sp, new MqttConnectPacket());
@@ -86,7 +86,7 @@ namespace MQTTnet.Tests
var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions()); var sm = new MqttClientSubscriptionsManager(s, new MqttServerEventDispatcher(new TestLogger()), new MqttServerOptions());
var sp = new MqttSubscribePacket(); var sp = new MqttSubscribePacket();
sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build()); sp.TopicFilters.Add(new MqttTopicFilterBuilder().WithTopic("A/B/C").Build());
await sm.SubscribeAsync(sp, new MqttConnectPacket()); await sm.SubscribeAsync(sp, new MqttConnectPacket());
+13 -13
View File
@@ -111,7 +111,7 @@ namespace MQTTnet.Tests
var c1 = await testEnvironment.ConnectClientAsync(); var c1 = await testEnvironment.ConnectClientAsync();
c1.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(c => Interlocked.Increment(ref receivedMessagesCount)); c1.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(c => Interlocked.Increment(ref receivedMessagesCount));
await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic("#").Build()); await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
var c2 = await testEnvironment.ConnectClientAsync(clientOptions); var c2 = await testEnvironment.ConnectClientAsync(clientOptions);
await c2.DisconnectAsync().ConfigureAwait(false); await c2.DisconnectAsync().ConfigureAwait(false);
@@ -137,7 +137,7 @@ namespace MQTTnet.Tests
var c1 = await testEnvironment.ConnectClientAsync(); var c1 = await testEnvironment.ConnectClientAsync();
c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic("#").Build()); await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
var c2 = await testEnvironment.ConnectClientAsync(clientOptions); var c2 = await testEnvironment.ConnectClientAsync(clientOptions);
c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent. c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent.
@@ -213,7 +213,7 @@ namespace MQTTnet.Tests
subscribeEventCalled = e.TopicFilter.Topic == "a" && e.ClientId == c1.Options.ClientId; subscribeEventCalled = e.TopicFilter.Topic == "a" && e.ClientId == c1.Options.ClientId;
}); });
await c1.SubscribeAsync(new TopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); await c1.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });
await Task.Delay(250); await Task.Delay(250);
Assert.IsTrue(subscribeEventCalled, "Subscribe event not called."); Assert.IsTrue(subscribeEventCalled, "Subscribe event not called.");
@@ -326,7 +326,7 @@ namespace MQTTnet.Tests
client.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); client.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build();
await client.SubscribeAsync(new TopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); await client.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });
await server.PublishAsync(message); await server.PublishAsync(message);
@@ -365,8 +365,8 @@ namespace MQTTnet.Tests
} }
}); });
await c1.SubscribeAsync(new TopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); await c1.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });
await c2.SubscribeAsync(new TopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); await c2.SubscribeAsync(new MqttTopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });
var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build();
@@ -653,7 +653,7 @@ namespace MQTTnet.Tests
var c2 = await testEnvironment.ConnectClientAsync(); var c2 = await testEnvironment.ConnectClientAsync();
c2.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); c2.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
await c2.SubscribeAsync(new TopicFilterBuilder().WithTopic("retained_other").Build()); await c2.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("retained_other").Build());
await Task.Delay(500); await Task.Delay(500);
@@ -683,7 +683,7 @@ namespace MQTTnet.Tests
} }
}); });
await c2.SubscribeAsync(new TopicFilterBuilder().WithTopic("retained").Build()); await c2.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("retained").Build());
await Task.Delay(500); await Task.Delay(500);
@@ -713,7 +713,7 @@ namespace MQTTnet.Tests
c2.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); c2.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
await Task.Delay(200); await Task.Delay(200);
await c2.SubscribeAsync(new TopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
await Task.Delay(500); await Task.Delay(500);
Assert.AreEqual(0, receivedMessagesCount); Assert.AreEqual(0, receivedMessagesCount);
@@ -741,7 +741,7 @@ namespace MQTTnet.Tests
c2.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); c2.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
await Task.Delay(200); await Task.Delay(200);
await c2.SubscribeAsync(new TopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
await Task.Delay(500); await Task.Delay(500);
Assert.AreEqual(0, receivedMessagesCount); Assert.AreEqual(0, receivedMessagesCount);
@@ -830,7 +830,7 @@ namespace MQTTnet.Tests
var c1 = await testEnvironment.ConnectClientAsync(); var c1 = await testEnvironment.ConnectClientAsync();
var c2 = await testEnvironment.ConnectClientAsync(); var c2 = await testEnvironment.ConnectClientAsync();
await c2.SubscribeAsync(new TopicFilterBuilder().WithTopic("test").Build()); await c2.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("test").Build());
var isIntercepted = false; var isIntercepted = false;
c2.UseApplicationMessageReceivedHandler(c => c2.UseApplicationMessageReceivedHandler(c =>
@@ -1241,7 +1241,7 @@ namespace MQTTnet.Tests
} }
}); });
await client2.SubscribeAsync(new TopicFilter { Topic = "y" }, new TopicFilter { Topic = "n" }); await client2.SubscribeAsync(new MqttTopicFilter { Topic = "y" }, new MqttTopicFilter { Topic = "n" });
await Task.Delay(500); await Task.Delay(500);
@@ -1299,7 +1299,7 @@ namespace MQTTnet.Tests
var c1 = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithClientId("receiver")); var c1 = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithClientId("receiver"));
c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount)); c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic(topicFilter).WithQualityOfServiceLevel(filterQualityOfServiceLevel).Build()); await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(topicFilter).WithQualityOfServiceLevel(filterQualityOfServiceLevel).Build());
var c2 = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithClientId("sender")); var c2 = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithClientId("sender"));
await c2.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(new byte[0]).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); await c2.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(new byte[0]).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
+7 -7
View File
@@ -1,12 +1,12 @@
using System; using MQTTnet.Client;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Client;
using MQTTnet.Client.Connecting; using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options; using MQTTnet.Client.Options;
using MQTTnet.Client.Receiving; using MQTTnet.Client.Receiving;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using System;
using System.Text;
using System.Threading.Tasks;
namespace MQTTnet.TestApp.NetCore namespace MQTTnet.TestApp.NetCore
{ {
@@ -42,12 +42,12 @@ namespace MQTTnet.TestApp.NetCore
{ {
Console.WriteLine("### CONNECTED WITH SERVER ###"); Console.WriteLine("### CONNECTED WITH SERVER ###");
await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("#").Build()); await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
Console.WriteLine("### SUBSCRIBED ###"); Console.WriteLine("### SUBSCRIBED ###");
}); });
client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(async e => client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(async e =>
{ {
Console.WriteLine("### DISCONNECTED FROM SERVER ###"); Console.WriteLine("### DISCONNECTED FROM SERVER ###");
await Task.Delay(TimeSpan.FromSeconds(5)); await Task.Delay(TimeSpan.FromSeconds(5));
@@ -77,7 +77,7 @@ namespace MQTTnet.TestApp.NetCore
{ {
Console.ReadLine(); Console.ReadLine();
await client.SubscribeAsync(new TopicFilter { Topic = "test", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); await client.SubscribeAsync(new MqttTopicFilter { Topic = "test", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
var applicationMessage = new MqttApplicationMessageBuilder() var applicationMessage = new MqttApplicationMessageBuilder()
.WithTopic("A/B/C") .WithTopic("A/B/C")
@@ -45,8 +45,8 @@ namespace MQTTnet.TestApp.NetCore
await managedClient.PublishAsync(builder => builder.WithTopic("Step").WithPayload("1")); await managedClient.PublishAsync(builder => builder.WithTopic("Step").WithPayload("1"));
await managedClient.PublishAsync(builder => builder.WithTopic("Step").WithPayload("2").WithAtLeastOnceQoS()); await managedClient.PublishAsync(builder => builder.WithTopic("Step").WithPayload("2").WithAtLeastOnceQoS());
await managedClient.SubscribeAsync(new TopicFilter { Topic = "xyz", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); await managedClient.SubscribeAsync(new MqttTopicFilter { Topic = "xyz", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
await managedClient.SubscribeAsync(new TopicFilter { Topic = "abc", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); await managedClient.SubscribeAsync(new MqttTopicFilter { Topic = "abc", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
await managedClient.PublishAsync(builder => builder.WithTopic("Step").WithPayload("3")); await managedClient.PublishAsync(builder => builder.WithTopic("Step").WithPayload("3"));
@@ -334,7 +334,7 @@ namespace MQTTnet.TestApp.UniversalWindows
qos = MqttQualityOfServiceLevel.ExactlyOnce; qos = MqttQualityOfServiceLevel.ExactlyOnce;
} }
var topicFilter = new TopicFilter { Topic = SubscribeTopic.Text, QualityOfServiceLevel = qos }; var topicFilter = new MqttTopicFilter { Topic = SubscribeTopic.Text, QualityOfServiceLevel = qos };
if (_mqttClient != null) if (_mqttClient != null)
{ {
@@ -559,13 +559,13 @@ namespace MQTTnet.TestApp.UniversalWindows
Console.WriteLine("### CONNECTED WITH SERVER ###"); Console.WriteLine("### CONNECTED WITH SERVER ###");
// Subscribe to a topic // Subscribe to a topic
await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build()); await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("my/topic").Build());
Console.WriteLine("### SUBSCRIBED ###"); Console.WriteLine("### SUBSCRIBED ###");
}); });
// Subscribe to a topic // Subscribe to a topic
await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build()); await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("my/topic").Build());
// Unsubscribe from a topic // Unsubscribe from a topic
await client.UnsubscribeAsync("my/topic"); await client.UnsubscribeAsync("my/topic");
@@ -791,7 +791,7 @@ namespace MQTTnet.TestApp.UniversalWindows
.Build(); .Build();
var mqttClient = new MqttFactory().CreateManagedMqttClient(); var mqttClient = new MqttFactory().CreateManagedMqttClient();
await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic("my/topic").Build()); await mqttClient.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("my/topic").Build());
await mqttClient.StartAsync(options); await mqttClient.StartAsync(options);
} }