diff --git a/MQTTnet.sln b/MQTTnet.sln index a5e74961..541f7cbe 100644 --- a/MQTTnet.sln +++ b/MQTTnet.sln @@ -57,11 +57,11 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Apps", "Apps", "{A56E3128-1 EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Blazor", "Blazor", "{61B165A0-5AA8-4E04-A53D-A22A84AA6EB7}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Test.BlazorApp.Server", "Tests\MQTTnet.Test.BlazorApp\Server\MQTTnet.Test.BlazorApp.Server.csproj", "{A9662AF3-3520-4BF8-9DFF-C55C0C33D08F}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Test.BlazorApp.Server", "Tests\MQTTnet.Test.BlazorApp\Server\MQTTnet.Test.BlazorApp.Server.csproj", "{A9662AF3-3520-4BF8-9DFF-C55C0C33D08F}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Test.BlazorApp.Client", "Tests\MQTTnet.Test.BlazorApp\Client\MQTTnet.Test.BlazorApp.Client.csproj", "{D260D63D-7902-4C55-9665-84C5CACBBB24}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Test.BlazorApp.Client", "Tests\MQTTnet.Test.BlazorApp\Client\MQTTnet.Test.BlazorApp.Client.csproj", "{D260D63D-7902-4C55-9665-84C5CACBBB24}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Test.BlazorApp.Shared", "Tests\MQTTnet.Test.BlazorApp\Shared\MQTTnet.Test.BlazorApp.Shared.csproj", "{DDB069BA-6E1A-48C7-B374-5D903B553CAD}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Test.BlazorApp.Shared", "Tests\MQTTnet.Test.BlazorApp\Shared\MQTTnet.Test.BlazorApp.Shared.csproj", "{DDB069BA-6E1A-48C7-B374-5D903B553CAD}" EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 6b9bc3f0..0102be15 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -219,7 +219,7 @@ namespace MQTTnet.Client { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - MqttTopicValidator.ThrowIfInvalid(applicationMessage.Topic); + MqttTopicValidator.ThrowIfInvalid(applicationMessage); ThrowIfDisposed(); ThrowIfNotConnected(); diff --git a/Source/MQTTnet/Formatter/IMqttDataConverter.cs b/Source/MQTTnet/Formatter/IMqttDataConverter.cs index 8c3ed59d..76079a1b 100644 --- a/Source/MQTTnet/Formatter/IMqttDataConverter.cs +++ b/Source/MQTTnet/Formatter/IMqttDataConverter.cs @@ -18,6 +18,8 @@ namespace MQTTnet.Formatter MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket); + MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket); + MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket); MqttClientAuthenticateResult CreateClientConnectResult(MqttConnAckPacket connAckPacket); diff --git a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs index 2c55696e..31ea9ed4 100644 --- a/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs +++ b/Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs @@ -40,6 +40,15 @@ namespace MQTTnet.Formatter.V3 }; } + public MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket) + { + return new MqttPubRecPacket + { + PacketIdentifier = publishPacket.PacketIdentifier, + ReasonCode = MqttPubRecReasonCode.Success + }; + } + public MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket) { if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); diff --git a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs index ae2aeed1..248d2c32 100644 --- a/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs +++ b/Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs @@ -58,6 +58,15 @@ namespace MQTTnet.Formatter.V5 }; } + public MqttBasePacket CreatePubRecPacket(MqttPublishPacket publishPacket) + { + return new MqttPubRecPacket + { + PacketIdentifier = publishPacket.PacketIdentifier, + ReasonCode = MqttPubRecReasonCode.Success + }; + } + public MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket) { return new MqttApplicationMessage diff --git a/Source/MQTTnet/Protocol/MqttTopicValidator.cs b/Source/MQTTnet/Protocol/MqttTopicValidator.cs index 1d7a19a7..9787e95d 100644 --- a/Source/MQTTnet/Protocol/MqttTopicValidator.cs +++ b/Source/MQTTnet/Protocol/MqttTopicValidator.cs @@ -1,9 +1,22 @@ -using MQTTnet.Exceptions; +using System; +using MQTTnet.Exceptions; namespace MQTTnet.Protocol { public static class MqttTopicValidator { + public static void ThrowIfInvalid(MqttApplicationMessage applicationMessage) + { + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + + if (applicationMessage.TopicAlias > 0) + { + return; + } + + ThrowIfInvalid(applicationMessage.Topic); + } + public static void ThrowIfInvalid(string topic) { if (string.IsNullOrEmpty(topic)) diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index a1ee8798..3d69e2d5 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -18,6 +18,7 @@ namespace MQTTnet.Server { public sealed class MqttClientConnection : IDisposable { + readonly Dictionary _topicAlias = new Dictionary(); readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource(); @@ -316,19 +317,26 @@ namespace MQTTnet.Server { Interlocked.Increment(ref _sentApplicationMessagesCount); + HandleTopicAlias(publishPacket); + + var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket); + _sessionsManager.DispatchApplicationMessage(applicationMessage, this); + switch (publishPacket.QualityOfServiceLevel) { case MqttQualityOfServiceLevel.AtMostOnce: { - return HandleIncomingPublishPacketWithQoS0Async(publishPacket); + return PlatformAbstractionLayer.CompletedTask; } case MqttQualityOfServiceLevel.AtLeastOnce: { - return HandleIncomingPublishPacketWithQoS1Async(publishPacket, cancellationToken); + var pubAckPacket = _dataConverter.CreatePubAckPacket(publishPacket); + return SendAsync(pubAckPacket, cancellationToken); } case MqttQualityOfServiceLevel.ExactlyOnce: { - return HandleIncomingPublishPacketWithQoS2Async(publishPacket, cancellationToken); + var pubRecPacket = _dataConverter.CreatePubRecPacket(publishPacket); + return SendAsync(pubRecPacket, cancellationToken); } default: { @@ -337,36 +345,33 @@ namespace MQTTnet.Server } } - Task HandleIncomingPublishPacketWithQoS0Async(MqttPublishPacket publishPacket) + void HandleTopicAlias(MqttPublishPacket publishPacket) { - var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket); - - _sessionsManager.DispatchApplicationMessage(applicationMessage, this); - - return PlatformAbstractionLayer.CompletedTask; - } - - Task HandleIncomingPublishPacketWithQoS1Async(MqttPublishPacket publishPacket, CancellationToken cancellationToken) - { - var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket); - _sessionsManager.DispatchApplicationMessage(applicationMessage, this); - - var pubAckPacket = _dataConverter.CreatePubAckPacket(publishPacket); - return SendAsync(pubAckPacket, cancellationToken); - } - - Task HandleIncomingPublishPacketWithQoS2Async(MqttPublishPacket publishPacket, CancellationToken cancellationToken) - { - var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket); - _sessionsManager.DispatchApplicationMessage(applicationMessage, this); - - var pubRecPacket = new MqttPubRecPacket + if (publishPacket.Properties?.TopicAlias == null) { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubRecReasonCode.Success - }; + return; + } - return SendAsync(pubRecPacket, cancellationToken); + var topicAlias = publishPacket.Properties.TopicAlias.Value; + + lock (_topicAlias) + { + if (!string.IsNullOrEmpty(publishPacket.Topic)) + { + _topicAlias[topicAlias] = publishPacket.Topic; + } + else + { + if (_topicAlias.TryGetValue(topicAlias, out var topic)) + { + publishPacket.Topic = topic; + } + else + { + + } + } + } } async Task SendPendingPacketsAsync(CancellationToken cancellationToken) diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index 94f0b612..d7403096 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs @@ -124,6 +124,20 @@ namespace MQTTnet.Tests.Mockups return client; } + public async Task ConnectClientAsync(Action optionsBuilder) + { + if (optionsBuilder == null) throw new ArgumentNullException(nameof(optionsBuilder)); + + var options = new MqttClientOptionsBuilder(); + options = options.WithTcpServer("localhost", ServerPort); + optionsBuilder.Invoke(options); + + var client = CreateClient(); + await client.ConnectAsync(options.Build()).ConfigureAwait(false); + + return client; + } + public async Task ConnectClientAsync(MqttClientOptionsBuilder options) { if (options == null) throw new ArgumentNullException(nameof(options)); diff --git a/Tests/MQTTnet.Core.Tests/Server_TopicAlias_Tests.cs b/Tests/MQTTnet.Core.Tests/Server_TopicAlias_Tests.cs new file mode 100644 index 00000000..e831491f --- /dev/null +++ b/Tests/MQTTnet.Core.Tests/Server_TopicAlias_Tests.cs @@ -0,0 +1,57 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Client; +using MQTTnet.Formatter; +using MQTTnet.Tests.Mockups; + +namespace MQTTnet.Tests +{ + [TestClass] + public class Server_TopicAlias_Tests + { + [TestMethod] + public async Task Publish_After_Client_Connects() + { + using (var testEnvironment = new TestEnvironment()) + { + await testEnvironment.StartServerAsync(); + + var receivedTopics = new List(); + + var c1 = await testEnvironment.ConnectClientAsync(options => options.WithProtocolVersion(MqttProtocolVersion.V500)); + c1.UseApplicationMessageReceivedHandler(e => + { + lock (receivedTopics) + { + receivedTopics.Add(e.ApplicationMessage.Topic); + } + }); + + await c1.SubscribeAsync("#"); + + var c2 = await testEnvironment.ConnectClientAsync(options => options.WithProtocolVersion(MqttProtocolVersion.V500)); + + var message = new MqttApplicationMessage + { + Topic = "this_is_the_topic", + TopicAlias = 22 + }; + + await c2.PublishAsync(message); + + message.Topic = null; + + await c2.PublishAsync(message); + await c2.PublishAsync(message); + + await Task.Delay(500); + + Assert.AreEqual(3, receivedTopics.Count); + CollectionAssert.AllItemsAreNotNull(receivedTopics); + Assert.IsTrue(receivedTopics.All(t => t.Equals("this_is_the_topic"))); + } + } + } +}