Compare commits

...
Author SHA1 Message Date
Christian cea751ff9d Add more conversion methos for application message payloads 2022-11-26 14:14:40 +01:00
Christian e88b280eb0 Reset release notes 2022-11-12 12:53:54 +01:00
Christian e4b07f2b8c Fix build errors 2022-11-12 09:43:55 +01:00
Christian 4eb6e77eb5 Fix build errors due to merge conflicts 2022-11-11 14:41:28 +01:00
Christian db1fc7625d 1546 possible bug in server repeatedly sending the same message to client subscribed with cleansession=false (#1569) 2022-11-11 08:14:15 +01:00
Christian bfec9dc5ad 802 client isnt being notified if message is not delivered (#1568) 2022-11-09 21:14:53 +01:00
Christian 56f0a2c31c Fix validation of topic in managed client (#1567) 2022-11-09 20:13:59 +01:00
Christian 5ee0bb4b65 1565 high cpu load from asyncsignal (#1566) 2022-11-09 19:56:29 +01:00
Christian 5c74728387 Fix client session takeover 2022-11-06 13:28:41 +01:00
Christian 97c46bea0f Update ReleaseNotes.md 2022-11-06 10:50:32 +01:00
Christian e6c1ab4e4c Fail all waiting tasks when the client connection gets closed (#1563) 2022-11-06 10:50:04 +01:00
Christian 6e1ec266a7 Update ReleaseNotes.md 2022-11-06 10:49:19 +01:00
Christian 3fc6982e9a Fix upgrading and downgrading of retained messages QoS for subscriptions (#1564) 2022-11-06 10:35:59 +01:00
SilverFox 5b1cc7d713 Also cleanup MqttClientSessionsManager._subscriberSessions when session is taken over by a new one with CleanSession=true, fix #1552 (#1553) 2022-11-06 10:35:15 +01:00
Christian 2842f27522 1552 mqttnetservermqttclientsessionsmanager subscribersessions is not proper removed on special case (#1562) 2022-11-06 10:34:20 +01:00
Christian 7ef137e26e No longer fire _ClientAcknowledgedPublishPacketEvent_ for PUBREC packets (#1550) (#1551) 2022-11-06 10:29:49 +01:00
Christian b1a336eb62 Reset release notes, update version 2022-11-06 10:24:58 +01:00
Christian d481e93832 Add new samples. 2022-10-25 17:29:45 +02:00
70 changed files with 2117 additions and 1048 deletions
+1 -2
View File
@@ -1,2 +1 @@
* [Core] Fixed dead lock and race conditions in new _AsyncLock_ implementation (#1542).
* [Client] Fixed connection freeze when using Xamarin etc.
+1 -1
View File
@@ -3,7 +3,7 @@ name: CI
on: [push, pull_request]
env:
VERSION: "4.1.2.${{github.run_number}}"
VERSION: "4.1.3.${{github.run_number}}"
jobs:
build:
+1
View File
@@ -242,5 +242,6 @@ See the LICENSE file in the project root for more information.</s:String>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGREQ/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=PINGRESP/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=PUBCOMP/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=Tnet/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/UserDictionary/Words/=unsub/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
@@ -0,0 +1,47 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
// ReSharper disable UnusedType.Global
// ReSharper disable UnusedMember.Global
// ReSharper disable InconsistentNaming
// ReSharper disable EmptyConstructor
// ReSharper disable MemberCanBeMadeStatic.Local
using MQTTnet.Internal;
using MQTTnet.Server;
namespace MQTTnet.Samples.Server;
public static class Server_Intercepting_Samples
{
public static async Task Intercept_Application_Messages()
{
/*
* This sample starts a simple MQTT server which manipulate all processed application messages.
* Please see _Server_Simple_Samples_ for more details on how to start a server.
*/
var mqttFactory = new MqttFactory();
var mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().Build();
using (var mqttServer = mqttFactory.CreateMqttServer(mqttServerOptions))
{
mqttServer.InterceptingPublishAsync += args =>
{
// Here we only change the topic of the received application message.
// but also changing the payload etc. is required. Changing the QoS after
// transmitting is not supported and makes no sense at all.
args.ApplicationMessage.Topic += "/manipulated";
return CompletedTask.Instance;
};
await mqttServer.StartAsync();
Console.WriteLine("Press Enter to exit.");
Console.ReadLine();
await mqttServer.StopAsync();
}
}
}
@@ -1,26 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Collections.Generic;
using System.IO.Pipelines;
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Http.Features;
namespace MQTTnet.AspNetCore.Tests.Mockups
{
public class ConnectionContextMockup : ConnectionContext
{
public override string ConnectionId { get; set; }
public override IFeatureCollection Features { get; }
public override IDictionary<object, object> Items { get; set; }
public override IDuplexPipe Transport { get; set; }
public ConnectionContextMockup()
{
//Transport = new DefaultConnectionContext
}
}
}
@@ -2,20 +2,24 @@
// 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.AspNetCore.Connections;
using MQTTnet.Adapter;
using MQTTnet.Formatter;
using MQTTnet.Server;
using System;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Connections;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
using MQTTnet.Server;
namespace MQTTnet.AspNetCore.Tests.Mockups
{
public class ConnectionHandlerMockup: IMqttServerAdapter
public sealed class ConnectionHandlerMockup : IMqttServerAdapter
{
public TaskCompletionSource<MqttConnectionContext> Context { get; } = new TaskCompletionSource<MqttConnectionContext>();
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public TaskCompletionSource<MqttConnectionContext> Context { get; } = new TaskCompletionSource<MqttConnectionContext>();
public void Dispose()
{
}
public async Task OnConnectedAsync(ConnectionContext connection)
{
@@ -42,9 +46,5 @@ namespace MQTTnet.AspNetCore.Tests.Mockups
{
return Task.CompletedTask;
}
public void Dispose()
{
}
}
}
}
@@ -6,7 +6,7 @@ using System.IO.Pipelines;
namespace MQTTnet.AspNetCore.Tests.Mockups
{
public class DuplexPipeMockup : IDuplexPipe
public sealed class DuplexPipeMockup : IDuplexPipe
{
public DuplexPipeMockup()
{
@@ -20,9 +20,8 @@ namespace MQTTnet.AspNetCore.Tests.Mockups
PipeWriter IDuplexPipe.Output => Send.Writer;
public Pipe Receive { get; }
public Pipe Receive { get; set; }
public Pipe Send { get; set; }
public Pipe Send { get; }
}
}
@@ -6,7 +6,7 @@ using System.Buffers;
namespace MQTTnet.AspNetCore.Tests.Mockups
{
public class LimitedMemoryPool : MemoryPool<byte>
public sealed class LimitedMemoryPool : MemoryPool<byte>
{
protected override void Dispose(bool disposing)
{
@@ -7,9 +7,9 @@ using System.Buffers;
namespace MQTTnet.AspNetCore.Tests.Mockups
{
public class MemoryOwner : IMemoryOwner<byte>
public sealed class MemoryOwner : IMemoryOwner<byte>
{
private readonly byte[] _raw;
readonly byte[] _raw;
public MemoryOwner(int size)
{
@@ -27,11 +27,11 @@ namespace MQTTnet.AspNetCore.Tests.Mockups
Memory = _raw;
}
public Memory<byte> Memory { get; }
public void Dispose()
{
ArrayPool<byte>.Shared.Return(_raw);
}
public Memory<byte> Memory { get; }
}
}
@@ -2,22 +2,21 @@
// 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.AspNetCore.Builder;
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using System;
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Adapter;
using MQTTnet.AspNetCore.Tests.Mockups;
using MQTTnet.Client;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Packets;
using MQTTnet.Client;
using MQTTnet.Protocol;
using MQTTnet.Tests.Extensions;
@@ -26,20 +25,6 @@ namespace MQTTnet.AspNetCore.Tests
[TestClass]
public class MqttConnectionContextTest
{
[TestMethod]
public async Task TestReceivePacketAsyncThrowsWhenReaderCompleted()
{
var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311, new MqttBufferWriter(4096, 65535));
var pipe = new DuplexPipeMockup();
var connection = new DefaultConnectionContext();
connection.Transport = pipe;
var ctx = new MqttConnectionContext(serializer, connection);
pipe.Receive.Writer.Complete();
await Assert.ThrowsExceptionAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
}
[TestMethod]
public async Task TestCorruptedConnectPacket()
{
@@ -49,13 +34,43 @@ namespace MQTTnet.AspNetCore.Tests
var connection = new DefaultConnectionContext();
connection.Transport = pipe;
var ctx = new MqttConnectionContext(serializer, connection);
await pipe.Receive.Writer.WriteAsync(writer.AddMqttHeader(MqttControlPacketType.Connect, new byte[0]));
await Assert.ThrowsExceptionAsync<MqttProtocolViolationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
// the first exception should complete the pipes so if someone tries to use the connection after that it should throw immidiatly
await Assert.ThrowsExceptionAsync<InvalidOperationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
await Assert.ThrowsExceptionAsync<InvalidOperationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
}
[TestMethod]
public async Task TestEndpoint()
{
var mockup = new ConnectionHandlerMockup();
using (var host = new WebHostBuilder().UseKestrel(kestrel => kestrel.ListenLocalhost(1883, listener => listener.Use((ctx, next) => mockup.OnConnectedAsync(ctx))))
.UseStartup<Startup>()
.ConfigureServices(
(hostContext, services) =>
{
services.AddHostedMqttServer(o => o.WithoutDefaultEndpoint());
services.AddSingleton<IMqttServerAdapter>(mockup);
})
.Build())
using (var client = new MqttFactory().CreateMqttClient())
{
host.Start();
await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build(), CancellationToken.None);
var ctx = await mockup.Context.Task;
#if NETCOREAPP3_1
var ep = IPEndPoint.Parse(ctx.Endpoint);
Assert.IsNotNull(ep);
#endif
Assert.IsNotNull(ctx);
}
}
// COMMENTED OUT DUE TO DEAD LOCK? OR VERY VERY SLOW PERFORMANCE ON LOCAL DEV MACHINE. TEST WAS STILL RUNNING AFTER SEVERAL MINUTES!
@@ -78,7 +93,7 @@ namespace MQTTnet.AspNetCore.Tests
// await Task.WhenAll(tasks).ConfigureAwait(false);
//}
[TestMethod]
public async Task TestLargePacket()
{
@@ -94,43 +109,25 @@ namespace MQTTnet.AspNetCore.Tests
Assert.IsTrue(readResult.Buffer.Length > 20000);
}
private class Startup
[TestMethod]
public async Task TestReceivePacketAsyncThrowsWhenReaderCompleted()
{
var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311, new MqttBufferWriter(4096, 65535));
var pipe = new DuplexPipeMockup();
var connection = new DefaultConnectionContext();
connection.Transport = pipe;
var ctx = new MqttConnectionContext(serializer, connection);
pipe.Receive.Writer.Complete();
await Assert.ThrowsExceptionAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
}
class Startup
{
public void Configure(IApplicationBuilder app)
{
{
}
}
[TestMethod]
public async Task TestEndpoint()
{
var mockup = new ConnectionHandlerMockup();
using (var host = new WebHostBuilder()
.UseKestrel(kestrel => kestrel.ListenLocalhost(1883, listener => listener.Use((ctx, next) => mockup.OnConnectedAsync(ctx))))
.UseStartup<Startup>()
.ConfigureServices((hostContext, services) =>
{
services.AddHostedMqttServer(o => o.WithoutDefaultEndpoint());
services.AddSingleton<IMqttServerAdapter>(mockup);
})
.Build())
using (var client = new MqttFactory().CreateMqttClient())
{
host.Start();
await client.ConnectAsync(new MqttClientOptionsBuilder()
.WithTcpServer("localhost")
.Build(), CancellationToken.None);
var ctx = await mockup.Context.Task;
#if NETCOREAPP3_1
var ep = IPEndPoint.Parse(ctx.Endpoint);
Assert.IsNotNull(ep);
#endif
Assert.IsNotNull(ctx);
}
}
}
}
}
@@ -12,7 +12,7 @@ using MQTTnet.Packets;
namespace MQTTnet.AspNetCore.Tests
{
[TestClass]
public class ReaderExtensionsTest
public sealed class ReaderExtensionsTest
{
[TestMethod]
public void TestTryDeserialize()
@@ -43,7 +43,7 @@ namespace MQTTnet.Extensions.ManagedClient
Task StartAsync(ManagedMqttClientOptions options);
Task StopAsync();
Task StopAsync(bool cleanDisconnect = true);
Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters);
@@ -62,6 +62,7 @@ namespace MQTTnet.Extensions.ManagedClient
CancellationTokenSource _publishingCancellationToken;
ManagedMqttClientStorageManager _storageManager;
bool _isCleanDisconnect;
public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
{
@@ -160,7 +161,7 @@ namespace MQTTnet.Extensions.ManagedClient
throw new InvalidOperationException("call StartAsync before publishing messages");
}
MqttTopicValidator.ThrowIfInvalid(applicationMessage.ApplicationMessage.Topic);
MqttTopicValidator.ThrowIfInvalid(applicationMessage.ApplicationMessage);
ManagedMqttApplicationMessage removedMessage = null;
ApplicationMessageSkippedEventArgs applicationMessageSkippedEventArgs = null;
@@ -255,10 +256,12 @@ namespace MQTTnet.Extensions.ManagedClient
_logger.Info("Started");
}
public async Task StopAsync()
public async Task StopAsync(bool cleanDisconnect = true)
{
ThrowIfDisposed();
_isCleanDisconnect = cleanDisconnect;
StopPublishing();
StopMaintainingConnection();
@@ -382,9 +385,12 @@ namespace MQTTnet.Extensions.ManagedClient
{
try
{
using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout))
if (_isCleanDisconnect)
{
await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false);
using (var disconnectTimeout = new CancellationTokenSource(Options.ClientOptions.Timeout))
{
await InternalClient.DisconnectAsync(new MqttClientDisconnectOptions(), disconnectTimeout.Token).ConfigureAwait(false);
}
}
}
catch (OperationCanceledException)
@@ -64,7 +64,7 @@ namespace MQTTnet.Tests.Client
await client.ConnectAsync(options, CancellationToken.None).ConfigureAwait(false);
}
[TestMethod]
[ExpectedException(typeof(MqttCommunicationException))]
public async Task Connect_To_Wrong_Host()
@@ -80,6 +80,8 @@ namespace MQTTnet.Tests.Client
{
using (var testEnvironment = CreateTestEnvironment())
{
testEnvironment.IgnoreServerLogErrors = true;
testEnvironment.ServerPort = 8364;
var server = await testEnvironment.StartServer();
@@ -116,6 +118,8 @@ namespace MQTTnet.Tests.Client
{
using (var testEnvironment = CreateTestEnvironment())
{
testEnvironment.IgnoreServerLogErrors = true;
var server = await testEnvironment.StartServer();
using (var lowLevelClient = testEnvironment.CreateLowLevelClient())
@@ -123,7 +127,7 @@ namespace MQTTnet.Tests.Client
Assert.IsFalse(lowLevelClient.IsConnected);
var clientOptions = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).WithTimeout(TimeSpan.FromSeconds(1)).Build();
await lowLevelClient.ConnectAsync(clientOptions, CancellationToken.None);
Assert.IsTrue(lowLevelClient.IsConnected);
@@ -148,7 +152,7 @@ namespace MQTTnet.Tests.Client
catch
{
}
Assert.IsFalse(lowLevelClient.IsConnected);
}
}
@@ -11,6 +11,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Protocol;
using MQTTnet.Server;
@@ -21,6 +22,65 @@ namespace MQTTnet.Tests.Client
[TestClass]
public sealed class ManagedMqttClient_Tests : BaseTestClass
{
[TestMethod]
public async Task Receive_While_Not_Cleanly_Disconnected()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
await testEnvironment.StartServer(o => o.WithPersistentSessions());
var senderClient = await testEnvironment.ConnectClient();
// Prepare managed client.
var managedClient = testEnvironment.Factory.CreateManagedMqttClient();
await managedClient.SubscribeAsync("#");
var receivedMessages = testEnvironment.CreateApplicationMessageHandler(managedClient);
var managedClientOptions = new ManagedMqttClientOptions
{
ClientOptions = testEnvironment.Factory.CreateClientOptionsBuilder()
.WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
.WithClientId(nameof(Receive_While_Not_Cleanly_Disconnected) + "_managed")
.WithCleanSession(false)
.Build()
};
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
// Send test data.
await senderClient.PublishStringAsync("topic1");
await LongTestDelay();
receivedMessages.AssertReceivedCountEquals(1);
// Stop the managed client but keep session at server (not clean disconnect required).
await managedClient.StopAsync(false);
await LongTestDelay();
// Send new messages in the meantime.
await senderClient.PublishStringAsync("topic2", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
await LongTestDelay();
// Start the managed client, it should receive the new message.
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
receivedMessages.AssertReceivedCountEquals(2);
// Stop and start again, no new message should be received.
for (var i = 0; i < 3; i++)
{
await managedClient.StopAsync(false);
await LongTestDelay();
await managedClient.StartAsync(managedClientOptions);
await LongTestDelay();
}
receivedMessages.AssertReceivedCountEquals(2);
}
}
[TestMethod]
public async Task Connect_To_Invalid_Server()
{
@@ -33,7 +93,6 @@ namespace MQTTnet.Tests.Client
var managedClient = testEnvironment.Factory.CreateManagedMqttClient(mqttClient);
ConnectingFailedEventArgs connectingFailedEventArgs = null;
managedClient.ConnectingFailedAsync += args =>
{
connectingFailedEventArgs = args;
@@ -13,13 +13,12 @@ using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Tests.Extensions;
using MQTTnet.Tests.Mockups;
namespace MQTTnet.Tests
namespace MQTTnet.Tests.Formatter
{
[TestClass]
public sealed class MqttPacketSerialization_V3_Binary_Tests
@@ -10,7 +10,7 @@ using MQTTnet.Formatter;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Tests
namespace MQTTnet.Tests.Formatter
{
[TestClass]
public sealed class MqttPacketSerialization_V3_Tests
@@ -9,7 +9,7 @@ using MQTTnet.Formatter;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Tests
namespace MQTTnet.Tests.Formatter
{
[TestClass]
public sealed class MqttPacketSerialization_V5_Tests
@@ -0,0 +1,128 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Internal;
namespace MQTTnet.Tests.Internal
{
[TestClass]
public sealed class AsyncSignal_Tests
{
[TestMethod]
[ExpectedException(typeof(TaskCanceledException))]
public async Task Cancel_If_No_Signal()
{
var asyncSignal = new AsyncSignal();
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(2)))
{
await asyncSignal.WaitAsync(timeout.Token);
Assert.Fail("There is no signal. So we must fail here!");
}
}
[TestMethod]
[ExpectedException(typeof(ObjectDisposedException))]
public async Task Dispose_Properly()
{
var asyncSignal = new AsyncSignal();
// The timeout will not be reached but another task will kill the async signal.
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(999)))
{
_ = Task.Run(
async () =>
{
await Task.Delay(2000, CancellationToken.None);
asyncSignal.Dispose();
},
CancellationToken.None);
await asyncSignal.WaitAsync(timeout.Token);
Assert.Fail("There is no signal. So we must fail here!");
}
}
[TestMethod]
public async Task Reset_Signal()
{
var asyncSignal = new AsyncSignal();
// WaitAsync should fail because no signal is available.
for (var i = 0; i < 10; i++)
{
try
{
using (var timeout = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)))
{
await asyncSignal.WaitAsync(timeout.Token);
}
Assert.Fail("This must fail because the signal is not yet set.");
}
catch (OperationCanceledException)
{
}
asyncSignal.Set();
// WaitAsync should return directly because the signal is available.
await asyncSignal.WaitAsync();
}
}
[TestMethod]
public async Task Signal()
{
var asyncSignal = new AsyncSignal();
// The timeout will not be reached but another task will kill the async signal.
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(3)))
{
var stopwatch = Stopwatch.StartNew();
_ = Task.Run(
async () =>
{
await Task.Delay(1000, CancellationToken.None);
asyncSignal.Set();
},
CancellationToken.None);
await asyncSignal.WaitAsync(timeout.Token);
stopwatch.Stop();
Assert.IsTrue(stopwatch.ElapsedMilliseconds > 900);
}
}
[TestMethod]
[ExpectedException(typeof(InvalidOperationException))]
public async Task Fail_For_Two_Waiters()
{
var asyncSignal = new AsyncSignal();
// This thread will wait properly because it is the first waiter.
_ = Task.Run(
async () =>
{
await asyncSignal.WaitAsync();
},
CancellationToken.None);
await Task.Delay(1000);
// Now the current thread must fail because there is already a waiter.
await asyncSignal.WaitAsync();
}
}
}
@@ -0,0 +1,43 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Internal;
using MQTTnet.Packets;
namespace MQTTnet.Tests.Internal
{
[TestClass]
public sealed class MqttPacketBusItem_Tests
{
[TestMethod]
public void Fire_Completed_Event()
{
var eventFired = false;
var item = new MqttPacketBusItem(new MqttPublishPacket());
item.Completed += (_, __) =>
{
eventFired = true;
};
item.Complete();
Assert.IsTrue(eventFired);
}
[TestMethod]
[ExpectedException(typeof(TaskCanceledException))]
public async Task Wait_Packet_Bus_Item_After_Already_Canceled()
{
var item = new MqttPacketBusItem(new MqttPublishPacket());
// Finish the item before the actual
item.Cancel();
await item.WaitAsync();
}
}
}
@@ -13,69 +13,26 @@ namespace MQTTnet.Tests.Internal
{
[TestClass]
public sealed class MqttPacketBus_Tests
{
[TestMethod]
[ExpectedException(typeof(OperationCanceledException))]
public async Task Wait_With_Empty_Bus()
{
var bus = new MqttPacketBus();
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1)))
{
await bus.DequeueItemAsync(timeout.Token);
}
}
[TestMethod]
public async Task Fill_From_Different_Task()
{
const int MessageCount = 500;
var delayRandom = new Random();
var bus = new MqttPacketBus();
_ = Task.Run(
() =>
{
for (int i = 0; i < MessageCount; i++)
{
bus.EnqueueItem(new MqttPacketBusItem(MqttPingReqPacket.Instance), MqttPacketBusPartition.Health);
Thread.Sleep(delayRandom.Next(0, 10));
}
});
for (var i = 0; i < MessageCount; i++)
{
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)))
{
await bus.DequeueItemAsync(timeout.Token);
}
}
Assert.AreEqual(0, bus.TotalItemsCount);
}
{
[TestMethod]
public void Alternate_Priorities()
{
var bus = new MqttPacketBus();
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
bus.EnqueueItem(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
Assert.AreEqual(9, bus.TotalItemsCount);
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
@@ -85,33 +42,10 @@ namespace MQTTnet.Tests.Internal
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
Assert.AreEqual(0, bus.TotalItemsCount);
}
[TestMethod]
public void Export_Packets_Without_Dequeue()
{
var bus = new MqttPacketBus();
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
Assert.AreEqual(3, bus.TotalItemsCount);
var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control);
Assert.AreEqual(0, exportedPackets.Count);
exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Health);
Assert.AreEqual(0, exportedPackets.Count);
exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data);
Assert.AreEqual(3, exportedPackets.Count);
Assert.AreEqual(3, bus.TotalItemsCount);
}
[TestMethod]
public void Await_Single_Packet()
{
@@ -121,31 +55,97 @@ namespace MQTTnet.Tests.Internal
var item1 = new MqttPacketBusItem(new MqttPublishPacket());
var item2 = new MqttPacketBusItem(new MqttPublishPacket());
var item3 = new MqttPacketBusItem(new MqttPublishPacket());
item3.Delivered += (_, __) =>
item3.Completed += (_, __) =>
{
delivered = true;
};
bus.EnqueueItem(item1, MqttPacketBusPartition.Data);
bus.EnqueueItem(item2, MqttPacketBusPartition.Data);
bus.EnqueueItem(item3, MqttPacketBusPartition.Data);
Assert.IsFalse(delivered);
bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered();
bus.DequeueItemAsync(CancellationToken.None).Result.Complete();
Assert.IsFalse(delivered);
bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered();
bus.DequeueItemAsync(CancellationToken.None).Result.Complete();
Assert.IsFalse(delivered);
bus.DequeueItemAsync(CancellationToken.None).Result.MarkAsDelivered();
bus.DequeueItemAsync(CancellationToken.None).Result.Complete();
// The third packet has the event attached.
Assert.IsTrue(delivered);
}
[TestMethod]
public void Export_Packets_Without_Dequeue()
{
var bus = new MqttPacketBus();
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
Assert.AreEqual(3, bus.TotalItemsCount);
var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control);
Assert.AreEqual(0, exportedPackets.Count);
exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Health);
Assert.AreEqual(0, exportedPackets.Count);
exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data);
Assert.AreEqual(3, exportedPackets.Count);
Assert.AreEqual(3, bus.TotalItemsCount);
}
[TestMethod]
public async Task Fill_From_Different_Task()
{
const int MessageCount = 500;
var delayRandom = new Random();
var bus = new MqttPacketBus();
_ = Task.Run(
() =>
{
for (var i = 0; i < MessageCount; i++)
{
bus.EnqueueItem(new MqttPacketBusItem(MqttPingReqPacket.Instance), MqttPacketBusPartition.Health);
Thread.Sleep(delayRandom.Next(0, 10));
}
});
for (var i = 0; i < MessageCount; i++)
{
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)))
{
await bus.DequeueItemAsync(timeout.Token);
}
}
Assert.AreEqual(0, bus.TotalItemsCount);
}
[TestMethod]
[ExpectedException(typeof(TaskCanceledException))]
public async Task Wait_With_Empty_Bus()
{
var bus = new MqttPacketBus();
using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1)))
{
await bus.DequeueItemAsync(timeout.Token);
}
}
}
}
+2 -2
View File
@@ -154,7 +154,7 @@ namespace MQTTnet.Tests.MQTTv5
var result = await client.PublishStringAsync("a", "b", MqttQualityOfServiceLevel.AtLeastOnce);
await client.DisconnectAsync();
Assert.AreEqual(MqttClientPublishReasonCode.Success, result.ReasonCode);
Assert.AreEqual(MqttClientPublishReasonCode.NoMatchingSubscribers, result.ReasonCode);
}
}
@@ -169,7 +169,7 @@ namespace MQTTnet.Tests.MQTTv5
var result = await client.PublishStringAsync("a", "b", MqttQualityOfServiceLevel.ExactlyOnce);
await client.DisconnectAsync();
Assert.AreEqual(MqttClientPublishReasonCode.Success, result.ReasonCode);
Assert.AreEqual(MqttClientPublishReasonCode.NoMatchingSubscribers, result.ReasonCode);
}
}
@@ -2,8 +2,10 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
@@ -17,7 +19,12 @@ namespace MQTTnet.Tests.Mockups
public TestApplicationMessageReceivedHandler(IMqttClient mqttClient)
{
mqttClient.ApplicationMessageReceivedAsync += MqttClientOnApplicationMessageReceivedAsync;
if (mqttClient == null)
{
throw new ArgumentNullException(nameof(mqttClient));
}
mqttClient.ApplicationMessageReceivedAsync += OnApplicationMessageReceivedAsync;
}
public List<MqttApplicationMessageReceivedEventArgs> ReceivedEventArgs
@@ -39,7 +46,25 @@ namespace MQTTnet.Tests.Mockups
}
}
Task MqttClientOnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
public string GeneratePayloadSequence()
{
var sequence = new StringBuilder();
lock (_receivedEventArgs)
{
foreach (var receivedEventArg in _receivedEventArgs)
{
var payload = receivedEventArg.ApplicationMessage.ConvertPayloadToString();
// An empty payload is not part of the sequence!
sequence.Append(payload);
}
}
return sequence.ToString();
}
Task OnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
{
lock (_receivedEventArgs)
{
@@ -11,6 +11,7 @@ using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Extensions.Rpc;
using MQTTnet.Formatter;
using MQTTnet.Internal;
@@ -109,7 +110,7 @@ namespace MQTTnet.Tests.Mockups
configureOptions.Invoke(optionsBuilder);
var options = optionsBuilder.Build();
var client = CreateClient();
if (timeout == TimeSpan.Zero)
@@ -137,7 +138,7 @@ namespace MQTTnet.Tests.Mockups
options = options.WithTcpServer("127.0.0.1", ServerPort);
var client = CreateClient();
if (timeout == TimeSpan.Zero)
{
await client.ConnectAsync(options.Build()).ConfigureAwait(false);
@@ -161,7 +162,7 @@ namespace MQTTnet.Tests.Mockups
}
var client = CreateClient();
if (timeout == TimeSpan.Zero)
{
await client.ConnectAsync(options).ConfigureAwait(false);
@@ -199,6 +200,16 @@ namespace MQTTnet.Tests.Mockups
return new TestApplicationMessageReceivedHandler(mqttClient);
}
public TestApplicationMessageReceivedHandler CreateApplicationMessageHandler(IManagedMqttClient managedClient)
{
if (managedClient == null)
{
throw new ArgumentNullException(nameof(managedClient));
}
return new TestApplicationMessageReceivedHandler(managedClient.InternalClient);
}
public IMqttClient CreateClient()
{
lock (_clients)
@@ -325,7 +336,7 @@ namespace MQTTnet.Tests.Mockups
var options = optionsBuilder.Build();
var server = CreateServer(options);
await server.StartAsync();
// The OS has chosen the port to we have to properly expose it to the tests.
ServerPort = options.DefaultEndpointOptions.Port;
return server;
@@ -344,7 +355,7 @@ namespace MQTTnet.Tests.Mockups
var options = optionsBuilder.Build();
var server = CreateServer(options);
await server.StartAsync();
// The OS has chosen the port to we have to properly expose it to the tests.
ServerPort = options.DefaultEndpointOptions.Port;
return server;
+1 -1
View File
@@ -55,7 +55,7 @@ namespace MQTTnet.Tests.Server
});
}
SpinWait.SpinUntil(() => receivedMessages == 100000, TimeSpan.FromSeconds(60));
SpinWait.SpinUntil(() => receivedMessages == 100000, TimeSpan.FromSeconds(120));
Assert.AreEqual(100000, receivedMessages);
}
@@ -130,9 +130,7 @@ namespace MQTTnet.Tests.Server
CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel applicationMessageQoSLevel, string senderClientId)
{
ulong topicHashMask; // not needed
bool hasWildcard; // not needed
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out topicHashMask, out hasWildcard);
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out _, out _);
return _subscriptionsManager.CheckSubscriptions(topic, topicHash, applicationMessageQoSLevel, senderClientId);
}
}
@@ -0,0 +1,58 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Formatter;
using MQTTnet.Protocol;
namespace MQTTnet.Tests.Server
{
[TestClass]
public sealed class Publish_Tests : BaseTestClass
{
[TestMethod]
public async Task Return_Success_When_Subscribed()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
await testEnvironment.StartServer();
var sender = await testEnvironment.ConnectClient();
var receiver = await testEnvironment.ConnectClient();
await receiver.SubscribeAsync("A");
// AtLeastOnce is required to get an ACK packet from the server.
var publishResult = await sender.PublishStringAsync("A", "Payload", MqttQualityOfServiceLevel.AtLeastOnce);
Assert.AreEqual(MqttClientPublishReasonCode.Success, publishResult.ReasonCode);
Assert.AreEqual(true, publishResult.IsSuccess);
}
}
[TestMethod]
public async Task Return_NoMatchingSubscribers_When_Not_Subscribed()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
await testEnvironment.StartServer();
var sender = await testEnvironment.ConnectClient();
var receiver = await testEnvironment.ConnectClient();
await receiver.SubscribeAsync("A");
// AtLeastOnce is required to get an ACK packet from the server.
var publishResult = await sender.PublishStringAsync("B", "Payload", MqttQualityOfServiceLevel.AtLeastOnce);
Assert.AreEqual(MqttClientPublishReasonCode.NoMatchingSubscribers, publishResult.ReasonCode);
Assert.AreEqual(true, publishResult.IsSuccess);
}
}
}
}
@@ -0,0 +1,31 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Protocol;
namespace MQTTnet.Tests.Server
{
[TestClass]
public sealed class Publishing_Tests : BaseTestClass
{
[TestMethod]
[ExpectedException(typeof(MqttClientDisconnectedException))]
public async Task Disconnect_While_Publishing()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
// The client will be disconnect directly after subscribing!
server.InterceptingPublishAsync += ev => server.DisconnectClientAsync(ev.ClientId, MqttDisconnectReasonCode.NormalDisconnection);
var client = await testEnvironment.ConnectClient();
await client.PublishStringAsync("test", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
}
}
}
}
+53 -11
View File
@@ -15,6 +15,57 @@ namespace MQTTnet.Tests.Server
[TestClass]
public sealed class QoS_Tests : BaseTestClass
{
[TestMethod]
public async Task Preserve_Message_Order_For_Queued_Messages()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer(o => o.WithPersistentSessions());
// Create a session which will contain the messages.
var dummyClient = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false));
await dummyClient.SubscribeAsync("#", MqttQualityOfServiceLevel.AtLeastOnce);
dummyClient.Dispose();
// Now inject messages which are appended to the queue of the client.
await server.InjectApplicationMessage("T", "0", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "2", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "1", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "4", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "3", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "6", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "5", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "8", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "7", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await server.InjectApplicationMessage("T", "9", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
await LongTestDelay();
// Create a new client for the existing message.
var client = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false));
var messages = testEnvironment.CreateApplicationMessageHandler(client);
await LongTestDelay();
var payloadSequence = messages.GeneratePayloadSequence();
Assert.AreEqual("0214365879", payloadSequence);
// Disconnect and reconnect to make sure that the server will not send the messages twice.
await client.DisconnectAsync();
await LongTestDelay();
await client.ReconnectAsync();
await LongTestDelay();
payloadSequence = messages.GeneratePayloadSequence();
Assert.AreEqual("0214365879", payloadSequence);
}
}
[TestMethod]
public async Task Fire_Event_On_Client_Acknowledges_QoS_0()
{
@@ -99,25 +150,16 @@ namespace MQTTnet.Tests.Server
await LongTestDelay();
Assert.AreEqual(2, eventArgs.Count);
Assert.AreEqual(1, eventArgs.Count);
var firstEvent = eventArgs[0];
Assert.IsNotNull(firstEvent);
Assert.IsNotNull(firstEvent.PublishPacket);
Assert.IsNotNull(firstEvent.AcknowledgePacket);
Assert.IsFalse(firstEvent.IsCompleted);
Assert.IsTrue(firstEvent.IsCompleted);
Assert.AreEqual("A", firstEvent.PublishPacket.Topic);
var secondEvent = eventArgs[1];
Assert.IsNotNull(secondEvent);
Assert.IsNotNull(secondEvent.PublishPacket);
Assert.IsNotNull(secondEvent.AcknowledgePacket);
Assert.IsTrue(secondEvent.IsCompleted);
Assert.AreEqual("A", secondEvent.PublishPacket.Topic);
}
}
}
@@ -65,6 +65,66 @@ namespace MQTTnet.Tests.Server
}
}
[TestMethod]
public async Task Downgrade_QoS_Level()
{
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
var c1 = await testEnvironment.ConnectClient();
// Add the retained message with QoS 2!
await c1.PublishAsync(
new MqttApplicationMessageBuilder().WithTopic("retained")
.WithPayload(new byte[3])
.WithRetainFlag()
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce)
.Build());
// The second client uses QoS 1 so a downgrade is required.
var c2 = await testEnvironment.ConnectClient();
var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });
await LongTestDelay();
messageHandler.AssertReceivedCountEquals(1);
Assert.AreEqual(MqttQualityOfServiceLevel.AtLeastOnce, messageHandler.ReceivedEventArgs.First().ApplicationMessage.QualityOfServiceLevel);
}
}
[TestMethod]
public async Task No_Upgrade_QoS_Level()
{
using (var testEnvironment = CreateTestEnvironment())
{
await testEnvironment.StartServer();
var c1 = await testEnvironment.ConnectClient();
// Add the retained message with QoS 1!
await c1.PublishAsync(
new MqttApplicationMessageBuilder().WithTopic("retained")
.WithPayload(new byte[3])
.WithRetainFlag()
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
.Build());
// The second client uses QoS 2 so an upgrade is expected but according to the MQTT spec this is not supported!
var c2 = await testEnvironment.ConnectClient();
var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce });
await LongTestDelay();
messageHandler.AssertReceivedCountEquals(1);
Assert.AreEqual(MqttQualityOfServiceLevel.AtLeastOnce, messageHandler.ReceivedEventArgs.First().ApplicationMessage.QualityOfServiceLevel);
}
}
[TestMethod]
public async Task Receive_No_Retained_Message_After_Subscribe()
{
@@ -124,7 +184,7 @@ namespace MQTTnet.Tests.Server
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
.WithRetainFlag()
.Build());
await c1.DisconnectAsync();
// Subscribe using a new client.
+2 -2
View File
@@ -150,7 +150,7 @@ namespace MQTTnet.Tests.Server
// Try to connect 50 clients at the same time.
var clients = await Task.WhenAll(Enumerable.Range(0, 50).Select(i => ConnectAndSubscribe(testEnvironment, options, OnReceive)));
await Task.Delay(5000);
await Task.Delay(TimeSpan.FromSeconds(10));
var connectedClients = clients.Where(c => c?.TryPingAsync().GetAwaiter().GetResult() == true).ToList();
@@ -162,7 +162,7 @@ namespace MQTTnet.Tests.Server
var sendClient = await testEnvironment.ConnectClient(option2);
await sendClient.PublishStringAsync("aaa", "1");
await Task.Delay(3000);
await Task.Delay(TimeSpan.FromSeconds(5));
Assert.AreEqual(true, hasReceive);
}
@@ -46,6 +46,22 @@ namespace MQTTnet.Tests.Server
}
}
[TestMethod]
[ExpectedException(typeof(MqttClientDisconnectedException))]
public async Task Disconnect_While_Subscribing()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
// The client will be disconnect directly after subscribing!
server.ClientSubscribedTopicAsync += ev => server.DisconnectClientAsync(ev.ClientId, MqttDisconnectReasonCode.NormalDisconnection);
var client = await testEnvironment.ConnectClient();
await client.SubscribeAsync("#");
}
}
[TestMethod]
public async Task Enqueue_Message_After_Subscription()
{
@@ -1,11 +1,11 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
using MQTTnet.Tests.Mockups;
@@ -13,49 +13,301 @@ using MQTTnet.Tests.Mockups;
namespace MQTTnet.Tests.Server
{
[TestClass]
public class Subscription_TopicHash_Tests
public sealed class SubscriptionTopicHashTests
{
enum TopicHashSelector { NoWildcard, SingleWildcard, MultiWildcard };
MqttSession _clientSession;
/// <summary>
/// Even fairly regularly named topics as generated by the topic generator should result in shallow hash buckets
/// </summary>
/// <returns></returns>
[TestMethod]
public void Check_Hash_Bucket_Depth()
{
const int numPublishers = 5000;
const int numTopicsPerPublisher = 10;
TopicGenerator.Generate(
numPublishers,
numTopicsPerPublisher,
out var topicsByPublisher,
out _,
out _);
// There will be many 'similar' topics ending with, i.e. "sensor100", "sensor101", ...
// Hash bucket depths should remain low.
var bucketDepths = new Dictionary<ulong, int>();
var maxBucketDepth = 0;
ulong maxBucketDepthHash = 0;
var topicsByHash = new Dictionary<ulong, List<string>>();
foreach (var t in topicsByPublisher)
{
var topics = t.Value;
foreach (var topic in topics)
{
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var hashMask, out var hasWildcard);
bucketDepths.TryGetValue(topicHash, out var currentValue);
++currentValue;
bucketDepths[topicHash] = currentValue;
if (currentValue > maxBucketDepth)
{
maxBucketDepth = currentValue;
maxBucketDepthHash = topicHash;
}
if (!topicsByHash.TryGetValue(topicHash, out var topicList))
{
topicList = new List<string>();
topicsByHash.Add(topicHash, topicList);
}
topicList.Add(topic);
}
}
var maxDepthTopics = topicsByHash[maxBucketDepthHash];
Console.Write("Max bucket depth is " + maxBucketDepth);
// for the test case the bucket depth should be less than 100
Assert.IsTrue(maxBucketDepth < 100, "Unexpected high topic hash bucket depth");
}
MQTTnet.Server.MqttSession _clientSession;
[TestMethod]
public void Match_Hash_Test_SingleWildCard()
public void Check_Selected_Topic_Hashes()
{
var l0 = "pub0";
var l1 = "topic1";
var l2 = "+";
var l3 = "prop1";
var topic = string.Format("{0}/{1}/{2}/{3}", l0, l1, l2, l3);
CheckTopicHash("client1/building1/level1/sensor1", 0x655D4AF100000000, 0xFFFFFFFFFFFFFFFF);
CheckTopicHash("client1/building1/+/sensor1", 0x655D00F100000000, 0xFFFF00FFFFFFFFFF);
CheckTopicHash("client1/+/level1/+", 0x65004A0000000000, 0xFF00FF00FFFFFFFF);
CheckTopicHash("client1/building1/level1/#", 0x655D4A0000000000, 0xFFFFFF0000000000);
CheckTopicHash("client1/+/level1/#", 0x65004A0000000000, 0xFF00FF0000000000);
}
/// <summary>
/// Test long topic name with last level being #
/// </summary>
[TestMethod]
public void Match_Hash_Test_LongTopic_DetectMultiLevelWildcard()
{
var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/#";
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
}
UInt64 topicHash;
UInt64 topicHashMask;
bool topicHasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
/// <summary>
/// Test long topic name with last level being +
/// </summary>
[TestMethod]
public void Match_Hash_Test_LongTopic_DetectSingleLevelWildcard()
{
var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/+";
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
}
/// <summary>
/// Test long topic name with last level being #
/// </summary>
[TestMethod]
public void Match_Hash_Test_LongTopic_MultiWildcard()
{
var sb = new StringBuilder();
const int numLevels = 8;
var levelNames = new string[numLevels];
for (var i = 0; i < numLevels; ++i)
{
if (i > 0)
{
sb.Append("/");
}
string levelName;
if (i == numLevels - 1)
{
// last one is #
levelName = "#";
}
else
{
levelName = "level" + i;
}
levelNames[i] = levelName;
sb.Append(levelName);
}
var topic = sb.ToString();
// UInt64 is limited to 8 levels
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
var hashBytes = GetBytes(topicHash);
Assert.AreNotEqual(hashBytes[0], 0, "checksum 0 mismatch");
Assert.AreNotEqual(hashBytes[1], 0, "checksum 1 mismatch");
Assert.AreEqual(hashBytes[2], 0, "checksum 2 mismatch");
Assert.AreNotEqual(hashBytes[3], 0, "checksum 3 mismatch");
Assert.AreEqual(hashBytes[4], 0, "checksum 4 mismatch");
Assert.AreEqual(hashBytes[5], 0, "checksum 5 mismatch");
Assert.AreEqual(hashBytes[6], 0, "checksum 6 mismatch");
Assert.AreEqual(hashBytes[7], 0, "checksum 7 mismatch");
// all bytes should contain checksum
var count = 0;
foreach (var h in hashBytes)
{
if (count < 7)
{
Assert.AreNotEqual(h, 0, "checksum mismatch");
}
else
{
Assert.AreEqual(h, 0, "checksum mismatch");
}
// The mask should have zeroes where the wildcard and ff at the end
++count;
}
// The mask should have ff except for last level
var hashMaskBytes = GetBytes(topicHashMask);
Assert.AreEqual(hashMaskBytes[0], 0xff, "mask 0 mismatch");
Assert.AreEqual(hashMaskBytes[1], 0xff, "mask 1 mismatch");
Assert.AreEqual(hashMaskBytes[2], 0, "mask 2 mismatch");
Assert.AreEqual(hashMaskBytes[3], 0xff, "mask 3 mismatch");
Assert.AreEqual(hashMaskBytes[4], 0xff, "mask 4 mismatch");
Assert.AreEqual(hashMaskBytes[5], 0xff, "mask 5 mismatch");
Assert.AreEqual(hashMaskBytes[6], 0xff, "mask 6 mismatch");
Assert.AreEqual(hashMaskBytes[7], 0xff, "mask 7 mismatch");
count = 0;
foreach (var h in hashMaskBytes)
{
if (count < 7)
{
Assert.AreEqual(h, 0xff, "mask mismatch");
}
else
{
Assert.AreEqual(h, 0, "last mask mismatch");
}
++count;
}
}
/// <summary>
/// Test long topic name exceeding 8 levels
/// </summary>
[TestMethod]
public void Match_Hash_Test_LongTopic_NoWildCard()
{
var sb = new StringBuilder();
const int numLevels = 9;
var levelNames = new string[numLevels];
for (var i = 0; i < numLevels; ++i)
{
if (i > 0)
{
sb.Append("/");
}
var levelName = "level" + i;
levelNames[i] = levelName;
sb.Append(levelName);
}
var topic = sb.ToString();
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsFalse(topicHasWildcard, "Wildcard detected when not present");
var hashBytes = GetBytes(topicHash);
// all bytes should contain checksum
var count = 0;
foreach (var h in hashBytes)
{
Assert.AreNotEqual(h, 0, "checksum mismatch");
++count;
}
// The mask should have ff
var hashMaskBytes = GetBytes(topicHashMask);
count = 0;
foreach (var h in hashMaskBytes)
{
Assert.AreEqual(h, 0xff, "mask mismatch");
++count;
}
}
/// <summary>
/// Test long topic name with last level being +
/// </summary>
[TestMethod]
public void Match_Hash_Test_LongTopic_SingleWildCard()
{
var sb = new StringBuilder();
const int numLevels = 8;
var levelNames = new string[numLevels];
for (var i = 0; i < numLevels; ++i)
{
if (i > 0)
{
sb.Append("/");
}
string levelName;
if (i == numLevels - 1)
{
// last one is +
levelName = "+";
}
else
{
levelName = "level" + i;
}
levelNames[i] = levelName;
sb.Append(levelName);
}
var topic = sb.ToString();
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
var hashBytes = GetBytes(topicHash);
// all bytes should contain checksum
var count = 0;
foreach (var h in hashBytes)
{
if (count < 7)
{
Assert.AreNotEqual(h, 0, "checksum mismatch");
}
else
{
// wildcard position
Assert.AreEqual(h, 0, "checksum mismatch");
}
++count;
}
// The mask should have ff
var hashMaskBytes = GetBytes(topicHashMask);
count = 0;
foreach (var h in hashMaskBytes)
{
if (count < 7)
{
Assert.AreEqual(h, 0xff, "mask mismatch");
}
else
{
Assert.AreEqual(h, 0, "last mask mismatch");
}
++count;
}
}
[TestMethod]
@@ -65,13 +317,9 @@ namespace MQTTnet.Tests.Server
var l1 = "topic1";
var l2 = "sub1";
var l3 = "#";
var topic = string.Format("{0}/{1}/{2}/{3}", l0, l1, l2, l3);
var topic = $"{l0}/{l1}/{l2}/{l3}";
UInt64 topicHash;
UInt64 topicHashMask;
bool topicHasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
@@ -104,13 +352,9 @@ namespace MQTTnet.Tests.Server
var l1 = "topic1";
var l2 = "sub1";
var l3 = "prop1";
var topic = string.Format("{0}/{1}/{2}/{3}", l0, l1, l2, l3);
var topic = $"{l0}/{l1}/{l2}/{l3}";
UInt64 topicHash;
UInt64 topicHashMask;
bool topicHasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
Assert.IsFalse(topicHasWildcard, "Wildcard detected when not wildcard present");
@@ -137,225 +381,13 @@ namespace MQTTnet.Tests.Server
Assert.AreEqual(hashMaskBytes[7], 0xff, "mask 7 mismatch");
}
/// <summary>
/// Test long topic name with last level being #
/// </summary>
[TestMethod]
public void Match_Hash_Test_LongTopic_MultiWildcard()
public async Task Match_Hash_Test_Search_MultiWildcard()
{
var sb = new StringBuilder();
const int NumLevels = 8;
var levelNames = new string[NumLevels];
for (var i = 0; i < NumLevels; ++i)
{
if (i > 0)
sb.Append("/");
string levelName;
if (i == NumLevels - 1)
{
// last one is #
levelName = "#";
}
else
{
levelName = "level" + i;
}
levelNames[i] = levelName;
sb.Append(levelName);
}
var topic = sb.ToString();
// UInt64 is limited to 8 levels
UInt64 topicHash;
UInt64 topicHashMask;
bool topicHasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
var hashBytes = GetBytes(topicHash);
// all bytes should contain checksum
int count = 0;
foreach (var h in hashBytes)
{
if (count < 7)
{
Assert.AreNotEqual(h, 0, "checksum mismatch");
}
else
{
Assert.AreEqual(h, 0, "checksum mismatch");
}
++count;
}
// The mask should have ff except for last level
var hashMaskBytes = GetBytes(topicHashMask);
count = 0;
foreach (var h in hashMaskBytes)
{
if (count < 7)
{
Assert.AreEqual(h, 0xff, "mask mismatch");
}
else
{
Assert.AreEqual(h, 0, "last mask mismatch");
}
++count;
}
}
/// <summary>
/// Test long topic name with last level being +
/// </summary>
[TestMethod]
public void Match_Hash_Test_LongTopic_SingleWildCard()
{
var sb = new StringBuilder();
const int NumLevels = 8;
var levelNames = new string[NumLevels];
for (var i = 0; i < NumLevels; ++i)
{
if (i > 0)
sb.Append("/");
string levelName;
if (i == NumLevels - 1)
{
// last one is +
levelName = "+";
}
else
{
levelName = "level" + i;
}
levelNames[i] = levelName;
sb.Append(levelName);
}
var topic = sb.ToString();
UInt64 topicHash;
UInt64 topicHashMask;
bool topicHasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
var hashBytes = GetBytes(topicHash);
// all bytes should contain checksum
int count = 0;
foreach (var h in hashBytes)
{
if (count < 7)
{
Assert.AreNotEqual(h, 0, "checksum mismatch");
}
else
{
// wildcard position
Assert.AreEqual(h, 0, "checksum mismatch");
}
++count;
}
// The mask should have ff
var hashMaskBytes = GetBytes(topicHashMask);
count = 0;
foreach (var h in hashMaskBytes)
{
if (count < 7)
{
Assert.AreEqual(h, 0xff, "mask mismatch");
}
else
{
Assert.AreEqual(h, 0, "last mask mismatch");
}
++count;
}
}
/// <summary>
/// Test long topic name exceeding 8 levels
/// </summary>
[TestMethod]
public void Match_Hash_Test_LongTopic_NoWildCard()
{
var sb = new StringBuilder();
const int NumLevels = 9;
var levelNames = new string[NumLevels];
for(var i=0; i < NumLevels; ++i)
{
if (i > 0)
sb.Append("/");
var levelName = "level" + i;
levelNames[i] = levelName;
sb.Append(levelName);
}
var topic = sb.ToString();
UInt64 topicHash;
UInt64 topicHashMask;
bool topicHasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
Assert.IsFalse(topicHasWildcard, "Wildcard detected when not present");
var hashBytes = GetBytes(topicHash);
// all bytes should contain checksum
int count = 0;
foreach(var h in hashBytes)
{
Assert.AreNotEqual(h, 0, "checksum mismatch");
++count;
}
// The mask should have ff
var hashMaskBytes = GetBytes(topicHashMask);
count = 0;
foreach (var h in hashMaskBytes)
{
Assert.AreEqual(h, 0xff, "mask mismatch");
++count;
}
}
/// <summary>
/// Test long topic name with last level being +
/// </summary>
[TestMethod]
public void Match_Hash_Test_LongTopic_DetectSingleLevelWildcard()
{
var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/+";
UInt64 topicHash;
UInt64 topicHashMask;
bool topicHasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
}
/// <summary>
/// Test long topic name with last level being #
/// </summary>
[TestMethod]
public void Match_Hash_Test_LongTopic_DetectMultiLevelWildcard()
{
var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/#";
UInt64 topicHash;
UInt64 topicHashMask;
bool topicHasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard);
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
var topics = await PrepareTopicHashSubscriptions(TopicHashSelector.MultiWildcard);
var matchCount = CheckTopicSubscriptions(_clientSession, topics, "Multi Wildcard");
// Should match all topics
Assert.AreEqual(topics.Count, matchCount, "Topics not matched");
}
[TestMethod]
@@ -407,134 +439,125 @@ namespace MQTTnet.Tests.Server
}
[TestMethod]
public async Task Match_Hash_Test_Search_MultiWildcard()
public void Match_Hash_Test_SingleWildCard()
{
var topics = await PrepareTopicHashSubscriptions(TopicHashSelector.MultiWildcard);
var matchCount = CheckTopicSubscriptions(_clientSession, topics, "Multi Wildcard");
// Should match all topics
Assert.AreEqual(topics.Count, matchCount, "Topics not matched");
}
var l0 = "pub0";
var l1 = "topic1";
var l2 = "+";
var l3 = "prop1";
var topic = $"{l0}/{l1}/{l2}/{l3}";
/// <summary>
/// Even fairly regularly named topics as generated by the topic generator should result in shallow hash buckets
/// </summary>
/// <returns></returns>
[TestMethod]
public void Check_Hash_Bucket_Depth()
{
Dictionary<string, List<string>> topicsByPublisher;
Dictionary<string, List<string>> singleWildcardTopicsByPublisher;
Dictionary<string, List<string>> multiWildcardTopicsByPublisher;
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard);
const int NumPublishers = 5000;
const int NumTopicsPerPublisher = 10;
Assert.IsTrue(topicHasWildcard, "Wildcard not detected");
TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out topicsByPublisher, out singleWildcardTopicsByPublisher, out multiWildcardTopicsByPublisher);
var hashBytes = GetBytes(topicHash);
Assert.AreNotEqual(hashBytes[0], 0, "checksum 0 mismatch");
Assert.AreNotEqual(hashBytes[1], 0, "checksum 1 mismatch");
Assert.AreEqual(hashBytes[2], 0, "checksum 2 mismatch");
Assert.AreNotEqual(hashBytes[3], 0, "checksum 3 mismatch");
Assert.AreEqual(hashBytes[4], 0, "checksum 4 mismatch");
Assert.AreEqual(hashBytes[5], 0, "checksum 5 mismatch");
Assert.AreEqual(hashBytes[6], 0, "checksum 6 mismatch");
Assert.AreEqual(hashBytes[7], 0, "checksum 7 mismatch");
// There will be many 'similar' topics ending with, i.e. "sensor100", "sensor101", ...
// Hash bucket depths should remain low.
var bucketDepths = new Dictionary<UInt64, int>();
int maxBucketDepth = 0;
UInt64 maxBucketDepthHash = 0;
var topicsByHash = new Dictionary<UInt64, List<string>>();
foreach (var t in topicsByPublisher)
{
var topics = t.Value;
foreach (var topic in topics)
{
UInt64 topicHash;
UInt64 hashMask;
bool hasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out hashMask, out hasWildcard);
bucketDepths.TryGetValue(topicHash, out var currentValue);
++currentValue;
bucketDepths[topicHash] = currentValue;
if (currentValue > maxBucketDepth)
{
maxBucketDepth = currentValue;
maxBucketDepthHash = topicHash;
}
if (!topicsByHash.TryGetValue(topicHash, out var topicList))
{
topicList = new List<string>();
topicsByHash.Add(topicHash, topicList);
}
topicList.Add(topic);
}
}
var maxDepthTopics = topicsByHash[maxBucketDepthHash];
Console.Write("Max bucket depth is " + maxBucketDepth);
// for the test case the bucket depth should be less than 100
Assert.IsTrue(maxBucketDepth < 100, "Unexpected high topic hash bucket depth");
}
[TestMethod]
public void Check_Selected_Topic_Hashes()
{
CheckTopicHash("client1/building1/level1/sensor1", 0x655D4AF100000000, 0xFFFFFFFFFFFFFFFF);
CheckTopicHash("client1/building1/+/sensor1", 0x655D00F100000000, 0xFFFF00FFFFFFFFFF);
CheckTopicHash("client1/+/level1/+", 0x65004A0000000000, 0xFF00FF00FFFFFFFF);
CheckTopicHash("client1/building1/level1/#", 0x655D4A0000000000, 0xFFFFFF0000000000);
CheckTopicHash("client1/+/level1/#", 0x65004A0000000000, 0xFF00FF0000000000);
// The mask should have zeroes where the wildcard and ff at the end
var hashMaskBytes = GetBytes(topicHashMask);
Assert.AreEqual(hashMaskBytes[0], 0xff, "mask 0 mismatch");
Assert.AreEqual(hashMaskBytes[1], 0xff, "mask 1 mismatch");
Assert.AreEqual(hashMaskBytes[2], 0, "mask 2 mismatch");
Assert.AreEqual(hashMaskBytes[3], 0xff, "mask 3 mismatch");
Assert.AreEqual(hashMaskBytes[4], 0xff, "mask 4 mismatch");
Assert.AreEqual(hashMaskBytes[5], 0xff, "mask 5 mismatch");
Assert.AreEqual(hashMaskBytes[6], 0xff, "mask 6 mismatch");
Assert.AreEqual(hashMaskBytes[7], 0xff, "mask 7 mismatch");
}
void CheckTopicHash(string topic, ulong expectedHash, ulong expectedHashMask)
{
ulong topicHash;
ulong hashMask;
bool hasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out hashMask, out hasWildcard);
MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var hashMask, out var hasWildcard);
Console.WriteLine();
Console.WriteLine("Topic: " + topic);
Console.WriteLine(string.Format("Hash: {0:X8}", topicHash));
Console.WriteLine(string.Format("Hash Mask: {0:X8}", hashMask));
Console.WriteLine($"Hash: {topicHash:X8}");
Console.WriteLine($"Hash Mask: {hashMask:X8}");
Assert.AreEqual(expectedHash, topicHash, "Topic hash not as expected. Has the hash function changed?");
Assert.AreEqual(expectedHashMask, hashMask, "Topic hash mask not as expected");
}
int CheckTopicSubscriptions(MqttSession clientSession, List<string> topicsToFind, string subject)
{
var matchCount = 0;
{
var resultCount = 0;
var stopWatch = new Stopwatch();
stopWatch.Start();
var countUp = 0;
var countDown = topicsToFind.Count - 1;
for (; countUp < topicsToFind.Count; ++countUp, --countDown)
{
var topicToFind = topicsToFind[countDown];
MqttSubscription.CalculateTopicHash(topicToFind, out var topicHash, out _, out _);
clientSession.TryCheckSubscriptions(topicToFind, topicHash, MqttQualityOfServiceLevel.AtMostOnce, "OtherClient", out var result);
if (result.IsSubscribed)
{
++matchCount;
}
++resultCount;
}
stopWatch.Stop();
Console.Write("Match count: " + matchCount + "; ");
Console.WriteLine(subject + " lookup milliseconds: " + stopWatch.ElapsedMilliseconds);
}
return matchCount;
}
byte[] GetBytes(ulong value)
{
var bytes = BitConverter.GetBytes(value);
// Ensure that highest byte comes first for comparison left to right
if (BitConverter.IsLittleEndian)
{
return bytes.Reverse().ToArray();
}
return bytes;
}
async Task<List<string>> PrepareTopicHashSubscriptions(TopicHashSelector selector)
{
Dictionary<string, List<string>> topicsByPublisher;
Dictionary<string, List<string>> singleWildcardTopicsByPublisher;
Dictionary<string, List<string>> multiWildcardTopicsByPublisher;
const int numPublishers = 1;
const int numTopicsPerPublisher = 10000;
const int NumPublishers = 1;
const int NumTopicsPerPublisher = 10000;
TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out topicsByPublisher, out singleWildcardTopicsByPublisher, out multiWildcardTopicsByPublisher);
TopicGenerator.Generate(
numPublishers,
numTopicsPerPublisher,
out var topicsByPublisher,
out var singleWildcardTopicsByPublisher,
out var multiWildcardTopicsByPublisher);
var topics = topicsByPublisher.FirstOrDefault().Value;
var singleWildcardTopics = singleWildcardTopicsByPublisher.FirstOrDefault().Value;
var multiWildcardTopics = multiWildcardTopicsByPublisher.FirstOrDefault().Value;
const string ClientId = "Client1";
var logger = new Mockups.TestLogger();
var serverOptions = new MQTTnet.Server.MqttServerOptions();
var eventContainer = new MQTTnet.Server.MqttServerEventContainer();
const string clientId = "Client1";
var logger = new TestLogger();
var serverOptions = new MqttServerOptions();
var eventContainer = new MqttServerEventContainer();
var retainedMessagesManager = new MqttRetainedMessagesManager(eventContainer, logger);
var sessionManager = new MqttClientSessionsManager(serverOptions, retainedMessagesManager, eventContainer, logger);
_clientSession = new MQTTnet.Server.MqttSession(
ClientId,
false,
new Dictionary<object, object>(),
serverOptions,
eventContainer,
retainedMessagesManager,
sessionManager
);
_clientSession = new MqttSession(clientId, false, new Dictionary<object, object>(), serverOptions, eventContainer, retainedMessagesManager, sessionManager);
List<string> topicsToSubscribe;
@@ -550,63 +573,26 @@ namespace MQTTnet.Tests.Server
topicsToSubscribe = topics;
break;
}
foreach (var t in topicsToSubscribe)
{
var subPacket = new Packets.MqttSubscribePacket();
var filter = new Packets.MqttTopicFilter();
filter.Topic = t;
var subPacket = new MqttSubscribePacket();
var filter = new MqttTopicFilter
{
Topic = t
};
subPacket.TopicFilters.Add(filter);
await _clientSession.SubscriptionsManager.Subscribe(subPacket, default(CancellationToken));
await _clientSession.Subscribe(subPacket, default);
}
return topics;
}
int CheckTopicSubscriptions(MQTTnet.Server.MqttSession clientSession, List<string> topicsToFind, string subject)
enum TopicHashSelector
{
var matchCount = 0;
{
int resultCount = 0;
var stopWatch = new System.Diagnostics.Stopwatch();
stopWatch.Start();
var countUp = 0;
var countDown = topicsToFind.Count - 1;
for (; countUp < topicsToFind.Count; ++countUp, --countDown)
{
var topicToFind = topicsToFind[countDown];
UInt64 topicHash;
UInt64 hashMask;
bool hasWildcard;
MQTTnet.Server.MqttSubscription.CalculateTopicHash((string)topicToFind, out topicHash, out hashMask, out hasWildcard);
var result = clientSession.SubscriptionsManager.CheckSubscriptions((string)topicToFind, topicHash, MqttQualityOfServiceLevel.AtMostOnce, "OtherClient");
if (result.IsSubscribed)
{
++matchCount;
}
++resultCount;
}
stopWatch.Stop();
Console.Write("Match count: " + matchCount + "; ");
Console.WriteLine(subject + " lookup milliseconds: " + stopWatch.ElapsedMilliseconds);
}
return matchCount;
}
byte[] GetBytes(UInt64 value)
{
var bytes = BitConverter.GetBytes(value);
// Ensure that highest byte comes first for comparison left to right
if (BitConverter.IsLittleEndian)
return bytes.Reverse().ToArray();
return bytes;
NoWildcard,
SingleWildcard,
MultiWildcard
}
}
}
}
@@ -0,0 +1,32 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Protocol;
namespace MQTTnet.Tests.Server
{
[TestClass]
public sealed class Unsubscribe_Tests : BaseTestClass
{
[TestMethod]
[ExpectedException(typeof(MqttClientDisconnectedException))]
public async Task Disconnect_While_Unsubscribing()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
// The client will be disconnect directly after subscribing!
server.ClientUnsubscribedTopicAsync += ev => server.DisconnectClientAsync(ev.ClientId, MqttDisconnectReasonCode.NormalDisconnection);
var client = await testEnvironment.ConnectClient();
await client.SubscribeAsync("#");
await client.UnsubscribeAsync("#");
}
}
}
}
+1 -2
View File
@@ -1,8 +1,7 @@
using System;
using System.Collections.Generic;
using System.Text;
namespace MQTTnet.Tests.Server
namespace MQTTnet.Tests
{
public class TopicGenerator
{
+22 -5
View File
@@ -13,7 +13,6 @@ using MQTTnet.Channel;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Implementations;
using MQTTnet.Internal;
using MQTTnet.Packets;
@@ -87,7 +86,7 @@ namespace MQTTnet.Adapter
await Task.WhenAny(connectTask, timeout.Task).ConfigureAwait(false);
if (timeout.Task.IsCompleted && !connectTask.IsCompleted)
{
throw new OperationCanceledException("MQTT connect cancelled.", cancellationToken);
throw new OperationCanceledException("MQTT connect canceled.", cancellationToken);
}
// Make sure that the exception from the connect task gets thrown.
@@ -296,11 +295,29 @@ namespace MQTTnet.Adapter
while (totalBytesRead < buffer.Length)
{
var bytesRead = await _channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
// Check two times for cancellation because the call to _ReadAsync_ might block for some time.
if (cancellationToken.IsCancellationRequested)
{
return ReadFixedHeaderResult.Cancelled;
return ReadFixedHeaderResult.Canceled;
}
int bytesRead;
try
{
bytesRead = await _channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
return ReadFixedHeaderResult.Canceled;
}
catch (SocketException)
{
return ReadFixedHeaderResult.ConnectionClosed;
}
if (cancellationToken.IsCancellationRequested)
{
return ReadFixedHeaderResult.Canceled;
}
if (bytesRead == 0)
@@ -0,0 +1,16 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using MQTTnet.Exceptions;
namespace MQTTnet.Client
{
public sealed class MqttClientDisconnectedException : MqttCommunicationException
{
public MqttClientDisconnectedException(Exception innerException) : base("The MQTT client is disconnected.", innerException)
{
}
}
}
@@ -3,14 +3,15 @@
// See the LICENSE file in the project root for more information.
using System.Collections.Generic;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Exceptions
namespace MQTTnet.Client
{
public class MqttUnexpectedDisconnectReceivedException : MqttCommunicationException
public sealed class MqttClientUnexpectedDisconnectReceivedException : MqttCommunicationException
{
public MqttUnexpectedDisconnectReceivedException(MqttDisconnectPacket disconnectPacket)
public MqttClientUnexpectedDisconnectReceivedException(MqttDisconnectPacket disconnectPacket)
: base($"Unexpected DISCONNECT (Reason code={disconnectPacket.ReasonCode}) received.")
{
ReasonCode = disconnectPacket.ReasonCode;
+15 -8
View File
@@ -32,25 +32,21 @@ namespace MQTTnet.Client
readonly object _disconnectLock = new object();
readonly AsyncEvent<InspectMqttPacketEventArgs> _inspectPacketEvent = new AsyncEvent<InspectMqttPacketEventArgs>();
readonly MqttNetSourceLogger _logger;
readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
readonly IMqttNetLogger _rootLogger;
IMqttChannelAdapter _adapter;
CancellationTokenSource _backgroundCancellationTokenSource;
bool _cleanDisconnectInitiated;
volatile int _connectionStatus;
MqttClientDisconnectReason _disconnectReason;
string _disconnectReasonString;
Task _keepAlivePacketsSenderTask;
DateTime _lastPacketSentTimestamp;
MqttPacketDispatcher _packetDispatcher;
Task _packetReceiverTask;
AsyncQueue<MqttPublishPacket> _publishPacketReceiverQueue;
Task _publishPacketReceiverTask;
@@ -128,7 +124,7 @@ namespace MQTTnet.Client
}
_packetIdentifierProvider.Reset();
_packetDispatcher.CancelAll();
_packetDispatcher = new MqttPacketDispatcher();
_backgroundCancellationTokenSource = new CancellationTokenSource();
var backgroundCancellationToken = _backgroundCancellationTokenSource.Token;
@@ -414,6 +410,10 @@ namespace MQTTnet.Client
_publishPacketReceiverQueue = null;
_adapter?.Dispose();
_adapter = null;
_packetDispatcher?.Dispose();
_packetDispatcher = null;
}
}
@@ -447,6 +447,8 @@ namespace MQTTnet.Client
try
{
_packetDispatcher.Dispose(new MqttClientDisconnectedException(exception));
var receiverTask = WaitForTaskAsync(_packetReceiverTask, sender);
var publishPacketReceiverTask = WaitForTaskAsync(_publishPacketReceiverTask, sender);
var keepAliveTask = WaitForTaskAsync(_keepAlivePacketsSenderTask, sender);
@@ -558,7 +560,7 @@ namespace MQTTnet.Client
_disconnectReasonString = disconnectPacket.ReasonString;
// Also dispatch disconnect to waiting threads to generate a proper exception.
_packetDispatcher.FailAll(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
_packetDispatcher.Dispose(new MqttClientUnexpectedDisconnectReceivedException(disconnectPacket));
return DisconnectInternalAsync(_packetReceiverTask, null, null);
}
@@ -776,7 +778,7 @@ namespace MQTTnet.Client
}
else
{
_logger.Error(exception, "Error while receiving packets.");
_logger.Error(exception, $"Error while processing received packet ({packet.GetType().Name}).");
}
_packetDispatcher.FailAll(exception);
@@ -827,6 +829,11 @@ namespace MQTTnet.Client
return;
}
if (exception is AggregateException aggregateException)
{
exception = aggregateException.GetBaseException();
}
if (exception is OperationCanceledException)
{
}
+31 -20
View File
@@ -14,26 +14,6 @@ namespace MQTTnet.Client
{
public static class MqttClientExtensions
{
public static async Task<bool> TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default)
{
if (client == null)
{
throw new ArgumentNullException(nameof(client));
}
try
{
await client.PingAsync(cancellationToken).ConfigureAwait(false);
return true;
}
catch
{
// Ignore errors.
}
return false;
}
public static Task DisconnectAsync(this IMqttClient client, MqttClientDisconnectReason reason = MqttClientDisconnectReason.NormalDisconnection, string reasonString = null)
{
if (client == null)
@@ -89,6 +69,17 @@ namespace MQTTnet.Client
return mqttClient.PublishBinaryAsync(topic, payloadBuffer, qualityOfServiceLevel, retain, cancellationToken);
}
public static Task ReconnectAsync(this IMqttClient client, CancellationToken cancellationToken = default)
{
if (client.Options == null)
{
throw new InvalidOperationException(
"The MQTT client was not connected before. A reconnect is only permitted when the client was already connected or at least tried to.");
}
return client.ConnectAsync(client.Options, cancellationToken);
}
public static Task SendExtendedAuthenticationExchangeDataAsync(this IMqttClient client, MqttExtendedAuthenticationExchangeData data)
{
if (client == null)
@@ -137,6 +128,26 @@ namespace MQTTnet.Client
return mqttClient.SubscribeAsync(subscribeOptions, cancellationToken);
}
public static async Task<bool> TryPingAsync(this IMqttClient client, CancellationToken cancellationToken = default)
{
if (client == null)
{
throw new ArgumentNullException(nameof(client));
}
try
{
await client.PingAsync(cancellationToken).ConfigureAwait(false);
return true;
}
catch
{
// Ignore errors.
}
return false;
}
public static Task<MqttClientUnsubscribeResult> UnsubscribeAsync(this IMqttClient mqttClient, string topic, CancellationToken cancellationToken = default)
{
if (mqttClient == null)
@@ -10,29 +10,38 @@ namespace MQTTnet.Client
public sealed class MqttClientPublishResult
{
/// <summary>
/// Gets the packet identifier which was used for this publish.
/// Returns if the overall status of the publish is a success. This can be the reason code _Success_ or
/// _NoMatchingSubscribers_. _NoMatchingSubscribers_ usually indicates only that no other client is interested in the topic but overall transmit
/// to the server etc. was a success.
/// </summary>
public bool IsSuccess => ReasonCode == MqttClientPublishReasonCode.Success || ReasonCode == MqttClientPublishReasonCode.NoMatchingSubscribers;
/// <summary>
/// Gets the packet identifier which was used for this publish.
/// </summary>
public ushort? PacketIdentifier { get; set; }
/// <summary>
/// Gets or sets the reason code.
/// Hint: MQTT 5 feature only.
/// Gets or sets the reason code.
/// Hint: MQTT 5 feature only.
/// </summary>
public MqttClientPublishReasonCode ReasonCode { get; set; } = MqttClientPublishReasonCode.Success;
/// <summary>
/// Gets or sets the reason string.
/// Hint: MQTT 5 feature only.
/// Gets or sets the reason string.
/// Hint: MQTT 5 feature only.
/// </summary>
public string ReasonString { get; set; }
/// <summary>
/// Gets or sets the user properties.
/// In MQTT 5, user properties are basic UTF-8 string key-value pairs that you can append to almost every type of MQTT packet.
/// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add metadata to MQTT messages and pass information between publisher, broker, and subscriber.
/// The feature is very similar to the HTTP header concept.
/// Hint: MQTT 5 feature only.
/// Gets or sets the user properties.
/// In MQTT 5, user properties are basic UTF-8 string key-value pairs that you can append to almost every type of MQTT
/// packet.
/// As long as you don’t exceed the maximum message size, you can use an unlimited number of user properties to add
/// metadata to MQTT messages and pass information between publisher, broker, and subscriber.
/// The feature is very similar to the HTTP header concept.
/// Hint: MQTT 5 feature only.
/// </summary>
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; internal set; }
}
}
}
@@ -8,12 +8,8 @@ namespace MQTTnet.Exceptions
{
public class MqttCommunicationException : Exception
{
protected MqttCommunicationException()
{
}
public MqttCommunicationException(Exception innerException)
: base(innerException.Message, innerException)
: base(innerException?.Message ?? "MQTT communication failed.", innerException)
{
}
@@ -7,6 +7,7 @@ namespace MQTTnet.Formatter
public sealed class MqttPacketFactories
{
public MqttConnAckPacketFactory ConnAck { get; } = new MqttConnAckPacketFactory();
public MqttConnectPacketFactory Connect { get; } = new MqttConnectPacketFactory();
public MqttDisconnectPacketFactory Disconnect { get; } = new MqttDisconnectPacketFactory();
@@ -12,7 +12,10 @@ namespace MQTTnet.Formatter
{
public sealed class MqttPubAckPacketFactory
{
public MqttPubAckPacket Create(MqttPublishPacket publishPacket, InterceptingPublishEventArgs interceptingPublishEventArgs)
public MqttPubAckPacket Create(
MqttPublishPacket publishPacket,
InterceptingPublishEventArgs interceptingPublishEventArgs,
DispatchApplicationMessageResult dispatchApplicationMessageResult)
{
if (publishPacket == null)
{
@@ -21,10 +24,19 @@ namespace MQTTnet.Formatter
var pubAckPacket = new MqttPubAckPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubAckReasonCode.Success
PacketIdentifier = publishPacket.PacketIdentifier
};
if (dispatchApplicationMessageResult.MatchingSubscribersCount == 0)
{
// NoMatchingSubscribers is ONLY sent by the server!
pubAckPacket.ReasonCode = MqttPubAckReasonCode.NoMatchingSubscribers;
}
else
{
pubAckPacket.ReasonCode = MqttPubAckReasonCode.Success;
}
if (interceptingPublishEventArgs != null)
{
pubAckPacket.ReasonCode = (MqttPubAckReasonCode)(int)interceptingPublishEventArgs.Response.ReasonCode;
@@ -42,24 +54,12 @@ namespace MQTTnet.Formatter
throw new ArgumentNullException(nameof(applicationMessageReceivedEventArgs));
}
var pubAckPacket = Create(applicationMessageReceivedEventArgs.PublishPacket, applicationMessageReceivedEventArgs.ReasonCode);
pubAckPacket.UserProperties = applicationMessageReceivedEventArgs.ResponseUserProperties;
pubAckPacket.ReasonString = applicationMessageReceivedEventArgs.ResponseReasonString;
return pubAckPacket;
}
static MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode)
{
if (publishPacket == null)
{
throw new ArgumentNullException(nameof(publishPacket));
}
var pubAckPacket = new MqttPubAckPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = (MqttPubAckReasonCode)(int)applicationMessageReceivedReasonCode
PacketIdentifier = applicationMessageReceivedEventArgs.PublishPacket.PacketIdentifier,
ReasonCode = (MqttPubAckReasonCode)(int)applicationMessageReceivedEventArgs.ReasonCode,
UserProperties = applicationMessageReceivedEventArgs.ResponseUserProperties,
ReasonString = applicationMessageReceivedEventArgs.ResponseReasonString
};
return pubAckPacket;
@@ -25,7 +25,10 @@ namespace MQTTnet.Formatter
return pubRecPacket;
}
public MqttPacket Create(MqttPublishPacket publishPacket, InterceptingPublishEventArgs interceptingPublishEventArgs)
public MqttPacket Create(
MqttPublishPacket publishPacket,
InterceptingPublishEventArgs interceptingPublishEventArgs,
DispatchApplicationMessageResult dispatchApplicationMessageResult)
{
if (publishPacket == null)
{
@@ -38,6 +41,11 @@ namespace MQTTnet.Formatter
ReasonCode = MqttPubRecReasonCode.Success
};
if (dispatchApplicationMessageResult.MatchingSubscribersCount == 0)
{
pubRecPacket.ReasonCode = MqttPubRecReasonCode.NoMatchingSubscribers;
}
if (interceptingPublishEventArgs != null)
{
pubRecPacket.ReasonCode = (MqttPubRecReasonCode)(int)interceptingPublishEventArgs.Response.ReasonCode;
@@ -5,6 +5,7 @@
using System;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Server;
namespace MQTTnet.Formatter
{
@@ -80,5 +81,17 @@ namespace MQTTnet.Formatter
return packet;
}
public MqttPublishPacket Create(MqttRetainedMessageMatch retainedMessage)
{
if (retainedMessage == null)
{
throw new ArgumentNullException(nameof(retainedMessage));
}
var publishPacket = Create(retainedMessage.ApplicationMessage);
publishPacket.QualityOfServiceLevel = retainedMessage.SubscriptionQualityOfServiceLevel;
return publishPacket;
}
}
}
@@ -10,16 +10,16 @@ namespace MQTTnet.Formatter
{
public sealed class MqttUnsubAckPacketFactory
{
public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, MqttUnsubscribeResult mqttUnsubscribeResult)
public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult)
{
if (unsubscribePacket == null)
{
throw new ArgumentNullException(nameof(unsubscribePacket));
}
if (mqttUnsubscribeResult == null)
if (unsubscribeResult == null)
{
throw new ArgumentNullException(nameof(mqttUnsubscribeResult));
throw new ArgumentNullException(nameof(unsubscribeResult));
}
var unsubAckPacket = new MqttUnsubAckPacket
@@ -28,7 +28,7 @@ namespace MQTTnet.Formatter
};
// MQTTv5.0.0 only.
unsubAckPacket.ReasonCodes = mqttUnsubscribeResult.ReasonCodes;
unsubAckPacket.ReasonCodes = unsubscribeResult.ReasonCodes;
return unsubAckPacket;
}
@@ -6,9 +6,9 @@ namespace MQTTnet.Formatter
{
public struct ReadFixedHeaderResult
{
public static ReadFixedHeaderResult Cancelled { get; } = new ReadFixedHeaderResult
public static ReadFixedHeaderResult Canceled { get; } = new ReadFixedHeaderResult
{
IsCancelled = true
IsCanceled = true
};
public static ReadFixedHeaderResult ConnectionClosed { get; } = new ReadFixedHeaderResult
@@ -16,7 +16,7 @@ namespace MQTTnet.Formatter
IsConnectionClosed = true
};
public bool IsCancelled { get; set; }
public bool IsCanceled { get; set; }
public bool IsConnectionClosed { get; set; }
+1 -1
View File
@@ -123,7 +123,7 @@ namespace MQTTnet.Internal
}
catch (Exception exception)
{
logger.Warning(exception, $"Error while invoking event ({typeof(TEventArgs)}).");
logger.Warning(exception, $"Error while invoking event with arguments of type {typeof(TEventArgs)}.");
}
}
}
+71 -66
View File
@@ -11,74 +11,15 @@ namespace MQTTnet.Internal
{
public sealed class AsyncQueue<TItem> : IDisposable
{
readonly AsyncSignal _signal = new AsyncSignal();
readonly object _syncRoot = new object();
SemaphoreSlim _semaphore = new SemaphoreSlim(0);
ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
bool _isDisposed;
public int Count => _queue.Count;
public void Enqueue(TItem item)
{
lock (_syncRoot)
{
_queue.Enqueue(item);
_semaphore?.Release();
}
}
public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
try
{
Task task;
lock (_syncRoot)
{
if (_semaphore == null)
{
return new AsyncQueueDequeueResult<TItem>(false, default);
}
task = _semaphore.WaitAsync(cancellationToken);
}
await task.ConfigureAwait(false);
if (cancellationToken.IsCancellationRequested)
{
return new AsyncQueueDequeueResult<TItem>(false, default);
}
if (_queue.TryDequeue(out var item))
{
return new AsyncQueueDequeueResult<TItem>(true, item);
}
}
catch (ArgumentNullException)
{
// The semaphore throws this internally sometimes.
return new AsyncQueueDequeueResult<TItem>(false, default);
}
catch (OperationCanceledException)
{
return new AsyncQueueDequeueResult<TItem>(false, default);
}
}
return new AsyncQueueDequeueResult<TItem>(false, default);
}
public AsyncQueueDequeueResult<TItem> TryDequeue()
{
if (_queue.TryDequeue(out var item))
{
return new AsyncQueueDequeueResult<TItem>(true, item);
}
return new AsyncQueueDequeueResult<TItem>(false, default);
}
public void Clear()
{
Interlocked.Exchange(ref _queue, new ConcurrentQueue<TItem>());
@@ -88,9 +29,73 @@ namespace MQTTnet.Internal
{
lock (_syncRoot)
{
_semaphore?.Dispose();
_semaphore = null;
_signal.Dispose();
_isDisposed = true;
}
}
public void Enqueue(TItem item)
{
lock (_syncRoot)
{
_queue.Enqueue(item);
_signal.Set();
}
}
public AsyncQueueDequeueResult<TItem> TryDequeue()
{
if (_queue.TryDequeue(out var item))
{
return AsyncQueueDequeueResult<TItem>.Success(item);
}
return AsyncQueueDequeueResult<TItem>.NonSuccess;
}
public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
try
{
Task task = null;
lock (_syncRoot)
{
if (_isDisposed)
{
return AsyncQueueDequeueResult<TItem>.NonSuccess;
}
if (_queue.IsEmpty)
{
task = _signal.WaitAsync(cancellationToken);
}
}
if (task != null)
{
await task.ConfigureAwait(false);
}
if (cancellationToken.IsCancellationRequested)
{
return AsyncQueueDequeueResult<TItem>.NonSuccess;
}
if (_queue.TryDequeue(out var item))
{
return AsyncQueueDequeueResult<TItem>.Success(item);
}
}
catch (OperationCanceledException)
{
return AsyncQueueDequeueResult<TItem>.NonSuccess;
}
}
return AsyncQueueDequeueResult<TItem>.NonSuccess;
}
}
}
}
@@ -4,8 +4,10 @@
namespace MQTTnet.Internal
{
public class AsyncQueueDequeueResult<TItem>
public sealed class AsyncQueueDequeueResult<TItem>
{
public static readonly AsyncQueueDequeueResult<TItem> NonSuccess = new AsyncQueueDequeueResult<TItem>(false, default);
public AsyncQueueDequeueResult(bool isSuccess, TItem item)
{
IsSuccess = isSuccess;
@@ -15,5 +17,10 @@ namespace MQTTnet.Internal
public bool IsSuccess { get; }
public TItem Item { get; }
public static AsyncQueueDequeueResult<TItem> Success(TItem item)
{
return new AsyncQueueDequeueResult<TItem>(true, item);
}
}
}
+133
View File
@@ -0,0 +1,133 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.Internal
{
public sealed class AsyncSignal : IDisposable
{
readonly object _syncRoot = new object();
bool _isDisposed;
bool _isSignaled;
AsyncSignalWaiter _waiter;
public void Dispose()
{
lock (_syncRoot)
{
_waiter?.Dispose();
_waiter = null;
_isDisposed = true;
}
}
public void Set()
{
lock (_syncRoot)
{
_isSignaled = true;
Cleanup();
// If there is already a waiting task let it run.
if (_waiter != null)
{
_waiter.Approve();
_waiter = null;
// Since we already got a waiter the signal must be reset right now!
_isSignaled = false;
}
}
}
public Task WaitAsync(CancellationToken cancellationToken = default)
{
cancellationToken.ThrowIfCancellationRequested();
lock (_syncRoot)
{
ThrowIfDisposed();
Cleanup();
if (_isSignaled)
{
_isSignaled = false;
return CompletedTask.Instance;
}
if (_waiter != null)
{
if (!_waiter.Task.IsCompleted)
{
throw new InvalidOperationException("Only one waiting task is permitted per async signal.");
}
_waiter.Dispose();
}
_waiter = new AsyncSignalWaiter(cancellationToken);
return _waiter.Task;
}
}
void Cleanup()
{
// Cleanup if the previous waiter was cancelled.
if (_waiter != null && _waiter.Task.IsCanceled)
{
_waiter = null;
}
}
void ThrowIfDisposed()
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(AsyncSignal));
}
}
sealed class AsyncSignalWaiter : IDisposable
{
readonly AsyncTaskCompletionSource<bool> _promise = new AsyncTaskCompletionSource<bool>();
// ReSharper disable once FieldCanBeMadeReadOnly.Local
CancellationTokenRegistration _cancellationTokenRegistration;
public AsyncSignalWaiter(CancellationToken cancellationToken)
{
if (cancellationToken.CanBeCanceled)
{
_cancellationTokenRegistration = cancellationToken.Register(Cancel);
}
}
public Task Task => _promise.Task;
public void Approve()
{
_promise.TrySetResult(true);
}
public void Dispose()
{
_cancellationTokenRegistration.Dispose();
_promise.TrySetException(new ObjectDisposedException(nameof(AsyncSignalWaiter)));
}
void Cancel()
{
_promise.TrySetCanceled();
}
}
}
}
+13 -8
View File
@@ -20,7 +20,7 @@ namespace MQTTnet.Internal
new LinkedList<MqttPacketBusItem>()
};
readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0, 1);
readonly AsyncSignal _signal = new AsyncSignal();
readonly object _syncRoot = new object();
int _activePartition = (int)MqttPacketBusPartition.Health;
@@ -82,7 +82,16 @@ namespace MQTTnet.Internal
// No partition contains data so that we have to wait and put
// the worker back to the thread pool.
await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await _signal.WaitAsync(cancellationToken).ConfigureAwait(false);
}
catch (ObjectDisposedException)
{
// The cancelled token should "hide" the disposal of the signal.
cancellationToken.ThrowIfCancellationRequested();
throw;
}
}
cancellationToken.ThrowIfCancellationRequested();
@@ -92,7 +101,7 @@ namespace MQTTnet.Internal
public void Dispose()
{
_semaphore?.Dispose();
_signal.Dispose();
}
public void DropFirstItem(MqttPacketBusPartition partition)
@@ -118,11 +127,7 @@ namespace MQTTnet.Internal
lock (_syncRoot)
{
_partitions[(int)partition].AddLast(item);
if (_semaphore.CurrentCount == 0)
{
_semaphore.Release();
}
_signal.Set();
}
}
+22 -26
View File
@@ -10,40 +10,36 @@ namespace MQTTnet.Internal
{
public sealed class MqttPacketBusItem
{
readonly AsyncTaskCompletionSource<int> _promise = new AsyncTaskCompletionSource<int>();
readonly AsyncTaskCompletionSource<bool> _promise = new AsyncTaskCompletionSource<bool>();
public MqttPacketBusItem(MqttPacket packet)
{
Packet = packet ?? throw new ArgumentNullException(nameof(packet));
}
public event EventHandler Completed;
public MqttPacket Packet { get; }
public event EventHandler Delivered;
public Task WaitForDeliveryAsync()
{
return _promise.Task;
}
public void MarkAsDelivered()
{
if (_promise.TrySetResult(0))
{
Delivered?.Invoke(this, EventArgs.Empty);
}
}
public void MarkAsFailed(Exception exception)
{
if (exception == null) throw new ArgumentNullException(nameof(exception));
_promise.TrySetException(exception);
}
public void MarkAsCancelled()
public void Cancel()
{
_promise.TrySetCanceled();
}
public void Complete()
{
_promise.TrySetResult(true);
Completed?.Invoke(this, EventArgs.Empty);
}
public void Fail(Exception exception)
{
_promise.TrySetException(exception);
}
public Task WaitAsync()
{
return _promise.Task;
}
}
}
@@ -2,6 +2,7 @@
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cconnecting/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cdiagnostics/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cdisconnecting/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cexceptions/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cextendedauthenticationexchange/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Coptions/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=client_005Cpublishing/@EntryIndexedValue">True</s:Boolean>
@@ -5,13 +5,78 @@
using System;
using System.Text;
// ReSharper disable MemberCanBePrivate.Global
namespace MQTTnet
{
public static class MqttApplicationMessageExtensions
{
public static short ConvertPayloadToByte(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
if (applicationMessage.Payload.Length != 1)
{
throw new ArgumentException("Payload requires to be 1 bytes long.");
}
return applicationMessage.Payload[0];
}
public static short ConvertPayloadToInt16(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
if (applicationMessage.Payload.Length != 2)
{
throw new ArgumentException("Payload requires to be 2 bytes long.");
}
return BitConverter.ToInt16(applicationMessage.Payload, 0);
}
public static int ConvertPayloadToInt32(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
if (applicationMessage.Payload?.Length != 4)
{
throw new ArgumentException("Payload requires to be 4 bytes long.");
}
return BitConverter.ToInt32(applicationMessage.Payload, 0);
}
public static long ConvertPayloadToInt64(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
if (applicationMessage.Payload.Length != 8)
{
throw new ArgumentException("Payload requires to be 8 bytes long.");
}
return BitConverter.ToInt64(applicationMessage.Payload, 0);
}
public static string ConvertPayloadToString(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
if (applicationMessage.Payload == null)
{
@@ -25,5 +90,65 @@ namespace MQTTnet
return Encoding.UTF8.GetString(applicationMessage.Payload, 0, applicationMessage.Payload.Length);
}
public static decimal ParsePayloadToDecimal(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
return decimal.Parse(ConvertPayloadToString(applicationMessage));
}
public static double ParsePayloadToDouble(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
return double.Parse(ConvertPayloadToString(applicationMessage));
}
public static short ParsePayloadToInt16(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
return short.Parse(ConvertPayloadToString(applicationMessage));
}
public static int ParsePayloadToInt32(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
return int.Parse(ConvertPayloadToString(applicationMessage));
}
public static long ParsePayloadToInt64(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
return long.Parse(ConvertPayloadToString(applicationMessage));
}
public static float ParsePayloadToSingle(this MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
return float.Parse(ConvertPayloadToString(applicationMessage));
}
}
}
}
@@ -2,48 +2,103 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using MQTTnet.Packets;
using System;
using System.Collections.Generic;
using MQTTnet.Packets;
namespace MQTTnet.PacketDispatcher
{
public sealed class MqttPacketDispatcher
public sealed class MqttPacketDispatcher : IDisposable
{
readonly List<IMqttPacketAwaitable> _awaitables = new List<IMqttPacketAwaitable>();
readonly List<IMqttPacketAwaitable> _waiters = new List<IMqttPacketAwaitable>();
public void FailAll(Exception exception)
bool _isDisposed;
public MqttPacketAwaitable<TResponsePacket> AddAwaitable<TResponsePacket>(ushort packetIdentifier) where TResponsePacket : MqttPacket
{
if (exception == null) throw new ArgumentNullException(nameof(exception));
var awaitable = new MqttPacketAwaitable<TResponsePacket>(packetIdentifier, this);
lock (_awaitables)
lock (_waiters)
{
foreach (var awaitable in _awaitables)
{
awaitable.Fail(exception);
}
_awaitables.Clear();
_waiters.Add(awaitable);
}
return awaitable;
}
public void CancelAll()
{
lock (_awaitables)
lock (_waiters)
{
foreach (var awaitable in _awaitables)
foreach (var awaitable in _waiters)
{
awaitable.Cancel();
}
_awaitables.Clear();
_waiters.Clear();
}
}
public void Dispose()
{
Dispose(new ObjectDisposedException(nameof(MqttPacketDispatcher)));
}
public void Dispose(Exception exception)
{
if (exception == null)
{
throw new ArgumentNullException(nameof(exception));
}
lock (_waiters)
{
FailAll(exception);
// Make sure that no task can start waiting after this instance is already disposed.
// This will prevent unexpected freezes.
_isDisposed = true;
}
}
public void FailAll(Exception exception)
{
if (exception == null)
{
throw new ArgumentNullException(nameof(exception));
}
lock (_waiters)
{
foreach (var awaitable in _waiters)
{
awaitable.Fail(exception);
}
_waiters.Clear();
}
}
public void RemoveAwaitable(IMqttPacketAwaitable awaitable)
{
if (awaitable == null)
{
throw new ArgumentNullException(nameof(awaitable));
}
lock (_waiters)
{
_waiters.Remove(awaitable);
}
}
public bool TryDispatch(MqttPacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (packet == null)
{
throw new ArgumentNullException(nameof(packet));
}
ushort identifier = 0;
if (packet is MqttPacketWithIdentifier packetWithIdentifier)
{
@@ -51,13 +106,15 @@ namespace MQTTnet.PacketDispatcher
}
var packetType = packet.GetType();
var awaitables = new List<IMqttPacketAwaitable>();
lock (_awaitables)
var waiters = new List<IMqttPacketAwaitable>();
lock (_waiters)
{
for (var i = _awaitables.Count - 1; i >= 0; i--)
ThrowIfDisposed();
for (var i = _waiters.Count - 1; i >= 0; i--)
{
var entry = _awaitables[i];
var entry = _waiters[i];
// Note: The PingRespPacket will also arrive here and has NO identifier but there
// is code which waits for it. So the code must be able to deal with filters which
@@ -66,39 +123,25 @@ namespace MQTTnet.PacketDispatcher
{
continue;
}
awaitables.Add(entry);
_awaitables.RemoveAt(i);
waiters.Add(entry);
_waiters.RemoveAt(i);
}
}
foreach (var matchingEntry in awaitables)
foreach (var matchingEntry in waiters)
{
matchingEntry.Complete(packet);
}
return awaitables.Count > 0;
}
public MqttPacketAwaitable<TResponsePacket> AddAwaitable<TResponsePacket>(ushort packetIdentifier) where TResponsePacket : MqttPacket
{
var awaitable = new MqttPacketAwaitable<TResponsePacket>(packetIdentifier, this);
lock (_awaitables)
{
_awaitables.Add(awaitable);
}
return awaitable;
return waiters.Count > 0;
}
public void RemoveAwaitable(IMqttPacketAwaitable awaitable)
void ThrowIfDisposed()
{
if (awaitable == null) throw new ArgumentNullException(nameof(awaitable));
lock (_awaitables)
if (_isDisposed)
{
_awaitables.Remove(awaitable);
throw new ObjectDisposedException(nameof(MqttPacketDispatcher));
}
}
}
@@ -7,7 +7,12 @@ namespace MQTTnet.Protocol
public enum MqttPubAckReasonCode
{
Success = 0,
/// <summary>
/// The message is accepted but there are no subscribers. This is sent only by the Server. If the Server knows that there are no matching subscribers, it MAY use this Reason Code instead of 0x00 (Success).
/// </summary>
NoMatchingSubscribers = 16,
UnspecifiedError = 128,
ImplementationSpecificError = 131,
NotAuthorized = 135,
@@ -10,15 +10,23 @@ namespace MQTTnet.Server
{
public sealed class ClientAcknowledgedPublishPacketEventArgs : EventArgs
{
public ClientAcknowledgedPublishPacketEventArgs(string clientId, IDictionary sessionItems, MqttPublishPacket publishPacket, MqttPacketWithIdentifier acknowledgePacket)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
PublishPacket = publishPacket ?? throw new ArgumentNullException(nameof(publishPacket));
AcknowledgePacket = acknowledgePacket ?? throw new ArgumentNullException(nameof(acknowledgePacket));
}
/// <summary>
/// Gets the packet which was used for acknowledge. This can be a PubAck or PubComp packet.
/// </summary>
public MqttPacketWithIdentifier AcknowledgePacket { get; internal set; }
public MqttPacketWithIdentifier AcknowledgePacket { get; }
/// <summary>
/// Gets the ID of the client which acknowledged a PUBLISH packet.
/// </summary>
public string ClientId { get; internal set; }
public string ClientId { get; }
/// <summary>
/// Gets whether the PUBLISH packet is fully acknowledged. This is the case for PUBACK (QoS 1) and PUBCOMP (QoS 2.
@@ -28,11 +36,11 @@ namespace MQTTnet.Server
/// <summary>
/// Gets the PUBLISH packet which was acknowledged.
/// </summary>
public MqttPublishPacket PublishPacket { get; internal set; }
public MqttPublishPacket PublishPacket { get; }
/// <summary>
/// Gets the session items which contain custom user data per session.
/// </summary>
public IDictionary SessionItems { get; internal set; }
public IDictionary SessionItems { get; }
}
}
@@ -0,0 +1,16 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
namespace MQTTnet.Server
{
public sealed class DispatchApplicationMessageResult
{
public DispatchApplicationMessageResult(int matchingSubscribersCount)
{
MatchingSubscribersCount = matchingSubscribersCount;
}
public int MatchingSubscribersCount { get; }
}
}
+37 -41
View File
@@ -11,7 +11,6 @@ using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Implementations;
using MQTTnet.Internal;
using MQTTnet.PacketDispatcher;
using MQTTnet.Packets;
@@ -34,6 +33,7 @@ namespace MQTTnet.Server
readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
CancellationTokenSource _cancellationToken;
bool _disconnectPacketSent;
public MqttClient(
MqttConnectPacket connectPacket,
@@ -147,19 +147,22 @@ namespace MQTTnet.Server
{
IsRunning = false;
// Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+.
if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
if (!_disconnectPacketSent)
{
// The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
// This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding
// a configuration option is requested in the future.
if (reason != MqttDisconnectReasonCode.NormalDisconnection)
// Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+.
if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
{
// Is is very important to send the DISCONNECT packet here BEFORE cancelling the
// token because the entire connection is closed (disposed) as soon as the cancellation
// token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
// at the client!
await TrySendDisconnectPacket(reason).ConfigureAwait(false);
// The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
// This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding
// a configuration option is requested in the future.
if (reason != MqttDisconnectReasonCode.NormalDisconnection)
{
// Is is very important to send the DISCONNECT packet here BEFORE cancelling the
// token because the entire connection is closed (disposed) as soon as the cancellation
// token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
// at the client!
await TrySendDisconnectPacket(reason).ConfigureAwait(false);
}
}
}
@@ -170,14 +173,7 @@ namespace MQTTnet.Server
{
if (_eventContainer.ClientAcknowledgedPublishPacketEvent.HasHandlers)
{
var eventArgs = new ClientAcknowledgedPublishPacketEventArgs
{
PublishPacket = publishPacket,
AcknowledgePacket = acknowledgePacket,
ClientId = Id,
SessionItems = Session.Items
};
var eventArgs = new ClientAcknowledgedPublishPacketEventArgs(Id, Session.Items, publishPacket, acknowledgePacket);
return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger);
}
@@ -241,9 +237,10 @@ namespace MQTTnet.Server
return;
}
DispatchApplicationMessageResult dispatchResult = null;
if (processPublish && applicationMessage != null)
{
await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false);
dispatchResult = await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false);
}
switch (publishPacket.QualityOfServiceLevel)
@@ -255,13 +252,13 @@ namespace MQTTnet.Server
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs);
var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs, dispatchResult);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket));
break;
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs);
var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs, dispatchResult);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket));
break;
}
@@ -272,17 +269,14 @@ namespace MQTTnet.Server
}
}
async Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
{
var acknowledgedPublishPacket = Session.PeekAcknowledgePublishPacket(pubRecPacket.PacketIdentifier);
if (acknowledgedPublishPacket != null)
{
await ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubRecPacket).ConfigureAwait(false);
}
// Do not fire the event _ClientAcknowledgedPublishPacket_ here because the QoS 2 process is only finished
// properly when the client has sent the PUBCOMP packet.
var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
return CompletedTask.Instance;
}
void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)
@@ -293,7 +287,7 @@ namespace MQTTnet.Server
async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{
var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
var subscribeResult = await Session.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult);
@@ -307,18 +301,17 @@ namespace MQTTnet.Server
if (subscribeResult.RetainedMessages != null)
{
foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages)
foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
{
var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
var publishPacket = _packetFactories.Publish.Create(retainedMessageMatch);
Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
}
}
async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
{
var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false);
var unsubscribeResult = await Session.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false);
var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult);
@@ -496,7 +489,7 @@ namespace MQTTnet.Server
try
{
while (!cancellationToken.IsCancellationRequested)
while (!cancellationToken.IsCancellationRequested && !IsTakenOver && IsRunning)
{
packetBusItem = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false);
@@ -514,15 +507,15 @@ namespace MQTTnet.Server
try
{
await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false);
packetBusItem.MarkAsDelivered();
packetBusItem.Complete();
}
catch (OperationCanceledException)
{
packetBusItem.MarkAsCancelled();
packetBusItem.Cancel();
}
catch (Exception exception)
{
packetBusItem.MarkAsFailed(exception);
packetBusItem.Fail(exception);
}
finally
{
@@ -570,7 +563,7 @@ namespace MQTTnet.Server
catch (ObjectDisposedException)
{
// This can happen when connections are created and dropped very quickly.
// It is not an issue if the cancellation token cannot be cancelled multiple times.
// It is not an issue if the cancellation token cannot be canceled multiple times.
}
}
@@ -578,6 +571,9 @@ namespace MQTTnet.Server
{
try
{
// This also indicates that it was tried at least!
_disconnectPacketSent = true;
var disconnectPacket = _packetFactories.Disconnect.Create(reasonCode);
using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
@@ -127,21 +127,20 @@ namespace MQTTnet.Server
_logger.Verbose("Session for client '{0}' deleted.", clientId);
}
public async Task DispatchApplicationMessage(string senderId, MqttApplicationMessage applicationMessage)
public async Task<DispatchApplicationMessageResult> DispatchApplicationMessage(string senderId, MqttApplicationMessage applicationMessage)
{
var matchingSubscribersCount = 0;
try
{
if (applicationMessage.Retain)
{
await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false);
}
var deliveryCount = 0;
List<MqttSession> subscriberSessions;
lock (_sessionsManagementLock)
{
// only subscriber clients are of interest here.
subscriberSessions = _subscriberSessions.ToList();
}
@@ -150,11 +149,17 @@ namespace MQTTnet.Server
foreach (var session in subscriberSessions)
{
var checkSubscriptionsResult = session.SubscriptionsManager.CheckSubscriptions(
if(!session.TryCheckSubscriptions(
applicationMessage.Topic,
topicHash,
applicationMessage.QualityOfServiceLevel,
senderId);
senderId,
out var checkSubscriptionsResult))
{
// Checking the subscriptions has failed for the session. The session
// will be ignored.
continue;
}
if (!checkSubscriptionsResult.IsSubscribed)
{
@@ -181,17 +186,22 @@ namespace MQTTnet.Server
}
session.EnqueueDataPacket(new MqttPacketBusItem(newPublishPacket));
deliveryCount++;
matchingSubscribersCount++;
_logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'.", session.Id, applicationMessage.Topic);
}
await FireApplicationMessageNotConsumedEvent(applicationMessage, deliveryCount, senderId);
if (matchingSubscribersCount == 0)
{
await FireApplicationMessageNotConsumedEvent(applicationMessage, senderId).ConfigureAwait(false);
}
}
catch (Exception exception)
{
_logger.Error(exception, "Unhandled exception while processing next queued application message.");
}
return new DispatchApplicationMessageResult(matchingSubscribersCount);
}
public void Dispose()
@@ -301,7 +311,7 @@ namespace MQTTnet.Server
channelAdapter.Endpoint,
client.Session.Items);
await _eventContainer.ClientConnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
await _eventContainer.ClientConnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
}
await client.RunAsync().ConfigureAwait(false);
@@ -414,13 +424,13 @@ namespace MQTTnet.Server
var clientSession = GetClientSession(clientId);
var subscribeResult = await clientSession.SubscriptionsManager.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false);
var subscribeResult = await clientSession.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false);
if (subscribeResult.RetainedMessages != null)
{
foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages)
foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
{
var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
var publishPacket = _packetFactories.Publish.Create(retainedMessageMatch);
clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
}
@@ -441,7 +451,7 @@ namespace MQTTnet.Server
var fakeUnsubscribePacket = new MqttUnsubscribePacket();
fakeUnsubscribePacket.TopicFilters.AddRange(topicFilters);
return GetClientSession(clientId).SubscriptionsManager.Unsubscribe(fakeUnsubscribePacket, CancellationToken.None);
return GetClientSession(clientId).Unsubscribe(fakeUnsubscribePacket, CancellationToken.None);
}
MqttClient CreateClient(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session)
@@ -484,10 +494,15 @@ namespace MQTTnet.Server
using (await _createConnectionSyncRoot.EnterAsync().ConfigureAwait(false))
{
MqttSession session;
MqttSession oldSession;
MqttClient oldClient;
lock (_sessionsManagementLock)
{
if (!_sessions.TryGetValue(connectPacket.ClientId, out session))
MqttSession session;
// Create a new session (if required).
if (!_sessions.TryGetValue(connectPacket.ClientId, out oldSession))
{
session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist);
}
@@ -496,11 +511,15 @@ namespace MQTTnet.Server
if (connectPacket.CleanSession)
{
_logger.Verbose("Deleting existing session of client '{0}' due to clean start.", connectPacket.ClientId);
_subscriberSessions.Remove(oldSession);
session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist);
}
else
{
_logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
session = oldSession;
oldSession = null;
// Session persistence could change for MQTT 5 clients that reconnect with different SessionExpiryInterval
session.IsPersistent = sessionShouldPersist;
connAckPacket.IsSessionPresent = true;
@@ -509,41 +528,45 @@ namespace MQTTnet.Server
}
_sessions[connectPacket.ClientId] = session;
// Create a new client (always required).
_clients.TryGetValue(connectPacket.ClientId, out oldClient);
if (oldClient != null)
{
// This will stop the current client from sending and receiving but remains the connection active
// for a later DISCONNECT packet.
oldClient.IsTakenOver = true;
}
client = CreateClient(connectPacket, channelAdapter, session);
_clients[connectPacket.ClientId] = client;
}
if (!connAckPacket.IsSessionPresent)
{
// TODO: This event is not yet final. It can already be used but restoring sessions from storage will be added later!
var preparingSessionEventArgs = new PreparingSessionEventArgs();
await _eventContainer.PreparingSessionEvent.InvokeAsync(preparingSessionEventArgs).ConfigureAwait(false);
await _eventContainer.PreparingSessionEvent.TryInvokeAsync(preparingSessionEventArgs, _logger).ConfigureAwait(false);
}
MqttClient existingClient;
lock (_clients)
if (oldClient != null)
{
_clients.TryGetValue(connectPacket.ClientId, out existingClient);
client = CreateClient(connectPacket, channelAdapter, session);
await oldClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false);
_clients[connectPacket.ClientId] = client;
}
if (existingClient != null)
{
existingClient.IsTakenOver = true;
await existingClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false);
if (_eventContainer.ClientConnectedEvent.HasHandlers)
if (_eventContainer.ClientDisconnectedEvent.HasHandlers)
{
var eventArgs = new ClientDisconnectedEventArgs(
existingClient.Id,
oldClient.Id,
MqttClientDisconnectType.Takeover,
existingClient.Endpoint,
existingClient.Session.Items);
oldClient.Endpoint,
oldClient.Session.Items);
await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
await _eventContainer.ClientDisconnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
}
}
oldSession?.Dispose();
}
return client;
@@ -556,13 +579,8 @@ namespace MQTTnet.Server
return new MqttSession(clientId, isPersistent, sessionItems, _options, _eventContainer, _retainedMessagesManager, this);
}
async Task FireApplicationMessageNotConsumedEvent(MqttApplicationMessage applicationMessage, int deliveryCount, string senderId)
async Task FireApplicationMessageNotConsumedEvent(MqttApplicationMessage applicationMessage, string senderId)
{
if (deliveryCount > 0)
{
return;
}
if (!_eventContainer.ApplicationMessageNotConsumedEvent.HasHandlers)
{
return;
@@ -7,6 +7,7 @@ using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -30,7 +31,7 @@ namespace MQTTnet.Server
readonly Dictionary<string, MqttSubscription> _subscriptions = new Dictionary<string, MqttSubscription>();
// Use subscription lock to maintain consistency across subscriptions and topic hash dictionaries
readonly SemaphoreSlim _subscriptionsLock = new SemaphoreSlim(1);
readonly AsyncLock _subscriptionsLock = new AsyncLock();
readonly Dictionary<ulong, TopicHashMaskSubscriptions> _wildcardSubscriptionsByTopicHash = new Dictionary<ulong, TopicHashMaskSubscriptions>();
public MqttClientSubscriptionsManager(
@@ -45,13 +46,12 @@ namespace MQTTnet.Server
_subscriptionChangedNotification = subscriptionChangedNotification;
}
public CheckSubscriptionsResult CheckSubscriptions(string topic, ulong topicHash, MqttQualityOfServiceLevel applicationMessageQoSLevel, string senderClientId)
public CheckSubscriptionsResult CheckSubscriptions(string topic, ulong topicHash, MqttQualityOfServiceLevel qualityOfServiceLevel, string senderId)
{
var possibleSubscriptions = new List<MqttSubscription>();
// Check for possible subscriptions. They might have collisions but this is fine.
_subscriptionsLock.Wait();
try
using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult())
{
if (_noWildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var noWildcardSubscriptions))
{
@@ -63,18 +63,14 @@ namespace MQTTnet.Server
var wildcardSubscriptions = wcs.Value;
var subscriptionHash = wcs.Key;
var subscriptionHashMask = wildcardSubscriptions.HashMask;
if ((topicHash & subscriptionHashMask) == subscriptionHash)
{
possibleSubscriptions.AddRange(wildcardSubscriptions.Subscriptions.ToList());
}
}
}
finally
{
_subscriptionsLock.Release();
}
// The pre check has evaluated that nothing is subscribed.
// If there were some possible candidates they get checked below
// again to avoid collisions.
@@ -83,7 +79,7 @@ namespace MQTTnet.Server
return CheckSubscriptionsResult.NotSubscribed;
}
var senderIsReceiver = string.Equals(senderClientId, _session.Id);
var senderIsReceiver = string.Equals(senderId, _session.Id);
var maxQoSLevel = -1; // Not subscribed.
HashSet<uint> subscriptionIdentifiers = null;
@@ -136,7 +132,7 @@ namespace MQTTnet.Server
SubscriptionIdentifiers = subscriptionIdentifiers?.ToList() ?? EmptySubscriptionIdentifiers,
// Start with the same QoS as the publisher.
QualityOfServiceLevel = applicationMessageQoSLevel
QualityOfServiceLevel = qualityOfServiceLevel
};
// Now downgrade if required.
@@ -148,7 +144,7 @@ namespace MQTTnet.Server
// Subscribing to a Topic Filter at QoS 2 is equivalent to saying "I would like to receive Messages matching this filter at the QoS with which they were published".
// This means a publisher is responsible for determining the maximum QoS a Message can be delivered at, but a subscriber is able to require that the Server
// downgrades the QoS to one more suitable for its usage.
if (maxQoSLevel < (int)applicationMessageQoSLevel)
if (maxQoSLevel < (int)qualityOfServiceLevel)
{
result.QualityOfServiceLevel = (MqttQualityOfServiceLevel)maxQoSLevel;
}
@@ -169,10 +165,7 @@ namespace MQTTnet.Server
}
var retainedApplicationMessages = await _retainedMessagesManager.GetMessages().ConfigureAwait(false);
var result = new SubscribeResult
{
ReasonCodes = new List<MqttSubscribeReasonCode>(subscribePacket.TopicFilters.Count)
};
var result = new SubscribeResult(subscribePacket.TopicFilters.Count);
var addedSubscriptions = new List<string>();
var finalTopicFilters = new List<MqttTopicFilter>();
@@ -212,7 +205,7 @@ namespace MQTTnet.Server
// This call will add the new subscription to the internal storage.
// So the event _ClientSubscribedTopicEvent_ must be called afterwards.
_subscriptionChangedNotification?.OnSubscriptionsAdded(_session, addedSubscriptions);
if (_eventContainer.ClientSubscribedTopicEvent.HasHandlers)
{
foreach (var finalTopicFilter in finalTopicFilters)
@@ -221,23 +214,22 @@ namespace MQTTnet.Server
await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
}
return result;
}
public async Task<MqttUnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
public async Task<UnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
{
if (unsubscribePacket == null)
{
throw new ArgumentNullException(nameof(unsubscribePacket));
}
var result = new MqttUnsubscribeResult();
var result = new UnsubscribeResult();
var removedSubscriptions = new List<string>();
await _subscriptionsLock.WaitAsync(cancellationToken).ConfigureAwait(false);
try
using (await _subscriptionsLock.EnterAsync(cancellationToken).ConfigureAwait(false))
{
foreach (var topicFilter in unsubscribePacket.TopicFilters)
{
@@ -268,7 +260,7 @@ namespace MQTTnet.Server
if (existingSubscription != null)
{
var topicHash = existingSubscription.TopicHash;
if (existingSubscription.TopicHasWildcard)
{
if (_wildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subscriptions))
@@ -297,11 +289,8 @@ namespace MQTTnet.Server
}
}
}
finally
{
_subscriptionsLock.Release();
_subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions);
}
_subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions);
if (_eventContainer.ClientUnsubscribedTopicEvent.HasHandlers)
{
@@ -348,8 +337,7 @@ namespace MQTTnet.Server
// Add to subscriptions and maintain topic hash dictionaries
_subscriptionsLock.Wait();
try
using (_subscriptionsLock.EnterAsync(CancellationToken.None).GetAwaiter().GetResult())
{
MqttSubscription.CalculateTopicHash(topicFilter.Topic, out var topicHash, out var topicHashMask, out var hasWildcard);
@@ -399,10 +387,6 @@ namespace MQTTnet.Server
subscriptions.Add(subscription);
}
}
finally
{
_subscriptionsLock.Release();
}
return new CreateSubscriptionResult
{
@@ -412,14 +396,14 @@ namespace MQTTnet.Server
}
static void FilterRetainedApplicationMessages(
IList<MqttApplicationMessage> retainedApplicationMessages,
IList<MqttApplicationMessage> retainedMessages,
CreateSubscriptionResult createSubscriptionResult,
SubscribeResult subscribeResult)
{
for (var index = retainedApplicationMessages.Count - 1; index >= 0; index--)
for (var index = retainedMessages.Count - 1; index >= 0; index--)
{
var retainedApplicationMessage = retainedApplicationMessages[index];
if (retainedApplicationMessage == null)
var retainedMessage = retainedMessages[index];
if (retainedMessage == null)
{
continue;
}
@@ -436,16 +420,21 @@ namespace MQTTnet.Server
continue;
}
if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, createSubscriptionResult.Subscription.Topic) != MqttTopicFilterCompareResult.IsMatch)
if (MqttTopicFilterComparer.Compare(retainedMessage.Topic, createSubscriptionResult.Subscription.Topic) != MqttTopicFilterCompareResult.IsMatch)
{
continue;
}
var retainedMessageMatch = new MqttRetainedMessageMatch
var retainedMessageMatch = new MqttRetainedMessageMatch(retainedMessage, createSubscriptionResult.Subscription.GrantedQualityOfServiceLevel);
if (retainedMessageMatch.SubscriptionQualityOfServiceLevel > retainedMessageMatch.ApplicationMessage.QualityOfServiceLevel)
{
ApplicationMessage = retainedApplicationMessage,
SubscriptionQualityOfServiceLevel = createSubscriptionResult.Subscription.GrantedQualityOfServiceLevel
};
// UPGRADING the QoS is not allowed!
// From MQTT spec: Subscribing to a Topic Filter at QoS 2 is equivalent to saying
// "I would like to receive Messages matching this filter at the QoS with which they were published".
// This means a publisher is responsible for determining the maximum QoS a Message can be delivered at,
// but a subscriber is able to require that the Server downgrades the QoS to one more suitable for its usage.
retainedMessageMatch.SubscriptionQualityOfServiceLevel = retainedMessageMatch.ApplicationMessage.QualityOfServiceLevel;
}
if (subscribeResult.RetainedMessages == null)
{
@@ -456,7 +445,7 @@ namespace MQTTnet.Server
// Clear the retained message from the list because the client should receive every message only
// one time even if multiple subscriptions affect them.
retainedApplicationMessages[index] = null;
retainedMessages[index] = null;
}
}
@@ -491,23 +480,20 @@ namespace MQTTnet.Server
async Task<InterceptingUnsubscriptionEventArgs> InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken)
{
var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, topicFilter, _session.Items, topicFilter);
if (mqttSubscription == null)
var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, topicFilter, _session.Items, topicFilter)
{
clientUnsubscribingTopicEventArgs.Response.ReasonCode = MqttUnsubscribeReasonCode.NoSubscriptionExisted;
}
else
{
clientUnsubscribingTopicEventArgs.Response.ReasonCode = MqttUnsubscribeReasonCode.Success;
}
Response =
{
ReasonCode = mqttSubscription == null ? MqttUnsubscribeReasonCode.NoSubscriptionExisted : MqttUnsubscribeReasonCode.Success
}
};
await _eventContainer.InterceptingUnsubscriptionEvent.InvokeAsync(clientUnsubscribingTopicEventArgs).ConfigureAwait(false);
return clientUnsubscribingTopicEventArgs;
}
public sealed class CreateSubscriptionResult
sealed class CreateSubscriptionResult
{
public bool IsNewSubscription { get; set; }
+78 -18
View File
@@ -4,7 +4,6 @@
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
@@ -23,8 +22,10 @@ namespace MQTTnet.Server
readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
readonly MqttServerOptions _serverOptions;
readonly MqttClientSubscriptionsManager _subscriptionsManager;
readonly ConcurrentDictionary<ushort, MqttPublishPacket> _unacknowledgedPublishPackets = new ConcurrentDictionary<ushort, MqttPublishPacket>();
// Do not use a dictionary in order to keep the ordering of the messages.
readonly List<MqttPublishPacket> _unacknowledgedPublishPackets = new List<MqttPublishPacket>();
// Bookkeeping to know if this is a subscribing client; lazy initialize later.
HashSet<string> _subscribedTopics;
@@ -45,7 +46,7 @@ namespace MQTTnet.Server
_serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
_clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));
SubscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager, clientSessionsManager);
_subscriptionsManager = new MqttClientSubscriptionsManager(this, eventContainer, retainedMessagesManager, clientSessionsManager);
}
public DateTime CreatedTimestamp { get; } = DateTime.UtcNow;
@@ -67,21 +68,18 @@ namespace MQTTnet.Server
public long PendingDataPacketsCount => _packetBus.PartitionItemsCount(MqttPacketBusPartition.Data);
public MqttClientSubscriptionsManager SubscriptionsManager { get; }
public bool WillMessageSent { get; set; }
public MqttPublishPacket PeekAcknowledgePublishPacket(ushort packetIdentifier)
{
// This will only return the matching PUBLISH packet but does not remove it.
// This is required for QoS 2.
_unacknowledgedPublishPackets.TryGetValue(packetIdentifier, out var publishPacket);
return publishPacket;
}
public MqttPublishPacket AcknowledgePublishPacket(ushort packetIdentifier)
{
_unacknowledgedPublishPackets.TryRemove(packetIdentifier, out var publishPacket);
MqttPublishPacket publishPacket;
lock (_unacknowledgedPublishPackets)
{
publishPacket = _unacknowledgedPublishPackets.FirstOrDefault(p => p.PacketIdentifier.Equals(packetIdentifier));
_unacknowledgedPublishPackets.Remove(publishPacket);
}
return publishPacket;
}
@@ -107,8 +105,8 @@ namespace MQTTnet.Server
public void Dispose()
{
_packetBus?.Dispose();
SubscriptionsManager.Dispose();
_packetBus.Dispose();
_subscriptionsManager.Dispose();
}
public void EnqueueControlPacket(MqttPacketBusItem packetBusItem)
@@ -139,7 +137,10 @@ namespace MQTTnet.Server
{
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
_unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket;
lock (_unacknowledgedPublishPackets)
{
_unacknowledgedPublishPackets.Add(publishPacket);
}
}
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data);
@@ -150,14 +151,48 @@ namespace MQTTnet.Server
_packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health);
}
public MqttPublishPacket PeekAcknowledgePublishPacket(ushort packetIdentifier)
{
// This will only return the matching PUBLISH packet but does not remove it.
// This is required for QoS 2.
lock (_unacknowledgedPublishPackets)
{
return _unacknowledgedPublishPackets.FirstOrDefault(p => p.PacketIdentifier.Equals(packetIdentifier));
}
}
public void Recover()
{
// TODO: Keep the bus and only insert pending items again.
// TODO: Check if packet identifier must be restarted or not.
// TODO: Recover package identifier.
/*
The Session state in the Client consists of:
· QoS 1 and QoS 2 messages which have been sent to the Server, but have not been completely acknowledged.
· QoS 2 messages which have been received from the Server, but have not been completely acknowledged.
The Session state in the Server consists of:
· The existence of a Session, even if the rest of the Session state is empty.
· The Client’s subscriptions.
· QoS 1 and QoS 2 messages which have been sent to the Client, but have not been completely acknowledged.
· QoS 1 and QoS 2 messages pending transmission to the Client.
· QoS 2 messages which have been received from the Client, but have not been completely acknowledged.
· Optionally, QoS 0 messages pending transmission to the Client.
*/
// Create a copy of all currently unacknowledged publish packets and clear the storage.
// We must re-enqueue them in order to trigger other code.
List<MqttPublishPacket> unacknowledgedPublishPackets;
lock (_unacknowledgedPublishPackets)
{
unacknowledgedPublishPackets = _unacknowledgedPublishPackets.ToList();
_unacknowledgedPublishPackets.Clear();
}
_packetBus.Clear();
foreach (var publishPacket in _unacknowledgedPublishPackets.Values.ToList())
foreach (var publishPacket in unacknowledgedPublishPackets)
{
EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
}
@@ -167,5 +202,30 @@ namespace MQTTnet.Server
{
_subscribedTopics?.Remove(topic);
}
public Task<SubscribeResult> Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{
return _subscriptionsManager.Subscribe(subscribePacket, cancellationToken);
}
public bool TryCheckSubscriptions(string topic, ulong topicHash, MqttQualityOfServiceLevel qualityOfServiceLevel, string senderId, out CheckSubscriptionsResult result)
{
result = null;
try
{
result = _subscriptionsManager.CheckSubscriptions(topic, topicHash, qualityOfServiceLevel, senderId);
return true;
}
catch
{
return false;
}
}
public Task<UnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
{
return _subscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken);
}
}
}
@@ -179,7 +179,7 @@ namespace MQTTnet.Server
RetainAsPublished = retainAsPublished;
GrantedQualityOfServiceLevel = qualityOfServiceLevel;
Identifier = identifier;
CalculateTopicHash(Topic, out var hash, out var hashMask, out var hasWildcard);
TopicHash = hash;
TopicHashMask = hashMask;
@@ -10,8 +10,13 @@ namespace MQTTnet.Server
{
public sealed class SubscribeResult
{
public SubscribeResult(int topicsCount)
{
ReasonCodes = new List<MqttSubscribeReasonCode>(topicsCount);
}
public bool CloseConnection { get; set; }
public List<MqttSubscribeReasonCode> ReasonCodes { get; set; }
public string ReasonString { get; set; }
@@ -7,7 +7,7 @@ using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public sealed class MqttUnsubscribeResult
public sealed class UnsubscribeResult
{
public List<MqttUnsubscribeReasonCode> ReasonCodes { get; } = new List<MqttUnsubscribeReasonCode>(128);
@@ -2,13 +2,20 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public sealed class MqttRetainedMessageMatch
{
public MqttApplicationMessage ApplicationMessage { get; set; }
public MqttRetainedMessageMatch(MqttApplicationMessage applicationMessage, MqttQualityOfServiceLevel subscriptionQualityOfServiceLevel)
{
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
SubscriptionQualityOfServiceLevel = subscriptionQualityOfServiceLevel;
}
public MqttApplicationMessage ApplicationMessage { get; }
public MqttQualityOfServiceLevel SubscriptionQualityOfServiceLevel { get; set; }
}
+69 -9
View File
@@ -3,29 +3,89 @@
// See the LICENSE file in the project root for more information.
using System;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public static class MqttServerExtensions
{
public static Task InjectApplicationMessage(
this MqttServer server,
string topic,
string payload = null,
MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
bool retain = false)
{
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
if (topic == null)
{
throw new ArgumentNullException(nameof(topic));
}
var payloadBuffer = EmptyBuffer.Array;
if (payload is string stringPayload)
{
payloadBuffer = Encoding.UTF8.GetBytes(stringPayload);
}
return server.InjectApplicationMessage(
new InjectedMqttApplicationMessage(
new MqttApplicationMessage
{
Topic = topic,
Payload = payloadBuffer,
QualityOfServiceLevel = qualityOfServiceLevel,
Retain = retain
}));
}
public static Task SubscribeAsync(this MqttServer server, string clientId, params MqttTopicFilter[] topicFilters)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
if (clientId == null)
{
throw new ArgumentNullException(nameof(clientId));
}
if (topicFilters == null)
{
throw new ArgumentNullException(nameof(topicFilters));
}
return server.SubscribeAsync(clientId, topicFilters);
}
public static Task SubscribeAsync(this MqttServer server, string clientId, string topic)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topic == null) throw new ArgumentNullException(nameof(topic));
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).Build());
if (clientId == null)
{
throw new ArgumentNullException(nameof(clientId));
}
if (topic == null)
{
throw new ArgumentNullException(nameof(topic));
}
var topicFilters = new MqttTopicFilterBuilder().WithTopic(topic).Build();
return server.SubscribeAsync(clientId, topicFilters);
}
}
}
}
@@ -45,7 +45,7 @@ namespace MQTTnet.Server
var packetBusItem = new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage));
_session.EnqueueDataPacket(packetBusItem);
return packetBusItem.WaitForDeliveryAsync();
return packetBusItem.WaitAsync();
}
public Task DeleteAsync()