Add topic filter builder

MQTTnet.Core/Server/MqttServerOptions.cs

@@ -16,7 +16,7 @@ namespace MQTTnet.Core.Server
 
         public Func<MqttConnectPacket, MqttConnectReturnCode> ConnectionValidator { get; set; }
 
-        public Action<MqttApplicationMessage> ApplicationMessageInterceptor { get; set; }
+        public Func<MqttApplicationMessage, MqttApplicationMessage> ApplicationMessageInterceptor { get; set; }
 
         public IMqttServerStorage Storage { get; set; }
     }

MQTTnet.Core/TopicFilter.cs

@@ -1,6 +1,6 @@
 using MQTTnet.Core.Protocol;
 
-namespace MQTTnet.Core.Packets
+namespace MQTTnet.Core
 {
     public sealed class TopicFilter
     {

MQTTnet.Core/TopicFilterBuilder.cs

@@ -0,0 +1,51 @@
+using MQTTnet.Core.Protocol;
+using MQTTnet.Core.Exceptions;
+
+namespace MQTTnet.Core
+{
+    public class TopicFilterBuilder
+    {
+        private string _topic;
+        private MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
+
+        public TopicFilterBuilder WithTopic(string topic)
+        {
+            _topic = topic;
+            return this;
+        }
+
+        public TopicFilterBuilder WithQualityOfServiceLevel(MqttQualityOfServiceLevel qualityOfServiceLevel)
+        {
+            _qualityOfServiceLevel = qualityOfServiceLevel;
+            return this;
+        }
+
+        public TopicFilterBuilder WithAtLeastOnceQoS()
+        {
+            _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce;
+            return this;
+        }
+
+        public TopicFilterBuilder WithAtMostOnceQoS()
+        {
+            _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
+            return this;
+        }
+
+        public TopicFilterBuilder WithExactlyOnceQoS()
+        {
+            _qualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce;
+            return this;
+        }
+
+        public TopicFilter Build()
+        {
+            if (string.IsNullOrEmpty(_topic))
+            {
+                throw new MqttProtocolViolationException("Topic is not set.");
+            }
+
+            return new TopicFilter(_topic, _qualityOfServiceLevel);
+        }
+    }
+}

README.md

@@ -27,6 +27,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
 * Included core _MqttClient_ with low level functionality
 * Also included _ManagedMqttClient_ which maintains the connection and subscriptions automatically. Also application messages are queued and re-scheduled for higher QoS levels automatically.
 * Rx support (via another project)
+* Compatible with Microsoft Azure IoT Hub
 
 ### Server (broker)
 * List of connected clients available

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -1,4 +1,6 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
+using System.Text;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Core.Adapter;
@@ -53,7 +55,7 @@ namespace MQTTnet.Core.Tests
             var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter }, new MqttNetTrace());
             await s.StartAsync();
 
-            var willMessage = new MqttApplicationMessage("My/last/will", new byte[0], MqttQualityOfServiceLevel.AtMostOnce, false);
+            var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build();
             var c1 = await serverAdapter.ConnectTestClient(s, "c1");
             var c2 = await serverAdapter.ConnectTestClient(s, "c2", willMessage);
 
@@ -83,7 +85,7 @@ namespace MQTTnet.Core.Tests
             var receivedMessagesCount = 0;
             c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
 
-            var message = new MqttApplicationMessage("a", new byte[0], MqttQualityOfServiceLevel.AtLeastOnce, false);
+            var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build();
 
             await c2.PublishAsync(message);
             Assert.AreEqual(0, receivedMessagesCount);
@@ -118,7 +120,7 @@ namespace MQTTnet.Core.Tests
             var receivedMessagesCount = 0;
             c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
 
-            var message = new MqttApplicationMessage("a", new byte[0], MqttQualityOfServiceLevel.AtLeastOnce, false);
+            var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build();
             await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce));
 
             s.Publish(message);
@@ -229,6 +231,41 @@ namespace MQTTnet.Core.Tests
             Assert.AreEqual(1, receivedMessagesCount);
         }
 
+        [TestMethod]
+        public async Task MqttServer_InterceptMessage()
+        {
+            var options = new MqttServerOptions
+            {
+                ApplicationMessageInterceptor = message =>
+                {
+                    message.Payload = Encoding.ASCII.GetBytes("extended");
+                    return message;
+                }
+            };
+
+            var serverAdapter = new TestMqttServerAdapter();
+            var s = new MqttServer(options, new List<IMqttServerAdapter> { serverAdapter }, new MqttNetTrace());
+            await s.StartAsync();
+
+            var c1 = await serverAdapter.ConnectTestClient(s, "c1");
+            var c2 = await serverAdapter.ConnectTestClient(s, "c2");
+            await c2.SubscribeAsync(new TopicFilterBuilder().WithTopic("test").Build());
+
+            var isIntercepted = false;
+            c2.ApplicationMessageReceived += (sender, args) =>
+            {
+                isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(args.ApplicationMessage.Payload), StringComparison.Ordinal) == 0;
+            };
+
+            var m = new MqttApplicationMessageBuilder().WithTopic("test").Build();
+            await c1.PublishAsync(m);
+            await c1.DisconnectAsync();
+
+            await Task.Delay(500);
+
+            Assert.IsTrue(isIntercepted);
+        }
+
         private class TestStorage : IMqttServerStorage
         {
             private IList<MqttApplicationMessage> _messages = new List<MqttApplicationMessage>();
@@ -262,8 +299,8 @@ namespace MQTTnet.Core.Tests
             var receivedMessagesCount = 0;
             c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
 
-            await c1.SubscribeAsync(new TopicFilter(topicFilter, filterQualityOfServiceLevel));
-            await c2.PublishAsync(new MqttApplicationMessage(topic, new byte[0], qualityOfServiceLevel, false));
+            await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic(topicFilter).WithQualityOfServiceLevel(filterQualityOfServiceLevel).Build());
+            await c2.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(new byte[0]).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
 
             await Task.Delay(500);
             await c1.UnsubscribeAsync(topicFilter);

Tests/MQTTnet.TestApp.NetCore/ServerTest.cs

@@ -48,6 +48,8 @@ namespace MQTTnet.TestApp.NetCore
                         // based payload with the timestamp is a suitable use case.
                         message.Payload = Encoding.UTF8.GetBytes(DateTime.Now.ToString("O"));
                     }
+
+                    return message;
                 };
 
                 //var certificate = new X509Certificate(@"C:\certs\test\test.cer", "");