From 4386ef1e56b26be027fe9e78afc9c5e26e28c792 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sun, 22 Oct 2017 14:19:12 +0200 Subject: [PATCH] Added interceptor for application messages --- Build/MQTTnet.nuspec | 5 +- .../MqttApplicationMessageExtensions.cs | 13 +-- MQTTnet.Core/MqttApplicationMessage.cs | 13 ++- MQTTnet.Core/MqttApplicationMessageBuilder.cs | 10 ++- .../Serializer/MqttPacketSerializer.cs | 14 ++-- .../MqttClientRetainedMessagesManager.cs | 29 ++++--- MQTTnet.Core/Server/MqttClientSession.cs | 19 +++-- .../Server/MqttClientSessionsManager.cs | 6 +- MQTTnet.Core/Server/MqttServer.cs | 4 +- MQTTnet.Core/Server/MqttServerOptions.cs | 2 + README.md | 1 + .../MqttPacketSerializerTests.cs | 24 +++--- .../PerformanceTest.cs | 12 +-- Tests/MQTTnet.TestApp.NetCore/Program.cs | 74 ++--------------- Tests/MQTTnet.TestApp.NetCore/ServerTest.cs | 81 +++++++++++++++++++ .../MainPage.xaml.cs | 11 +-- 16 files changed, 182 insertions(+), 136 deletions(-) create mode 100644 Tests/MQTTnet.TestApp.NetCore/ServerTest.cs diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index b78712df..50ac4373 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -14,11 +14,12 @@ * [Core] Migrated the trace to a non-static approach (Breaking Change!) * [Core] Added a builder for application messages using a fluent API * [Client] Added a first version of a managed client which will manage the connection, subscription etc. automatically (Thanks to @JTrotta) -* [Server] Added support for WebSockets via ASP.NET Core 2.0 (Thanks to @ChristianRiedl) * [Client] The session state response from the server is now returned in the _ConnectAsync_ method and also part of the _Connected_ event args +* [Server] Added support for WebSockets via ASP.NET Core 2.0 (Thanks to @ChristianRiedl) +* [Server] Added support for a custom application message interceptor Copyright Christian Kratky 2016-2017 - MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M + MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation diff --git a/MQTTnet.Core/Internal/MqttApplicationMessageExtensions.cs b/MQTTnet.Core/Internal/MqttApplicationMessageExtensions.cs index 7d3c6658..7ca7d005 100644 --- a/MQTTnet.Core/Internal/MqttApplicationMessageExtensions.cs +++ b/MQTTnet.Core/Internal/MqttApplicationMessageExtensions.cs @@ -6,12 +6,13 @@ namespace MQTTnet.Core.Internal { public static MqttApplicationMessage ToApplicationMessage(this MqttPublishPacket publishPacket) { - return new MqttApplicationMessage( - publishPacket.Topic, - publishPacket.Payload, - publishPacket.QualityOfServiceLevel, - publishPacket.Retain - ); + return new MqttApplicationMessage + { + Topic = publishPacket.Topic, + Payload = publishPacket.Payload, + QualityOfServiceLevel = publishPacket.QualityOfServiceLevel, + Retain = publishPacket.Retain + }; } public static MqttPublishPacket ToPublishPacket(this MqttApplicationMessage applicationMessage) diff --git a/MQTTnet.Core/MqttApplicationMessage.cs b/MQTTnet.Core/MqttApplicationMessage.cs index 382d702c..b07ee18f 100644 --- a/MQTTnet.Core/MqttApplicationMessage.cs +++ b/MQTTnet.Core/MqttApplicationMessage.cs @@ -5,6 +5,11 @@ namespace MQTTnet.Core { public sealed class MqttApplicationMessage { + public MqttApplicationMessage() + { + } + + [Obsolete("Use object initializer or _MqttApplicationMessageBuilder_ instead.")] public MqttApplicationMessage(string topic, byte[] payload, MqttQualityOfServiceLevel qualityOfServiceLevel, bool retain) { Topic = topic ?? throw new ArgumentNullException(nameof(topic)); @@ -13,12 +18,12 @@ namespace MQTTnet.Core Retain = retain; } - public string Topic { get; } + public string Topic { get; set; } - public byte[] Payload { get; } + public byte[] Payload { get; set; } - public MqttQualityOfServiceLevel QualityOfServiceLevel { get; } + public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; } - public bool Retain { get; } + public bool Retain { get; set; } } } diff --git a/MQTTnet.Core/MqttApplicationMessageBuilder.cs b/MQTTnet.Core/MqttApplicationMessageBuilder.cs index 9905b098..c2e04318 100644 --- a/MQTTnet.Core/MqttApplicationMessageBuilder.cs +++ b/MQTTnet.Core/MqttApplicationMessageBuilder.cs @@ -13,7 +13,7 @@ namespace MQTTnet.Core private string _topic; private byte[] _payload; private bool _retain; - + public MqttApplicationMessageBuilder WithTopic(string topic) { _topic = topic; @@ -102,7 +102,13 @@ namespace MQTTnet.Core throw new MqttProtocolViolationException("Topic is not set."); } - return new MqttApplicationMessage(_topic, _payload ?? new byte[0], _qualityOfServiceLevel, _retain); + return new MqttApplicationMessage + { + Topic = _topic, + Payload = _payload ?? new byte[0], + QualityOfServiceLevel = _qualityOfServiceLevel, + Retain = _retain + }; } } } diff --git a/MQTTnet.Core/Serializer/MqttPacketSerializer.cs b/MQTTnet.Core/Serializer/MqttPacketSerializer.cs index aa4d3b93..00906f4a 100644 --- a/MQTTnet.Core/Serializer/MqttPacketSerializer.cs +++ b/MQTTnet.Core/Serializer/MqttPacketSerializer.cs @@ -65,7 +65,7 @@ namespace MQTTnet.Core.Serializer case MqttUnsubscribePacket unsubscribePacket: return Serialize(unsubscribePacket, writer); case MqttUnsubAckPacket unsubAckPacket: return Serialize(unsubAckPacket, writer); default: throw new MqttProtocolViolationException("Packet type invalid."); - } + } } private static MqttBasePacket Deserialize(MqttPacketHeader header, MqttPacketReader reader) @@ -233,11 +233,13 @@ namespace MQTTnet.Core.Serializer if (willFlag) { - packet.WillMessage = new MqttApplicationMessage( - reader.ReadStringWithLengthPrefix(), - reader.ReadWithLengthPrefix(), - (MqttQualityOfServiceLevel)willQoS, - willRetain); + packet.WillMessage = new MqttApplicationMessage + { + Topic = reader.ReadStringWithLengthPrefix(), + Payload = reader.ReadWithLengthPrefix(), + QualityOfServiceLevel = (MqttQualityOfServiceLevel)willQoS, + Retain = willRetain + }; } if (usernameFlag) diff --git a/MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs b/MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs index fe95f019..22de3054 100644 --- a/MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs +++ b/MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs @@ -3,15 +3,14 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; using MQTTnet.Core.Diagnostics; -using MQTTnet.Core.Internal; using MQTTnet.Core.Packets; namespace MQTTnet.Core.Server { public sealed class MqttClientRetainedMessagesManager { + private readonly Dictionary _retainedMessages = new Dictionary(); private readonly MqttNetTrace _trace; - private readonly Dictionary _retainedMessages = new Dictionary(); private readonly MqttServerOptions _options; public MqttClientRetainedMessagesManager(MqttServerOptions options, MqttNetTrace trace) @@ -35,7 +34,7 @@ namespace MQTTnet.Core.Server _retainedMessages.Clear(); foreach (var retainedMessage in retainedMessages) { - _retainedMessages[retainedMessage.Topic] = retainedMessage.ToPublishPacket(); + _retainedMessages[retainedMessage.Topic] = retainedMessage; } } } @@ -45,25 +44,25 @@ namespace MQTTnet.Core.Server } } - public async Task HandleMessageAsync(string clientId, MqttPublishPacket publishPacket) + public async Task HandleMessageAsync(string clientId, MqttApplicationMessage applicationMessage) { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - List allRetainedMessages; + List allRetainedMessages; lock (_retainedMessages) { - if (publishPacket.Payload?.Any() == false) + if (applicationMessage.Payload?.Any() == false) { - _retainedMessages.Remove(publishPacket.Topic); - _trace.Information(nameof(MqttClientRetainedMessagesManager), "Client '{0}' cleared retained message for topic '{1}'.", clientId, publishPacket.Topic); + _retainedMessages.Remove(applicationMessage.Topic); + _trace.Information(nameof(MqttClientRetainedMessagesManager), "Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic); } else { - _retainedMessages[publishPacket.Topic] = publishPacket; - _trace.Information(nameof(MqttClientRetainedMessagesManager), "Client '{0}' updated retained message for topic '{1}'.", clientId, publishPacket.Topic); + _retainedMessages[applicationMessage.Topic] = applicationMessage; + _trace.Information(nameof(MqttClientRetainedMessagesManager), "Client '{0}' updated retained message for topic '{1}'.", clientId, applicationMessage.Topic); } - allRetainedMessages = new List(_retainedMessages.Values); + allRetainedMessages = new List(_retainedMessages.Values); } try @@ -71,7 +70,7 @@ namespace MQTTnet.Core.Server // ReSharper disable once UseNullPropagation if (_options.Storage != null) { - await _options.Storage.SaveRetainedMessagesAsync(allRetainedMessages.Select(p => p.ToApplicationMessage()).ToList()); + await _options.Storage.SaveRetainedMessagesAsync(allRetainedMessages); } } catch (Exception exception) @@ -80,9 +79,9 @@ namespace MQTTnet.Core.Server } } - public List GetMessages(MqttSubscribePacket subscribePacket) + public List GetMessages(MqttSubscribePacket subscribePacket) { - var retainedMessages = new List(); + var retainedMessages = new List(); lock (_retainedMessages) { foreach (var retainedMessage in _retainedMessages.Values) diff --git a/MQTTnet.Core/Server/MqttClientSession.cs b/MQTTnet.Core/Server/MqttClientSession.cs index 969ec015..f913901e 100644 --- a/MQTTnet.Core/Server/MqttClientSession.cs +++ b/MQTTnet.Core/Server/MqttClientSession.cs @@ -72,7 +72,7 @@ namespace MQTTnet.Core.Server { if (_willMessage != null) { - _mqttClientSessionsManager.DispatchPublishPacket(this, _willMessage.ToPublishPacket()); + _mqttClientSessionsManager.DispatchApplicationMessage(this, _willMessage); } _cancellationTokenSource?.Cancel(false); @@ -175,27 +175,30 @@ namespace MQTTnet.Core.Server var retainedMessages = _mqttClientSessionsManager.RetainedMessagesManager.GetMessages(subscribePacket); foreach (var publishPacket in retainedMessages) { - EnqueuePublishPacket(publishPacket); + EnqueuePublishPacket(publishPacket.ToPublishPacket()); } } private async Task HandleIncomingPublishPacketAsync(IMqttCommunicationAdapter adapter, MqttPublishPacket publishPacket) { - if (publishPacket.Retain) + var applicationMessage = publishPacket.ToApplicationMessage(); + _options.ApplicationMessageInterceptor?.Invoke(applicationMessage); + + if (applicationMessage.Retain) { - await _mqttClientSessionsManager.RetainedMessagesManager.HandleMessageAsync(ClientId, publishPacket); + await _mqttClientSessionsManager.RetainedMessagesManager.HandleMessageAsync(ClientId, applicationMessage); } - switch (publishPacket.QualityOfServiceLevel) + switch (applicationMessage.QualityOfServiceLevel) { case MqttQualityOfServiceLevel.AtMostOnce: { - _mqttClientSessionsManager.DispatchPublishPacket(this, publishPacket); + _mqttClientSessionsManager.DispatchApplicationMessage(this, applicationMessage); return; } case MqttQualityOfServiceLevel.AtLeastOnce: { - _mqttClientSessionsManager.DispatchPublishPacket(this, publishPacket); + _mqttClientSessionsManager.DispatchApplicationMessage(this, applicationMessage); await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }); @@ -210,7 +213,7 @@ namespace MQTTnet.Core.Server _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier); } - _mqttClientSessionsManager.DispatchPublishPacket(this, publishPacket); + _mqttClientSessionsManager.DispatchApplicationMessage(this, applicationMessage); await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }); diff --git a/MQTTnet.Core/Server/MqttClientSessionsManager.cs b/MQTTnet.Core/Server/MqttClientSessionsManager.cs index 3a93d089..aedd3d76 100644 --- a/MQTTnet.Core/Server/MqttClientSessionsManager.cs +++ b/MQTTnet.Core/Server/MqttClientSessionsManager.cs @@ -117,11 +117,11 @@ namespace MQTTnet.Core.Server } } - public void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket) + public void DispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) { try { - var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, publishPacket.ToApplicationMessage()); + var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, applicationMessage); ApplicationMessageReceived?.Invoke(this, eventArgs); } catch (Exception exception) @@ -133,7 +133,7 @@ namespace MQTTnet.Core.Server { foreach (var clientSession in _clientSessions.Values.ToList()) { - clientSession.EnqueuePublishPacket(publishPacket); + clientSession.EnqueuePublishPacket(applicationMessage.ToPublishPacket()); } } } diff --git a/MQTTnet.Core/Server/MqttServer.cs b/MQTTnet.Core/Server/MqttServer.cs index 7da59575..341b3701 100644 --- a/MQTTnet.Core/Server/MqttServer.cs +++ b/MQTTnet.Core/Server/MqttServer.cs @@ -4,7 +4,6 @@ using System.Threading; using System.Threading.Tasks; using MQTTnet.Core.Adapter; using MQTTnet.Core.Diagnostics; -using MQTTnet.Core.Internal; namespace MQTTnet.Core.Server { @@ -42,7 +41,8 @@ namespace MQTTnet.Core.Server { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - _clientSessionsManager.DispatchPublishPacket(null, applicationMessage.ToPublishPacket()); + _options.ApplicationMessageInterceptor?.Invoke(applicationMessage); + _clientSessionsManager.DispatchApplicationMessage(null, applicationMessage); } public async Task StartAsync() diff --git a/MQTTnet.Core/Server/MqttServerOptions.cs b/MQTTnet.Core/Server/MqttServerOptions.cs index c3673b9d..27dc21c5 100644 --- a/MQTTnet.Core/Server/MqttServerOptions.cs +++ b/MQTTnet.Core/Server/MqttServerOptions.cs @@ -16,6 +16,8 @@ namespace MQTTnet.Core.Server public Func ConnectionValidator { get; set; } + public Action ApplicationMessageInterceptor { get; set; } + public IMqttServerStorage Storage { get; set; } } } diff --git a/README.md b/README.md index fdd38deb..bc7c13b1 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov * Extensible client credential validation * Retained messages are supported including persisting via interface methods (own implementation required) * WebSockets supported (via ASP.NET Core 2.0) +* A custom message interceptor can be added which allows transforming or extending every received application message # Supported frameworks * .NET Standard 1.3+ diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs index c78e3aee..bdd07516 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs @@ -53,11 +53,13 @@ namespace MQTTnet.Core.Tests Username = "USER", KeepAlivePeriod = 123, CleanSession = true, - WillMessage = new MqttApplicationMessage( - "My/last/will", - Encoding.UTF8.GetBytes("Good byte."), - MqttQualityOfServiceLevel.AtLeastOnce, - true) + WillMessage = new MqttApplicationMessage + { + Topic = "My/last/will", + Payload = Encoding.UTF8.GetBytes("Good byte."), + QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, + Retain = true + } }; SerializeAndCompare(p, "EDUABE1RVFQE7gB7AANYWVoADE15L2xhc3Qvd2lsbAAKR29vZCBieXRlLgAEVVNFUgAEUEFTUw=="); @@ -88,11 +90,13 @@ namespace MQTTnet.Core.Tests Username = "USER", KeepAlivePeriod = 123, CleanSession = true, - WillMessage = new MqttApplicationMessage( - "My/last/will", - Encoding.UTF8.GetBytes("Good byte."), - MqttQualityOfServiceLevel.AtLeastOnce, - true) + WillMessage = new MqttApplicationMessage + { + Topic = "My/last/will", + Payload = Encoding.UTF8.GetBytes("Good byte."), + QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce, + Retain = true + } }; DeserializeAndCompare(p, "EDUABE1RVFQE7gB7AANYWVoADE15L2xhc3Qvd2lsbAAKR29vZCBieXRlLgAEVVNFUgAEUEFTUw=="); diff --git a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs index f8b8c33b..638bdc0a 100644 --- a/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs @@ -151,12 +151,12 @@ namespace MQTTnet.TestApp.NetCore private static MqttApplicationMessage CreateMessage() { - return new MqttApplicationMessage( - "A/B/C", - Encoding.UTF8.GetBytes("Hello World"), - MqttQualityOfServiceLevel.AtMostOnce, - false - ); + return new MqttApplicationMessage + { + Topic = "A/B/C", + Payload = Encoding.UTF8.GetBytes("Hello World"), + QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce + }; } private static Task PublishSingleMessage(IMqttClient client, MqttApplicationMessage applicationMessage, ref int count) diff --git a/Tests/MQTTnet.TestApp.NetCore/Program.cs b/Tests/MQTTnet.TestApp.NetCore/Program.cs index 0b5cf64e..63c4ee24 100644 --- a/Tests/MQTTnet.TestApp.NetCore/Program.cs +++ b/Tests/MQTTnet.TestApp.NetCore/Program.cs @@ -32,7 +32,7 @@ namespace MQTTnet.TestApp.NetCore } else if (pressedKey.KeyChar == '2') { - Task.Run(RunServerAsync); + Task.Run(ServerTest.RunAsync); } else if (pressedKey.KeyChar == '3') { @@ -119,12 +119,12 @@ namespace MQTTnet.TestApp.NetCore { Console.ReadLine(); - var applicationMessage = new MqttApplicationMessage( - "A/B/C", - Encoding.UTF8.GetBytes("Hello World"), - MqttQualityOfServiceLevel.AtLeastOnce, - false - ); + var applicationMessage = new MqttApplicationMessage + { + Topic = "A/B/C", + Payload = Encoding.UTF8.GetBytes("Hello World"), + QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce + }; await client.PublishAsync(applicationMessage); } @@ -135,66 +135,6 @@ namespace MQTTnet.TestApp.NetCore } } - private static Task RunServerAsync() - { - MqttNetTrace.TraceMessagePublished += (s, e) => - { - Console.WriteLine($">> [{e.TraceMessage.Timestamp:O}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Source}] [{e.TraceMessage.Level}]: {e.TraceMessage.Message}"); - if (e.TraceMessage.Exception != null) - { - Console.WriteLine(e.TraceMessage.Exception); - } - }; - - try - { - var options = new MqttServerOptions - { - ConnectionValidator = p => - { - if (p.ClientId == "SpecialClient") - { - if (p.Username != "USER" || p.Password != "PASS") - { - return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; - } - } - - return MqttConnectReturnCode.ConnectionAccepted; - } - }; - - options.Storage = new RetainedMessageHandler(); - - - //var certificate = new X509Certificate(@"C:\certs\test\test.cer", ""); - //options.TlsEndpointOptions.Certificate = certificate.Export(X509ContentType.Cert); - //options.ConnectionBacklog = 5; - //options.DefaultEndpointOptions.IsEnabled = true; - //options.TlsEndpointOptions.IsEnabled = false; - - var mqttServer = new MqttServerFactory().CreateMqttServer(options); - mqttServer.ClientDisconnected += (s, e) => - { - Console.Write("Client disconnected event fired."); - }; - - mqttServer.StartAsync(); - - Console.WriteLine("Press any key to exit."); - Console.ReadLine(); - - mqttServer.StopAsync(); - } - catch (Exception e) - { - Console.WriteLine(e); - } - - Console.ReadLine(); - return Task.FromResult(0); - } - // ReSharper disable once UnusedMember.Local private static async void WikiCode() { diff --git a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs new file mode 100644 index 00000000..639920fc --- /dev/null +++ b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs @@ -0,0 +1,81 @@ +using System; +using System.Text; +using System.Threading.Tasks; +using MQTTnet.Core.Diagnostics; +using MQTTnet.Core.Protocol; +using MQTTnet.Core.Server; + +namespace MQTTnet.TestApp.NetCore +{ + public static class ServerTest + { + public static Task RunAsync() + { + MqttNetTrace.TraceMessagePublished += (s, e) => + { + Console.WriteLine($">> [{e.TraceMessage.Timestamp:O}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Source}] [{e.TraceMessage.Level}]: {e.TraceMessage.Message}"); + if (e.TraceMessage.Exception != null) + { + Console.WriteLine(e.TraceMessage.Exception); + } + }; + + try + { + var options = new MqttServerOptions + { + ConnectionValidator = p => + { + if (p.ClientId == "SpecialClient") + { + if (p.Username != "USER" || p.Password != "PASS") + { + return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; + } + } + + return MqttConnectReturnCode.ConnectionAccepted; + } + }; + + options.Storage = new RetainedMessageHandler(); + + options.ApplicationMessageInterceptor = message => + { + if (MqttTopicFilterComparer.IsMatch(message.Topic, "/myTopic/WithTimestamp/#")) + { + // Replace the payload with the timestamp. But also extending a JSON + // based payload with the timestamp is a suitable use case. + message.Payload = Encoding.UTF8.GetBytes(DateTime.Now.ToString("O")); + } + }; + + //var certificate = new X509Certificate(@"C:\certs\test\test.cer", ""); + //options.TlsEndpointOptions.Certificate = certificate.Export(X509ContentType.Cert); + //options.ConnectionBacklog = 5; + //options.DefaultEndpointOptions.IsEnabled = true; + //options.TlsEndpointOptions.IsEnabled = false; + + var mqttServer = new MqttServerFactory().CreateMqttServer(options); + mqttServer.ClientDisconnected += (s, e) => + { + Console.Write("Client disconnected event fired."); + }; + + mqttServer.StartAsync(); + + Console.WriteLine("Press any key to exit."); + Console.ReadLine(); + + mqttServer.StopAsync(); + } + catch (Exception e) + { + Console.WriteLine(e); + } + + Console.ReadLine(); + return Task.FromResult(0); + } + } +} diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs index 8826e135..981d4a01 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs @@ -119,11 +119,12 @@ namespace MQTTnet.TestApp.UniversalWindows payload = Convert.FromBase64String(Payload.Text); } - var message = new MqttApplicationMessage( - Topic.Text, - payload, - qos, - Retain.IsChecked == true); + var message = new MqttApplicationMessageBuilder() + .WithTopic(Topic.Text) + .WithPayload(payload) + .WithQualityOfServiceLevel(qos) + .WithRetainFlag(Retain.IsChecked == true) + .Build(); await _mqttClient.PublishAsync(message); }