Merge branch 'develop'

Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs

@@ -216,6 +216,7 @@ namespace MQTTnet.Formatter.V3
 
             return new MqttClientPublishResult
             {
+                PacketIdentifier = pubCompPacket.PacketIdentifier,
                 ReasonCode = MqttClientPublishReasonCode.Success
             };
         }

Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs

@@ -223,6 +223,7 @@ namespace MQTTnet.Formatter.V5
             {
                 return new MqttClientPublishResult
                 {
+                    PacketIdentifier = pubCompPacket.PacketIdentifier,
                     ReasonCode = MqttClientPublishReasonCode.UnspecifiedError,
                     ReasonString = pubCompPacket.Properties?.ReasonString,
                     UserProperties = pubCompPacket.Properties?.UserProperties
@@ -231,6 +232,7 @@ namespace MQTTnet.Formatter.V5
 
             var result = new MqttClientPublishResult
             {
+                PacketIdentifier = pubCompPacket.PacketIdentifier,
                 ReasonCode = MqttClientPublishReasonCode.Success,
                 ReasonString = pubCompPacket.Properties?.ReasonString,
                 UserProperties = pubCompPacket.Properties?.UserProperties

Source/MQTTnet/Server/MqttRetainedMessagesManager.cs

@@ -61,8 +61,9 @@ namespace MQTTnet.Server
                 using (await _messagesLock.WaitAsync().ConfigureAwait(false))
                 {
                     var saveIsRequired = false;
+                    var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
 
-                    if (applicationMessage.Payload?.Any() != true)
+                    if (!hasPayload)
                     {
                         saveIsRequired = _messages.Remove(applicationMessage.Topic);
                         _logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);

Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs

@@ -11,7 +11,6 @@ using MQTTnet.Client.Disconnecting;
 using MQTTnet.Client.Options;
 using MQTTnet.Client.Subscribing;
 using MQTTnet.Exceptions;
-using MQTTnet.Formatter;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 using MQTTnet.Tests.Mockups;
@@ -21,6 +20,34 @@ namespace MQTTnet.Tests
     [TestClass]
     public class Client_Tests
     {
+        [TestMethod]
+        public async Task PacketIdentifier_In_Publish_Result()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+                var client = await testEnvironment.ConnectClientAsync();
+
+                var result = await client.PublishAsync("a", "a", MqttQualityOfServiceLevel.AtMostOnce);
+                Assert.AreEqual(null, result.PacketIdentifier);
+
+                result = await client.PublishAsync("b", "b", MqttQualityOfServiceLevel.AtMostOnce);
+                Assert.AreEqual(null, result.PacketIdentifier);
+
+                result = await client.PublishAsync("a", "a", MqttQualityOfServiceLevel.AtLeastOnce);
+                Assert.AreEqual((ushort)1, result.PacketIdentifier);
+
+                result = await client.PublishAsync("b", "b", MqttQualityOfServiceLevel.AtLeastOnce);
+                Assert.AreEqual((ushort)2, result.PacketIdentifier);
+
+                result = await client.PublishAsync("a", "a", MqttQualityOfServiceLevel.ExactlyOnce);
+                Assert.AreEqual((ushort)3, result.PacketIdentifier);
+
+                result = await client.PublishAsync("b", "b", MqttQualityOfServiceLevel.ExactlyOnce);
+                Assert.AreEqual((ushort)4, result.PacketIdentifier);
+            }
+        }
+        
         [TestMethod]
         public async Task Invalid_Connect_Throws_Exception()
         {

Tests/MQTTnet.Core.Tests/MqttTcpChannel_Tests.cs

@@ -15,11 +15,11 @@ namespace MQTTnet.Tests
         public async Task Dispose_Channel_While_Used()
         {
             var ct = new CancellationTokenSource();
+            var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
 
             try
             {
-                var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
-                serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50000));
+                serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001));
                 serverSocket.Listen(0);
 
 #pragma warning disable 4014
@@ -35,7 +35,7 @@ namespace MQTTnet.Tests
                 }, ct.Token);
 
                 var clientSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
-                await clientSocket.ConnectAsync(IPAddress.Loopback, 50000);
+                await clientSocket.ConnectAsync(IPAddress.Loopback, 50001);
 
                 await Task.Delay(100, ct.Token);
 
@@ -68,6 +68,7 @@ namespace MQTTnet.Tests
             finally
             {
                 ct.Cancel(false);
+                serverSocket.Dispose();
             }
         }
     }

Tests/MQTTnet.Core.Tests/Server_Tests.cs

@@ -54,6 +54,20 @@ namespace MQTTnet.Tests
                 1);
         }
 
+        [TestMethod]
+        public async Task Use_Clean_Session()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+
+                var client = testEnvironment.CreateClient();
+                var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build());
+
+                Assert.IsFalse(connectResult.IsSessionPresent);
+            }
+        }
+
         [TestMethod]
         public async Task Will_Message_Do_Not_Send()
         {
@@ -61,7 +75,7 @@ namespace MQTTnet.Tests
             {
                 var receivedMessagesCount = 0;
 
-                await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder());
+                await testEnvironment.StartServerAsync();
 
                 var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build();
 
@@ -540,7 +554,7 @@ namespace MQTTnet.Tests
         }
 
         [TestMethod]
-        public async Task Clear_Retained_Message()
+        public async Task Clear_Retained_Message_With_Empty_Payload()
         {
             using (var testEnvironment = new TestEnvironment())
             {
@@ -566,7 +580,35 @@ namespace MQTTnet.Tests
                 Assert.AreEqual(0, receivedMessagesCount);
             }
         }
-        
+
+        [TestMethod]
+        public async Task Clear_Retained_Message_With_Null_Payload()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                var receivedMessagesCount = 0;
+
+                await testEnvironment.StartServerAsync();
+
+                var c1 = await testEnvironment.ConnectClientAsync();
+
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload((byte[])null).WithRetainFlag().Build());
+
+                await c1.DisconnectAsync();
+
+                var c2 = await testEnvironment.ConnectClientAsync();
+
+                c2.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
+
+                await Task.Delay(200);
+                await c2.SubscribeAsync(new TopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                await Task.Delay(500);
+
+                Assert.AreEqual(0, receivedMessagesCount);
+            }
+        }
+
         [TestMethod]
         public async Task Intercept_Application_Message()
         {

Tests/MQTTnet.TestApp.NetCore/Program.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.Net.Security;
+using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Client;
@@ -81,7 +83,7 @@ namespace MQTTnet.TestApp.NetCore
             {
                 Task.Run(PerformanceTest.RunQoS0Test);
             }
-            
+
             Thread.Sleep(Timeout.Infinite);
         }
     }
@@ -118,4 +120,22 @@ namespace MQTTnet.TestApp.NetCore
             return Task.FromResult(retainedMessages);
         }
     }
+
+    public class WikiCode
+    {
+        public void Code()
+        {
+            //Validate certificate.
+            var options = new MqttClientOptionsBuilder()
+                .WithTls(new MqttClientOptionsBuilderTlsParameters
+                {
+                    CertificateValidationCallback = (X509Certificate x, X509Chain y, SslPolicyErrors z, IMqttClientOptions o) =>
+                        {
+                            // TODO: Check conditions of certificate by using above parameters.
+                            return true;
+                        }
+                })
+                .Build();
+        }
+    }
 }

Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.ObjectModel;
+using System.Net.Security;
+using System.Security.Cryptography.X509Certificates;
 using System.Text;
 using System.Threading.Tasks;
 using Windows.Security.Cryptography.Certificates;
@@ -595,7 +597,7 @@ namespace MQTTnet.TestApp.UniversalWindows
                     };
                 }
             }
-
+            
             // ----------------------------------
             {
                 var options = new MqttServerOptions();
@@ -714,6 +716,25 @@ namespace MQTTnet.TestApp.UniversalWindows
                 var mqttServer = new MqttFactory().CreateMqttServer();
             }
 
+            {
+                // Setup subscription interceptor.
+                var options = new MqttServerOptionsBuilder()
+                    .WithSubscriptionInterceptor(context =>
+                    {
+                        if (context.TopicFilter.Topic.StartsWith("admin/foo/bar") && context.ClientId != "theAdmin")
+                        {
+                            context.AcceptSubscription = false;
+                        }
+
+                        if (context.TopicFilter.Topic.StartsWith("the/secret/stuff") && context.ClientId != "Imperator")
+                        {
+                            context.AcceptSubscription = false;
+                            context.CloseConnection = true;
+                        }
+                    })
+                    .Build();
+            }
+
             {
                 // Setup and start a managed MQTT client.
                 var options = new ManagedMqttClientOptionsBuilder()