Refactor namespaces.

MQTTnet.sln

@@ -3,7 +3,7 @@ Microsoft Visual Studio Solution File, Format Version 12.00
 # Visual Studio 15
 VisualStudioVersion = 15.0.27004.2010
 MinimumVisualStudioVersion = 10.0.40219.1
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Core.Tests", "Tests\MQTTnet.Core.Tests\MQTTnet.Core.Tests.csproj", "{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Tests", "Tests\MQTTnet.Core.Tests\MQTTnet.Tests.csproj", "{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}"
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Tests", "Tests", "{9248C2E1-B9D6-40BF-81EC-86004D7765B4}"
 EndProject

README.md

@@ -23,9 +23,10 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
 * Extensible communication channels (e.g. In-Memory, TCP, TCP+TLS, WS)
 * Lightweight (only the low level implementation of MQTT, no overhead)
 * Performance optimized (processing ~70.000 messages / second)*
+* Uniform API over all versions of MQTT protocol
 * Interfaces included for mocking and testing
 * Access to internal trace messages
-* Unit tested (~100 tests)
+* Unit tested (~120 tests)
 
 \* Tested on local machine (Intel i7 8700K) with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_.
 

Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs

@@ -3,6 +3,7 @@
 using MQTTnet.Adapter;
 using MQTTnet.AspNetCore.Client.Tcp;
 using MQTTnet.Client;
+using MQTTnet.Client.Options;
 using MQTTnet.Diagnostics;
 using MQTTnet.Formatter;
 

Source/MQTTnet.AspnetCore/ReaderExtensions.cs

@@ -36,7 +36,7 @@ public static bool TryDecode(this MqttPacketFormatterAdapter formatter, in ReadO
 
             var bodySlice = copy.Slice(0, bodyLength);
             var buffer = bodySlice.GetArray();
-            packet = formatter.Decode(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(buffer, 0, buffer.Length)));
+            packet = formatter.Decode(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(buffer, 0, buffer.Length), buffer.Length + 2));
             consumed = bodySlice.End;
             observed = bodySlice.End;
             return true;

Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs

@@ -1,7 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
-using MQTTnet.Client;
+using MQTTnet.Client.Connecting;
+using MQTTnet.Client.Disconnecting;
 
 namespace MQTTnet.Extensions.ManagedClient
 {

Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs

@@ -1,5 +1,6 @@
 using System;
 using MQTTnet.Client;
+using MQTTnet.Client.Options;
 using MQTTnet.Server;
 
 namespace MQTTnet.Extensions.ManagedClient

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -4,13 +4,14 @@
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Client;
+using MQTTnet.Client.Publishing;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
-using MqttClientConnectedEventArgs = MQTTnet.Client.MqttClientConnectedEventArgs;
-using MqttClientDisconnectedEventArgs = MQTTnet.Client.MqttClientDisconnectedEventArgs;
+using MqttClientConnectedEventArgs = MQTTnet.Client.Connecting.MqttClientConnectedEventArgs;
+using MqttClientDisconnectedEventArgs = MQTTnet.Client.Disconnecting.MqttClientDisconnectedEventArgs;
 
 namespace MQTTnet.Extensions.ManagedClient
 {
@@ -27,7 +28,7 @@ public class ManagedMqttClient : IManagedMqttClient
         private CancellationTokenSource _publishingCancellationToken;
 
         private ManagedMqttClientStorageManager _storageManager;
-        
+
         private bool _subscriptionsNotPushed;
 
         public ManagedMqttClient(IMqttClient mqttClient, IMqttNetChildLogger logger)
@@ -102,11 +103,12 @@ public Task StopAsync()
             return Task.FromResult(0);
         }
 
-        public Task PublishAsync(MqttApplicationMessage applicationMessage)
+        public async Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage)
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
-            return PublishAsync(new ManagedMqttApplicationMessageBuilder().WithApplicationMessage(applicationMessage).Build());
+            await PublishAsync(new ManagedMqttApplicationMessageBuilder().WithApplicationMessage(applicationMessage).Build()).ConfigureAwait(false);
+            return new MqttClientPublishResult();
         }
 
         public async Task PublishAsync(ManagedMqttApplicationMessage applicationMessage)
@@ -206,7 +208,15 @@ private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
             }
             finally
             {
-                await _mqttClient.DisconnectAsync().ConfigureAwait(false);
+                try
+                {
+                    await _mqttClient.DisconnectAsync().ConfigureAwait(false);
+                }
+                catch (Exception exception)
+                {
+                    _logger.Error(exception, "Error while disconnecting.");
+                }
+
                 _logger.Info("Stopped");
             }
         }
@@ -345,7 +355,7 @@ private async Task SynchronizeSubscriptionsAsync()
 
             lock (_subscriptions)
             {
-                subscriptions = _subscriptions.Select(i => new TopicFilter(i.Key, i.Value)).ToList();
+                subscriptions = _subscriptions.Select(i => new TopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }).ToList();
 
                 unsubscriptions = new HashSet<string>(_unsubscriptions);
                 _unsubscriptions.Clear();

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs

@@ -1,5 +1,6 @@
 using System;
 using MQTTnet.Client;
+using MQTTnet.Client.Options;
 using MQTTnet.Server;
 
 namespace MQTTnet.Extensions.ManagedClient

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs

@@ -1,5 +1,6 @@
 using System;
 using MQTTnet.Client;
+using MQTTnet.Client.Options;
 using MQTTnet.Server;
 
 namespace MQTTnet.Extensions.ManagedClient

Source/MQTTnet/Adapter/IMqttChannelAdapter.cs

@@ -11,7 +11,7 @@ public interface IMqttChannelAdapter : IDisposable
         string Endpoint { get; }
 
         MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
-
+        
         event EventHandler ReadingPacketStarted;
 
         event EventHandler ReadingPacketCompleted;

Source/MQTTnet/Client/IMqttClientAdapterFactory.cs → Source/MQTTnet/Adapter/IMqttClientAdapterFactory.cs

@@ -1,7 +1,7 @@
-using MQTTnet.Adapter;
+using MQTTnet.Client.Options;
 using MQTTnet.Diagnostics;
 
-namespace MQTTnet.Client
+namespace MQTTnet.Adapter
 {
     public interface IMqttClientAdapterFactory
     {

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -16,7 +16,7 @@ namespace MQTTnet.Adapter
     public class MqttChannelAdapter : IMqttChannelAdapter
     {
         private const uint ErrorOperationAborted = 0x800703E3;
-        private const uint ReadBufferSize = 4096;  // TODO: Move buffer size to config
+        private const int ReadBufferSize = 4096;  // TODO: Move buffer size to config
 
         private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);
 
@@ -102,7 +102,7 @@ public async Task SendPacketAsync(MqttBasePacket packet, CancellationToken cance
                 await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false);
                 PacketFormatterAdapter.FreeBuffer();
 
-                _logger.Verbose("TX >>> {0}", packet);
+                _logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Count, packet);
             }
             catch (Exception exception)
             {
@@ -152,7 +152,7 @@ public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, Cancellat
                     throw new MqttProtocolViolationException("Received malformed packet.");
                 }
 
-                _logger.Verbose("RX <<< {0}", packet);
+                _logger.Verbose("RX ({0} bytes) <<< {1}", receivedMqttPacket.TotalLength, packet);
 
                 return packet;
             }
@@ -182,7 +182,7 @@ private async Task<ReceivedMqttPacket> ReceiveAsync(CancellationToken cancellati
 
                 if (fixedHeader.RemainingLength == 0)
                 {
-                    return new ReceivedMqttPacket(fixedHeader.Flags, null);
+                    return new ReceivedMqttPacket(fixedHeader.Flags, null, 2);
                 }
 
                 var body = new byte[fixedHeader.RemainingLength];
@@ -194,15 +194,15 @@ private async Task<ReceivedMqttPacket> ReceiveAsync(CancellationToken cancellati
                     var bytesLeft = body.Length - bodyOffset;
                     if (chunkSize > bytesLeft)
                     {
-                        chunkSize = (uint)bytesLeft;
+                        chunkSize = bytesLeft;
                     }
 
 #if WINDOWS_UWP
                     var readBytes = await _channel.ReadAsync(body, bodyOffset, (int)chunkSize, cancellationToken).ConfigureAwait(false);
 #else
-                    // async/await is not used to avoid the overhead of context switches. We assume that the reamining data
+                    // async/await is not used to avoid the overhead of context switches. We assume that the remaining data
                     // has been sent from the sender directly after the initial bytes.
-                    var readBytes = _channel.ReadAsync(body, bodyOffset, (int)chunkSize, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult();
+                    var readBytes = _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult();
 #endif
 
                     cancellationToken.ThrowIfCancellationRequested();
@@ -211,7 +211,8 @@ private async Task<ReceivedMqttPacket> ReceiveAsync(CancellationToken cancellati
                     bodyOffset += readBytes;
                 } while (bodyOffset < body.Length);
 
-                return new ReceivedMqttPacket(fixedHeader.Flags, new MqttPacketBodyReader(body, 0, body.Length));
+                var bodyReader = new MqttPacketBodyReader(body, 0, body.Length);
+                return new ReceivedMqttPacket(fixedHeader.Flags, bodyReader, fixedHeader.TotalLength);
             }
             finally
             {

Source/MQTTnet/Adapter/MqttConnectingFailedException.cs

@@ -1,16 +1,16 @@
-using MQTTnet.Exceptions;
-using MQTTnet.Protocol;
+using MQTTnet.Client.Connecting;
+using MQTTnet.Exceptions;
 
 namespace MQTTnet.Adapter
 {
     public class MqttConnectingFailedException : MqttCommunicationException
     {
-        public MqttConnectingFailedException(MqttConnectReturnCode returnCode)
-            : base($"Connecting with MQTT server failed ({returnCode}).")
+        public MqttConnectingFailedException(MqttClientConnectResultCode resultCode)
+            : base($"Connecting with MQTT server failed ({resultCode.ToString()}).")
         {
-            ReturnCode = returnCode;
+            ResultCode = resultCode;
         }
 
-        public MqttConnectReturnCode ReturnCode { get; }
+        public MqttClientConnectResultCode ResultCode { get; }
     }
 }

Source/MQTTnet/Adapter/ReceivedMqttPacket.cs

@@ -4,14 +4,17 @@ namespace MQTTnet.Adapter
 {
     public class ReceivedMqttPacket
     {
-        public ReceivedMqttPacket(byte fixedHeader, MqttPacketBodyReader body)
+        public ReceivedMqttPacket(byte fixedHeader, MqttPacketBodyReader body, int totalLength)
         {
             FixedHeader = fixedHeader;
             Body = body;
+            TotalLength = totalLength;
         }
 
         public byte FixedHeader { get; }
 
         public MqttPacketBodyReader Body { get; }
+
+        public int TotalLength { get; }
     }
 }

Source/MQTTnet/ApplicationMessagePublisherExtensions.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
+using MQTTnet.Client.Publishing;
 using MQTTnet.Protocol;
 
 namespace MQTTnet
@@ -29,7 +30,7 @@ public static async Task PublishAsync(this IApplicationMessagePublisher publishe
             }
         }
 
-        public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic)
+        public static Task<MqttClientPublishResult> PublishAsync(this IApplicationMessagePublisher publisher, string topic)
         {
             if (publisher == null) throw new ArgumentNullException(nameof(publisher));
             if (topic == null) throw new ArgumentNullException(nameof(topic));
@@ -38,7 +39,7 @@ public static Task PublishAsync(this IApplicationMessagePublisher publisher, str
                 .WithTopic(topic));
         }
 
-        public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload)
+        public static Task<MqttClientPublishResult> PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload)
         {
             if (publisher == null) throw new ArgumentNullException(nameof(publisher));
             if (topic == null) throw new ArgumentNullException(nameof(topic));
@@ -48,7 +49,7 @@ public static Task PublishAsync(this IApplicationMessagePublisher publisher, str
                 .WithPayload(payload));
         }
 
-        public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        public static Task<MqttClientPublishResult> PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
         {
             if (publisher == null) throw new ArgumentNullException(nameof(publisher));
             if (topic == null) throw new ArgumentNullException(nameof(topic));
@@ -59,7 +60,7 @@ public static Task PublishAsync(this IApplicationMessagePublisher publisher, str
                 .WithQualityOfServiceLevel(qualityOfServiceLevel));
         }
 
-        public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, bool retain)
+        public static Task<MqttClientPublishResult> PublishAsync(this IApplicationMessagePublisher publisher, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, bool retain)
         {
             if (publisher == null) throw new ArgumentNullException(nameof(publisher));
             if (topic == null) throw new ArgumentNullException(nameof(topic));
@@ -71,7 +72,7 @@ public static Task PublishAsync(this IApplicationMessagePublisher publisher, str
                 .WithRetainFlag(retain));
         }
 
-        public static Task PublishAsync(this IApplicationMessagePublisher publisher, Func<MqttApplicationMessageBuilder, MqttApplicationMessageBuilder> builder)
+        public static Task<MqttClientPublishResult> PublishAsync(this IApplicationMessagePublisher publisher, Func<MqttApplicationMessageBuilder, MqttApplicationMessageBuilder> builder)
         {
             var message = builder(new MqttApplicationMessageBuilder()).Build();
             return publisher.PublishAsync(message);

Source/MQTTnet/Client/Connecting/MqttClientConnectResult.cs

@@ -0,0 +1,9 @@
+namespace MQTTnet.Client.Connecting
+{
+    public class MqttClientConnectResult
+    {
+        public bool IsSessionPresent { get; set; }
+
+        public MqttClientConnectResultCode ResultCode { get; set; }
+    }
+}

Source/MQTTnet/Client/Connecting/MqttClientConnectResultCode.cs

@@ -0,0 +1,28 @@
+namespace MQTTnet.Client.Connecting
+{
+    public enum MqttClientConnectResultCode
+    {
+        Success = 0,
+        UnspecifiedError = 128,
+        MalformedPacket = 129,
+        ProtocolError = 130,
+        ImplementationSpecificError = 131,
+        UnsupportedProtocolVersion = 132,
+        ClientIdentifierNotValid = 133,
+        BadUserNameOrPassword = 134,
+        NotAuthorized = 135,
+        ServerUnavailable = 136,
+        ServerBusy = 137,
+        Banned = 138,
+        BadAuthenticationMethod = 140,
+        TopicNameInvalid = 144,
+        PacketTooLarge = 149,
+        QuotaExceeded = 151,
+        PayloadFormatInvalid = 153,
+        RetainNotSupported = 154,
+        QoSNotSupported = 155,
+        UseAnotherServer = 156,
+        ServerMoved = 157,
+        ConnectionRateExceeded = 159
+    }
+}

Source/MQTTnet/Client/Connecting/MqttClientConnectedEventArgs.cs

@@ -0,0 +1,14 @@
+using System;
+
+namespace MQTTnet.Client.Connecting
+{
+    public class MqttClientConnectedEventArgs : EventArgs
+    {
+        public MqttClientConnectedEventArgs(MqttClientConnectResult connectResult)
+        {
+            ConnectResult = connectResult ?? throw new ArgumentNullException(nameof(connectResult));
+        }
+
+        public MqttClientConnectResult ConnectResult { get; }
+    }
+}

Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectOptions.cs

@@ -0,0 +1,9 @@
+namespace MQTTnet.Client.Disconnecting
+{
+    public class MqttClientDisconnectOptions
+    {
+        public MqttClientDisconnectReason ReasonCode { get; set; } = MqttClientDisconnectReason.NormalDisconnection;
+
+        public string ReasonString { get; set; }
+    }
+}

Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectReason.cs

@@ -0,0 +1,8 @@
+namespace MQTTnet.Client.Disconnecting
+{
+    public enum MqttClientDisconnectReason
+    {
+        NormalDisconnection = 0,
+
+    }
+}

Source/MQTTnet/Client/MqttClientDisconnectedEventArgs.cs → Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs

@@ -1,6 +1,6 @@
 using System;
 
-namespace MQTTnet.Client
+namespace MQTTnet.Client.Disconnecting
 {
     public class MqttClientDisconnectedEventArgs : EventArgs
     {