diff --git a/Source/MQTTnet/PacketDispatcher/MqttPacketAwaitable.cs b/Source/MQTTnet/PacketDispatcher/MqttPacketAwaitable.cs index de0df74d..70fbd29f 100644 --- a/Source/MQTTnet/PacketDispatcher/MqttPacketAwaitable.cs +++ b/Source/MQTTnet/PacketDispatcher/MqttPacketAwaitable.cs @@ -3,12 +3,13 @@ using MQTTnet.Packets; using System; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Internal; namespace MQTTnet.PacketDispatcher { public sealed class MqttPacketAwaitable : IMqttPacketAwaitable where TPacket : MqttBasePacket { - readonly TaskCompletionSource _taskCompletionSource; + readonly CrossPlatformPromise _promise = new CrossPlatformPromise(); readonly MqttPacketDispatcher _owningPacketDispatcher; public MqttPacketAwaitable(ushort packetIdentifier, MqttPacketDispatcher owningPacketDispatcher) @@ -20,11 +21,6 @@ namespace MQTTnet.PacketDispatcher }; _owningPacketDispatcher = owningPacketDispatcher ?? throw new ArgumentNullException(nameof(owningPacketDispatcher)); -#if NET452 - _taskCompletionSource = new TaskCompletionSource(); -#else - _taskCompletionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); -#endif } public MqttPacketAwaitableFilter Filter { get; } @@ -35,7 +31,7 @@ namespace MQTTnet.PacketDispatcher { using (timeoutToken.Token.Register(() => Fail(new MqttCommunicationTimedOutException()))) { - var packet = await _taskCompletionSource.Task.ConfigureAwait(false); + var packet = await _promise.Task.ConfigureAwait(false); return (TPacket)packet; } } @@ -45,45 +41,19 @@ namespace MQTTnet.PacketDispatcher { if (packet == null) throw new ArgumentNullException(nameof(packet)); -#if NET452 - // To prevent deadlocks it is required to call the _TrySetResult_ method - // from a new thread because the awaiting code will not(!) be executed in - // a new thread automatically (due to await). Furthermore _this_ thread will - // do it. But _this_ thread is also reading incoming packets -> deadlock. - // NET452 does not support RunContinuationsAsynchronously - Task.Run(() => _taskCompletionSource.TrySetResult(packet)); -#else - _taskCompletionSource.TrySetResult(packet); -#endif + _promise.TrySetResult(packet); } public void Fail(Exception exception) { if (exception == null) throw new ArgumentNullException(nameof(exception)); -#if NET452 - // To prevent deadlocks it is required to call the _TrySetResult_ method - // from a new thread because the awaiting code will not(!) be executed in - // a new thread automatically (due to await). Furthermore _this_ thread will - // do it. But _this_ thread is also reading incoming packets -> deadlock. - // NET452 does not support RunContinuationsAsynchronously - Task.Run(() => _taskCompletionSource.TrySetException(exception)); -#else - _taskCompletionSource.TrySetException(exception); -#endif + + _promise.TrySetException(exception); } public void Cancel() { -#if NET452 - // To prevent deadlocks it is required to call the _TrySetResult_ method - // from a new thread because the awaiting code will not(!) be executed in - // a new thread automatically (due to await). Furthermore _this_ thread will - // do it. But _this_ thread is also reading incoming packets -> deadlock. - // NET452 does not support RunContinuationsAsynchronously - Task.Run(() => _taskCompletionSource.TrySetCanceled()); -#else - _taskCompletionSource.TrySetCanceled(); -#endif + _promise.TrySetCanceled(); } public void Dispose() diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 3deb23da..3179abde 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -62,9 +62,11 @@ namespace MQTTnet.Server.Internal if (interceptorContext.CloseConnection) { + // When any of the interceptor calls leads to a connection close the connection + // must be closed. So do not revert to false! result.CloseConnection = true; } - + if (!acceptSubscription || string.IsNullOrEmpty(finalTopicFilter.Topic)) { continue; @@ -104,14 +106,17 @@ namespace MQTTnet.Server.Internal var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription).ConfigureAwait(false); var acceptUnsubscription = interceptorContext.ReasonCode == MqttUnsubscribeReasonCode.Success; + result.ReasonCodes.Add(interceptorContext.ReasonCode); + if (interceptorContext.CloseConnection) { + // When any of the interceptor calls leads to a connection close the connection + // must be closed. So do not revert to false! result.CloseConnection = true; } - + if (!acceptUnsubscription) { - result.ReasonCodes.Add(interceptorContext.ReasonCode); continue; } @@ -128,8 +133,7 @@ namespace MQTTnet.Server.Internal foreach (var topicFilter in unsubscribePacket.TopicFilters) { - await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter) - .ConfigureAwait(false); + await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false); } return result; diff --git a/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs b/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs index 386c40ff..c56a6b81 100644 --- a/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs +++ b/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs @@ -249,12 +249,12 @@ namespace MQTTnet.Server return this; } - public MqttServerOptionsBuilder WithPersistentSessions() + public MqttServerOptionsBuilder WithPersistentSessions(bool value = true) { - _options.EnablePersistentSessions = true; + _options.EnablePersistentSessions = value; return this; } - + /// /// Gets or sets the client ID which is used when publishing messages from the server directly. /// diff --git a/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs b/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs index 9f1cd2b5..6d66b11a 100644 --- a/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs @@ -1,4 +1,5 @@ -using System.Collections.Generic; +using System; +using System.Collections.Generic; using System.Threading.Tasks; namespace MQTTnet.Server.Status @@ -6,7 +7,12 @@ namespace MQTTnet.Server.Status public interface IMqttSessionStatus { /// - /// Gets the client identifier. + /// Fired when this session is being deleted. + /// + event EventHandler Deleted; + + /// + /// Gets or sets the client identifier. /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// string ClientId { get; } @@ -16,10 +22,27 @@ namespace MQTTnet.Server.Status /// long PendingApplicationMessagesCount { get; } + /// + /// This items can be used by the library user in order to store custom information. + /// IDictionary Items { get; } - Task ClearPendingApplicationMessagesAsync(); + /// + /// Enqueues a new application message. + /// + /// The application message. + /// A task. + Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage); + + /// + /// Clears the entire queue with pending application messages. + /// + /// A task. + Task ClearApplicationMessagesQueueAsync(); + [Obsolete("Please use _ClearApplicationMessagesQueueAsync_ instead. This will be removed soon.")] + Task ClearPendingApplicationMessagesAsync(); + Task DeleteAsync(); } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index a8f7b70c..99be2101 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; +using MQTTnet.Implementations; using MQTTnet.Server.Internal; namespace MQTTnet.Server.Status @@ -17,36 +18,47 @@ namespace MQTTnet.Server.Status session.Deleted += (_, __) => Deleted?.Invoke(this, EventArgs.Empty); } - - /// - /// Fired when this session is being deleted. - /// + public event EventHandler Deleted; - /// - /// Gets or sets the client identifier. - /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. - /// public string ClientId { get; set; } public long PendingApplicationMessagesCount { get; set; } public DateTime CreatedTimestamp { get; set; } - - /// - /// This items can be used by the library user in order to store custom information. - /// + public IDictionary Items { get; set; } - public Task DeleteAsync() + public Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage) { - return _sessionsManager.DeleteSessionAsync(ClientId); + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + + _session.ApplicationMessagesQueue.Enqueue(new MqttQueuedApplicationMessage + { + ApplicationMessage = applicationMessage, + IsDuplicate = false, + IsRetainedMessage = false, + SubscriptionQualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, + SenderClientId = null + }); + + return PlatformAbstractionLayer.CompletedTask; } - public Task ClearPendingApplicationMessagesAsync() + public Task ClearApplicationMessagesQueueAsync() { _session.ApplicationMessagesQueue.Clear(); return Task.FromResult(0); } + + public Task DeleteAsync() + { + return _sessionsManager.DeleteSessionAsync(ClientId); + } + + public Task ClearPendingApplicationMessagesAsync() + { + return ClearApplicationMessagesQueueAsync(); + } } } diff --git a/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs b/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs index 11a20ac4..e2fe4832 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs @@ -1,5 +1,4 @@ using System.Collections.Generic; -using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; @@ -10,20 +9,16 @@ using MQTTnet.Client.Subscribing; using MQTTnet.Client.Unsubscribing; using MQTTnet.Formatter; using MQTTnet.Protocol; -using MQTTnet.Server; -using MQTTnet.Tests.Mockups; namespace MQTTnet.Tests.MQTTv5 { [TestClass] - public sealed class Client_Tests + public sealed class Client_Tests : BaseTestClass { - public TestContext TestContext { get; set; } - [TestMethod] public async Task Connect_With_New_Mqtt_Features() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -64,7 +59,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Connect() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); await testEnvironment.ConnectClient(o => o.WithProtocolVersion(MqttProtocolVersion.V500).Build()); @@ -74,7 +69,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Connect_And_Disconnect() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -86,7 +81,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Subscribe() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -112,7 +107,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Unsubscribe() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -130,7 +125,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Publish_QoS_0() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -145,7 +140,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Publish_QoS_1() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -160,7 +155,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Publish_QoS_2() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -175,7 +170,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Publish_With_Properties() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -204,7 +199,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Subscribe_And_Publish() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -230,7 +225,7 @@ namespace MQTTnet.Tests.MQTTv5 await client1.DisconnectAsync(); Assert.AreEqual(1, receivedMessages.Count); - Assert.AreEqual("client1", receivedMessages[0].ClientId); + Assert.AreEqual("Subscribe_And_Publish_client1", receivedMessages[0].ClientId); Assert.AreEqual("a", receivedMessages[0].ApplicationMessage.Topic); Assert.AreEqual("b", receivedMessages[0].ApplicationMessage.ConvertPayloadToString()); } @@ -239,7 +234,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Publish_And_Receive_New_Properties() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs index bbe4a2fd..a314b9a0 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs @@ -1,4 +1,5 @@ -using Microsoft.VisualStudio.TestTools.UnitTesting; +using System; +using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; @@ -17,7 +18,7 @@ namespace MQTTnet.Tests.Mockups { public TestClientWrapper(IMqttClient implementation, TestContext testContext) { - Implementation = implementation; + Implementation = implementation ?? throw new ArgumentNullException(nameof(implementation)); TestContext = testContext; } @@ -56,7 +57,7 @@ namespace MQTTnet.Tests.Mockups var existingClientId = clientOptions.ClientId; if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName)) { - clientOptions.ClientId = TestContext.TestName + existingClientId; + clientOptions.ClientId = TestContext.TestName + "_" + existingClientId; } } diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index 6c190a8f..cda0340f 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -11,7 +11,6 @@ using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Options; using MQTTnet.Client.Receiving; -using MQTTnet.Client.Subscribing; using MQTTnet.Formatter; using MQTTnet.Protocol; using MQTTnet.Server; @@ -20,10 +19,8 @@ using MQTTnet.Tests.Mockups; namespace MQTTnet.Tests.Server { [TestClass] - public sealed class General_Tests + public sealed class General_Tests : BaseTestClass { - public TestContext TestContext { get; set; } - [TestMethod] [DataRow("", null)] [DataRow("", "")] @@ -51,7 +48,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Use_Empty_Client_ID() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -77,8 +74,7 @@ namespace MQTTnet.Tests.Server MqttQualityOfServiceLevel.AtMostOnce, "A/B/C", MqttQualityOfServiceLevel.AtMostOnce, - 1, - TestContext); + 1); } [TestMethod] @@ -89,8 +85,7 @@ namespace MQTTnet.Tests.Server MqttQualityOfServiceLevel.AtLeastOnce, "A/B/C", MqttQualityOfServiceLevel.AtLeastOnce, - 1, - TestContext); + 1); } [TestMethod] @@ -101,14 +96,13 @@ namespace MQTTnet.Tests.Server MqttQualityOfServiceLevel.ExactlyOnce, "A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, - 1, - TestContext); + 1); } [TestMethod] public async Task Use_Clean_Session() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -122,7 +116,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Will_Message_Do_Not_Send_On_Clean_Disconnect() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var receivedMessagesCount = 0; @@ -148,7 +142,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Will_Message_Do_Not_Send_On_Takeover() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var receivedMessagesCount = 0; @@ -180,7 +174,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Will_Message_Send() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var receivedMessagesCount = 0; @@ -206,7 +200,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Publish_From_Server() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(); @@ -231,7 +225,7 @@ namespace MQTTnet.Tests.Server { var receivedMessagesCount = 0; - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -262,7 +256,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Session_Takeover() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -290,7 +284,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task No_Messages_If_No_Subscription() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(); @@ -320,7 +314,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Set_Subscription_At_Server() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(); @@ -356,7 +350,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Shutdown_Disconnects_Clients_Gracefully() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder()); @@ -378,7 +372,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Handle_Clean_Disconnect() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder()); @@ -409,7 +403,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Client_Disconnect_Without_Errors() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { bool clientWasConnected; @@ -440,7 +434,7 @@ namespace MQTTnet.Tests.Server { const int ClientCount = 50; - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(); @@ -494,7 +488,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Intercept_Application_Message() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer( new MqttServerOptionsBuilder().WithApplicationMessageInterceptor( @@ -517,7 +511,7 @@ namespace MQTTnet.Tests.Server { var serverStorage = new TestServerStorage(); - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithStorage(serverStorage)); @@ -534,7 +528,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Publish_After_Client_Connects() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(); server.UseClientConnectedHandler(async e => { await server.PublishAsync("/test/1", "true", MqttQualityOfServiceLevel.ExactlyOnce, false); }); @@ -564,7 +558,7 @@ namespace MQTTnet.Tests.Server context.ApplicationMessage.Payload = Encoding.ASCII.GetBytes("extended"); } - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithApplicationMessageInterceptor(Interceptor)); @@ -587,7 +581,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Send_Long_Body() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { const int PayloadSizeInMB = 30; const int CharCount = PayloadSizeInMB * 1024 * 1024; @@ -630,7 +624,7 @@ namespace MQTTnet.Tests.Server { var serverOptions = new MqttServerOptionsBuilder().WithConnectionValidator(context => { context.ReasonCode = MqttConnectReasonCode.NotAuthorized; }); - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { testEnvironment.IgnoreClientLogErrors = true; @@ -659,7 +653,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Same_Client_Id_Refuse_Connection() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { testEnvironment.IgnoreClientLogErrors = true; @@ -747,7 +741,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Same_Client_Id_Connect_Disconnect_Event_Order() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder()); @@ -832,7 +826,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Remove_Session() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder()); @@ -851,7 +845,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Stop_And_Restart() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { testEnvironment.IgnoreClientLogErrors = true; @@ -877,7 +871,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Do_Not_Send_Retained_Messages_For_Denied_Subscription() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithSubscriptionInterceptor(c => { @@ -921,7 +915,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Collect_Messages_In_Disconnected_Session() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithPersistentSessions()); @@ -952,15 +946,14 @@ namespace MQTTnet.Tests.Server } } - static async Task TestPublishAsync( + async Task TestPublishAsync( string topic, MqttQualityOfServiceLevel qualityOfServiceLevel, string topicFilter, MqttQualityOfServiceLevel filterQualityOfServiceLevel, - int expectedReceivedMessagesCount, - TestContext testContext) + int expectedReceivedMessagesCount) { - using (var testEnvironment = new TestEnvironment(testContext)) + using (var testEnvironment = CreateTestEnvironment()) { var receivedMessagesCount = 0; @@ -985,7 +978,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Intercept_Undelivered() { - using (var testEnvironment = new TestEnvironment()) + using (var testEnvironment = CreateTestEnvironment()) { var undeliverd = string.Empty; @@ -1009,7 +1002,7 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Intercept_ClientMessageQueue() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(new MqttServerOptionsBuilder() .WithClientMessageQueueInterceptor(c => c.ApplicationMessage.Topic = "a")); @@ -1047,7 +1040,7 @@ namespace MQTTnet.Tests.Server { const string topic = "a"; - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { await testEnvironment.StartServer(new MqttServerOptionsBuilder() .WithClientMessageQueueInterceptor(c => { })); // Interceptor does nothing but has to be present. diff --git a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs index e9000cdc..69cb66e5 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs @@ -1,42 +1,47 @@ -using System.Linq; +using System; +using System.Linq; using System.Text; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Client.Options; using MQTTnet.Client.Subscribing; +using MQTTnet.Protocol; using MQTTnet.Server; using MQTTnet.Tests.Mockups; namespace MQTTnet.Tests.Server { [TestClass] - public class Session_Tests - { - public TestContext TestContext { get; set; } - + public sealed class Session_Tests : BaseTestClass + { [TestMethod] public async Task Set_Session_Item() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var serverOptions = new MqttServerOptionsBuilder() - .WithConnectionValidator(delegate (MqttConnectionValidatorContext context) + .WithConnectionValidator(delegate(MqttConnectionValidatorContext context) { // Don't validate anything. Just set some session items. context.SessionItems["can_subscribe_x"] = true; context.SessionItems["default_payload"] = "Hello World"; }) - .WithSubscriptionInterceptor(delegate (MqttSubscriptionInterceptorContext context) + .WithSubscriptionInterceptor(delegate(MqttSubscriptionInterceptorContext context) { if (context.TopicFilter.Topic == "x") { - context.AcceptSubscription = context.SessionItems["can_subscribe_x"] as bool? == true; + if (context.SessionItems["can_subscribe_x"] as bool? == false) + { + context.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError; + } } }) - .WithApplicationMessageInterceptor(delegate (MqttApplicationMessageInterceptorContext context) + .WithApplicationMessageInterceptor(delegate(MqttApplicationMessageInterceptorContext context) { - context.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(context.SessionItems["default_payload"] as string); + context.ApplicationMessage.Payload = + Encoding.UTF8.GetBytes( + context.SessionItems["default_payload"] as string ?? String.Empty); }); await testEnvironment.StartServer(serverOptions); @@ -44,7 +49,7 @@ namespace MQTTnet.Tests.Server string receivedPayload = null; var client = await testEnvironment.ConnectClient(); - client.UseApplicationMessageReceivedHandler(delegate (MqttApplicationMessageReceivedEventArgs args) + client.UseApplicationMessageReceivedHandler(delegate(MqttApplicationMessageReceivedEventArgs args) { receivedPayload = args.ApplicationMessage.ConvertPayloadToString(); }); @@ -65,10 +70,10 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Get_Session_Items_In_Status() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { var serverOptions = new MqttServerOptionsBuilder() - .WithConnectionValidator(delegate (MqttConnectionValidatorContext context) + .WithConnectionValidator(delegate(MqttConnectionValidatorContext context) { // Don't validate anything. Just set some session items. context.SessionItems["can_subscribe_x"] = true; @@ -77,7 +82,7 @@ namespace MQTTnet.Tests.Server await testEnvironment.StartServer(serverOptions); - var client = await testEnvironment.ConnectClient(); + await testEnvironment.ConnectClient(); var sessionStatus = await testEnvironment.Server.GetSessionStatusAsync(); var session = sessionStatus.First(); @@ -86,11 +91,10 @@ namespace MQTTnet.Tests.Server } } - [TestMethod] public async Task Manage_Session_MaxParallel() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { testEnvironment.IgnoreClientLogErrors = true; var serverOptions = new MqttServerOptionsBuilder(); @@ -106,6 +110,61 @@ namespace MQTTnet.Tests.Server Assert.AreEqual(1, connectedClients.Count); } } + + [TestMethod] + public async Task Fire_Deleted_Event() + { + using (var testEnvironment = CreateTestEnvironment()) + { + // Arrange client and server. + var server = await testEnvironment.StartServer(o => o.WithPersistentSessions(false)); + var client = await testEnvironment.ConnectClient(); + + // Arrange session status tracking. + var status = await server.GetSessionStatusAsync(); + var clientStatus = status[0]; + + var deletedEventFired = false; + clientStatus.Deleted += (_, __) => + { + deletedEventFired = true; + }; + + // Act: Disconnect the client -> Event must be fired. + await client.DisconnectAsync(); + + await LongTestDelay(); + + // Assert that the event was fired properly. + Assert.IsTrue(deletedEventFired); + } + } + + [TestMethod] + public async Task Inject_Application_Message() + { + using (var testEnvironment = CreateTestEnvironment()) + { + // Arrange client and server. + var server = await testEnvironment.StartServer(); + var client = await testEnvironment.ConnectClient(); + var messageReceivedHandler = testEnvironment.CreateApplicationMessageHandler(client); + + // Arrange session status tracking. + var status = await server.GetSessionStatusAsync(); + var clientStatus = status[0]; + + // Act: Inject the application message. + await clientStatus.EnqueueApplicationMessageAsync(new MqttApplicationMessageBuilder() + .WithTopic("injected_one").Build()); + + await LongTestDelay(); + + // Assert + Assert.AreEqual(1, messageReceivedHandler.ReceivedApplicationMessages.Count); + Assert.AreEqual("injected_one", messageReceivedHandler.ReceivedApplicationMessages[0].Topic); + } + } async Task TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options) { @@ -113,10 +172,10 @@ namespace MQTTnet.Tests.Server { return await testEnvironment.ConnectClient(options); } - catch (System.Exception) + catch { return null; } } } -} +} \ No newline at end of file