* [Server] Added support for complex client IDs.

* [Server] Fixed an issue with not correctly removed old client sessions.
* [Server] Several minor performance improvements.
* [Server] An existing client session is no longer closed if a new client connection is invalid.
* [Client] Added support for sending "CleanSession" flag.

MQTTnet.Core/Client/MqttClient.cs

@@ -53,6 +53,7 @@ namespace MQTTnet.Core.Client
                 ClientId = _options.ClientId,
                 Username = _options.UserName,
                 Password = _options.Password,
+                CleanSession = _options.CleanSession,
                 KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds,
                 WillMessage = willApplicationMessage
             };

MQTTnet.Core/Client/MqttClientOptions.cs

@@ -14,6 +14,8 @@ namespace MQTTnet.Core.Client
 
         public string ClientId { get; set; } = Guid.NewGuid().ToString().Replace("-", string.Empty);
 
+        public bool CleanSession { get; set; } = true;
+
         public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(5);
 
         public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);

MQTTnet.Core/MQTTnet.Core.csproj

@@ -9,15 +9,15 @@
     <Product>MQTTnet</Product>
     <Company>Christian Kratky</Company>
     <Authors>Christian Kratky</Authors>
-    <Version>2.1.1.0</Version>
+    <Version>2.1.2.0</Version>
     <PackageId>MQTTnet.Core</PackageId>
     <Copyright>Copyright © Christian Kratky 2016-2017</Copyright>
     <PackageProjectUrl>https://github.com/chkr1011/MQTTnet</PackageProjectUrl>
     <PackageIconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</PackageIconUrl>
     <RepositoryUrl>https://github.com/chkr1011/MQTTnet</RepositoryUrl>
     <PackageTags>MQTT MQTTClient MQTTServer MQTTBroker Broker</PackageTags>
-    <FileVersion>2.1.1.0</FileVersion>
+    <FileVersion>2.1.2.0</FileVersion>
-    <AssemblyVersion>2.1.1.0</AssemblyVersion>
+    <AssemblyVersion>2.1.2.0</AssemblyVersion>
     <PackageLicenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</PackageLicenseUrl>
   </PropertyGroup>
 

MQTTnet.Core/Serializer/DefaultMqttV311PacketSerializer.cs

@@ -354,11 +354,6 @@ namespace MQTTnet.Core.Serializer
             {
                 throw new MqttProtocolViolationException("CleanSession must be set if ClientId is empty [MQTT-3.1.3-7].");
             }
-
-            if (!string.IsNullOrEmpty(packet.ClientId) && !Regex.IsMatch(packet.ClientId, "^[a-zA-Z0-9]*$"))
-            {
-                throw new MqttProtocolViolationException("ClientId contains invalid characters [MQTT-3.1.3-5].");
-            }
         }
 
         private void ValidatePublishPacket(MqttPublishPacket packet)

MQTTnet.Core/Server/GetOrCreateClientSessionResult.cs

@@ -0,0 +1,9 @@
+namespace MQTTnet.Core.Server
+{
+    public class GetOrCreateClientSessionResult
+    {
+        public bool IsExistingSession { get; set; }
+
+        public MqttClientSession Session { get; set; }
+    }
+}

MQTTnet.Core/Server/MqttClientMessageQueue.cs

@@ -11,7 +11,7 @@ using MQTTnet.Core.Packets;
 
 namespace MQTTnet.Core.Server
 {
-    public class MqttOutgoingPublicationsManager
+    public class MqttClientMessageQueue
     {
         private readonly List<MqttClientPublishPacketContext> _pendingPublishPackets = new List<MqttClientPublishPacketContext>();
         private readonly AsyncGate _gate = new AsyncGate();
@@ -20,7 +20,7 @@ namespace MQTTnet.Core.Server
         private CancellationTokenSource _cancellationTokenSource;
         private IMqttCommunicationAdapter _adapter;
 
-        public MqttOutgoingPublicationsManager(MqttServerOptions options)
+        public MqttClientMessageQueue(MqttServerOptions options)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
         }
@@ -29,13 +29,13 @@ namespace MQTTnet.Core.Server
         {
             if (_cancellationTokenSource != null)
             {
-                throw new InvalidOperationException($"{nameof(MqttOutgoingPublicationsManager)} already started.");
+                throw new InvalidOperationException($"{nameof(MqttClientMessageQueue)} already started.");
             }
 
             _adapter = adapter ?? throw new ArgumentNullException(nameof(adapter));
             _cancellationTokenSource = new CancellationTokenSource();
 
-            Task.Run(() => SendPendingPublishPacketsAsync(_cancellationTokenSource.Token)).Forget();
+            Task.Run(() => SendPendingPublishPacketsAsync(_cancellationTokenSource.Token));
         }
 
         public void Stop()
@@ -87,7 +87,7 @@ namespace MQTTnet.Core.Server
                 }
                 catch (Exception e)
                 {
-                    MqttTrace.Error(nameof(MqttOutgoingPublicationsManager), e, "Error while sending pending publish packets.");
+                    MqttTrace.Error(nameof(MqttClientMessageQueue), e, "Error while sending pending publish packets.");
                 }
                 finally
                 {
@@ -112,11 +112,11 @@ namespace MQTTnet.Core.Server
             }
             catch (MqttCommunicationException exception)
             {
-                MqttTrace.Warning(nameof(MqttOutgoingPublicationsManager), exception, "Sending publish packet failed.");
+                MqttTrace.Warning(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
             }
             catch (Exception exception)
             {
-                MqttTrace.Error(nameof(MqttOutgoingPublicationsManager), exception, "Sending publish packet failed.");
+                MqttTrace.Error(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
             }
             finally
             {

MQTTnet.Core/Server/MqttClientSession.cs

@@ -11,12 +11,12 @@ using MQTTnet.Core.Protocol;
 
 namespace MQTTnet.Core.Server
 {
-    public class MqttClientSession
+    public sealed class MqttClientSession : IDisposable
     {
         private readonly ConcurrentDictionary<ushort, MqttPublishPacket> _pendingIncomingPublications = new ConcurrentDictionary<ushort, MqttPublishPacket>();
 
         private readonly MqttClientSubscriptionsManager _subscriptionsManager = new MqttClientSubscriptionsManager();
-        private readonly MqttOutgoingPublicationsManager _outgoingPublicationsManager;
+        private readonly MqttClientMessageQueue _messageQueue;
         private readonly Action<MqttClientSession, MqttPublishPacket> _publishPacketReceivedCallback;
         private readonly MqttServerOptions _options;
 
@@ -24,15 +24,17 @@ namespace MQTTnet.Core.Server
         private IMqttCommunicationAdapter _adapter;
         private string _identifier;
         private MqttApplicationMessage _willApplicationMessage;
-        
+
         public MqttClientSession(MqttServerOptions options, Action<MqttClientSession, MqttPublishPacket> publishPacketReceivedCallback)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
             _publishPacketReceivedCallback = publishPacketReceivedCallback ?? throw new ArgumentNullException(nameof(publishPacketReceivedCallback));
-            
+
-            _outgoingPublicationsManager = new MqttOutgoingPublicationsManager(options);
+            _messageQueue = new MqttClientMessageQueue(options);
         }
 
+        public bool IsConnected => _adapter != null;
+
         public async Task RunAsync(string identifier, MqttApplicationMessage willApplicationMessage, IMqttCommunicationAdapter adapter)
         {
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -45,7 +47,7 @@ namespace MQTTnet.Core.Server
                 _adapter = adapter;
                 _cancellationTokenSource = new CancellationTokenSource();
 
-                _outgoingPublicationsManager.Start(adapter);
+                _messageQueue.Start(adapter);
                 while (!_cancellationTokenSource.IsCancellationRequested)
                 {
                     var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero);
@@ -65,8 +67,8 @@ namespace MQTTnet.Core.Server
                 {
                     _publishPacketReceivedCallback(this, _willApplicationMessage.ToPublishPacket());
                 }
-                
+
-                _outgoingPublicationsManager.Stop();
+                _messageQueue.Stop();
                 _cancellationTokenSource.Cancel();
                 _adapter = null;
 
@@ -84,66 +86,68 @@ namespace MQTTnet.Core.Server
                 return;
             }
 
-            _outgoingPublicationsManager.Enqueue(senderClientSession, publishPacket);
+            _messageQueue.Enqueue(senderClientSession, publishPacket);
             MqttTrace.Verbose(nameof(MqttClientSession), $"Client '{_identifier}: Enqueued pending publish packet.");
         }
 
-        private async Task HandleIncomingPacketAsync(MqttBasePacket packet)
+        public void Dispose()
+        {
+            _cancellationTokenSource?.Cancel();
+            _cancellationTokenSource?.Dispose();
+        }
+
+        private Task HandleIncomingPacketAsync(MqttBasePacket packet)
         {
             var subscribePacket = packet as MqttSubscribePacket;
             if (subscribePacket != null)
             {
-                await _adapter.SendPacketAsync(_subscriptionsManager.Subscribe(subscribePacket), _options.DefaultCommunicationTimeout);
+                return _adapter.SendPacketAsync(_subscriptionsManager.Subscribe(subscribePacket), _options.DefaultCommunicationTimeout);
-                return;
             }
 
             var unsubscribePacket = packet as MqttUnsubscribePacket;
             if (unsubscribePacket != null)
             {
-                await _adapter.SendPacketAsync(_subscriptionsManager.Unsubscribe(unsubscribePacket), _options.DefaultCommunicationTimeout);
+                return _adapter.SendPacketAsync(_subscriptionsManager.Unsubscribe(unsubscribePacket), _options.DefaultCommunicationTimeout);
-                return;
             }
 
             var publishPacket = packet as MqttPublishPacket;
             if (publishPacket != null)
             {
-                await HandleIncomingPublishPacketAsync(publishPacket);
+                return HandleIncomingPublishPacketAsync(publishPacket);
-                return;
             }
 
             var pubRelPacket = packet as MqttPubRelPacket;
             if (pubRelPacket != null)
             {
-                await HandleIncomingPubRelPacketAsync(pubRelPacket);
+                return HandleIncomingPubRelPacketAsync(pubRelPacket);
-                return;
             }
 
             var pubAckPacket = packet as MqttPubAckPacket;
             if (pubAckPacket != null)
             {
-                await HandleIncomingPubAckPacketAsync(pubAckPacket);
+                return HandleIncomingPubAckPacketAsync(pubAckPacket);
-                return;
             }
 
             if (packet is MqttPingReqPacket)
             {
-                await _adapter.SendPacketAsync(new MqttPingRespPacket(), _options.DefaultCommunicationTimeout);
+                return _adapter.SendPacketAsync(new MqttPingRespPacket(), _options.DefaultCommunicationTimeout);
-                return;
             }
 
             if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
             {
                 _cancellationTokenSource.Cancel();
-                return;
+                return Task.FromResult((object)null);
             }
 
             MqttTrace.Warning(nameof(MqttClientSession), $"Client '{_identifier}': Received not supported packet ({packet}). Closing connection.");
             _cancellationTokenSource.Cancel();
+
+            return Task.FromResult((object)null);
         }
 
         private async Task HandleIncomingPubAckPacketAsync(MqttPubAckPacket pubAckPacket)
         {
-            await Task.FromResult(0);
+            await Task.FromResult((object)null);
         }
 
         private async Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)

MQTTnet.Core/Server/MqttClientSessionManager.cs

@@ -1,93 +0,0 @@
-using System;
-using System.Collections.Concurrent;
-using System.Linq;
-using System.Threading.Tasks;
-using MQTTnet.Core.Adapter;
-using MQTTnet.Core.Diagnostics;
-using MQTTnet.Core.Exceptions;
-using MQTTnet.Core.Packets;
-using MQTTnet.Core.Protocol;
-
-namespace MQTTnet.Core.Server
-{
-    public class MqttClientSessionManager
-    {
-        private readonly ConcurrentDictionary<string, MqttClientSession> _clientSessions = new ConcurrentDictionary<string, MqttClientSession>();
-        private readonly MqttServerOptions _options;
-
-        public MqttClientSessionManager(MqttServerOptions options)
-        {
-            _options = options ?? throw new ArgumentNullException(nameof(options));
-        }
-
-        public async Task RunClientSessionAsync(MqttClientConnectedEventArgs eventArgs)
-        {
-            try
-            {
-                var connectPacket = await eventArgs.ClientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout) as MqttConnectPacket;
-                if (connectPacket == null)
-                {
-                    throw new MqttProtocolViolationException("The first packet from a client must be a 'Connect' packet [MQTT-3.1.0-1].");
-                }
-
-                var connectReturnCode = MqttConnectReturnCode.ConnectionAccepted;
-                if (_options.ConnectionValidator != null)
-                {
-                    connectReturnCode = _options.ConnectionValidator(connectPacket);
-                }
-
-                MqttClientSession clientSession = null;
-                var isSessionPresent = _clientSessions.ContainsKey(connectPacket.ClientId);
-                if (isSessionPresent && connectPacket.CleanSession)
-                {
-                    MqttClientSession _;
-                    _clientSessions.TryRemove(connectPacket.ClientId, out _);
-                }
-                else if (!connectPacket.CleanSession)
-                {
-                    _clientSessions.TryGetValue(connectPacket.ClientId, out clientSession);
-                }
-
-                await eventArgs.ClientAdapter.SendPacketAsync(new MqttConnAckPacket
-                {
-                    ConnectReturnCode = connectReturnCode,
-                    IsSessionPresent = clientSession != null
-                }, _options.DefaultCommunicationTimeout);
-
-                if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
-                {
-                    return;
-                }
-
-                if (clientSession == null)
-                {
-                    clientSession = new MqttClientSession(_options, DispatchPublishPacket);
-                    _clientSessions.TryAdd(connectPacket.ClientId, clientSession);
-                }
-
-                await clientSession.RunAsync(eventArgs.Identifier, connectPacket.WillMessage, eventArgs.ClientAdapter);
-            }
-            catch (Exception exception)
-            {
-                MqttTrace.Error(nameof(MqttServer), exception, exception.Message);
-            }
-            finally
-            {
-                await eventArgs.ClientAdapter.DisconnectAsync();
-            }
-        }
-
-        public void Clear()
-        {
-            _clientSessions.Clear();
-        }
-
-        private void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
-        {
-            foreach (var clientSession in _clientSessions.Values.ToList())
-            {
-                clientSession.EnqueuePublishPacket(senderClientSession, publishPacket);
-            }
-        }
-    }
-}

MQTTnet.Core/Server/MqttClientSessionsManager.cs

@@ -0,0 +1,135 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using MQTTnet.Core.Adapter;
+using MQTTnet.Core.Diagnostics;
+using MQTTnet.Core.Exceptions;
+using MQTTnet.Core.Packets;
+using MQTTnet.Core.Protocol;
+
+namespace MQTTnet.Core.Server
+{
+    public class MqttClientSessionsManager
+    {
+        private readonly object _syncRoot = new object();
+        private readonly Dictionary<string, MqttClientSession> _clientSessions = new Dictionary<string, MqttClientSession>();
+        private readonly MqttServerOptions _options;
+
+        public MqttClientSessionsManager(MqttServerOptions options)
+        {
+            _options = options ?? throw new ArgumentNullException(nameof(options));
+        }
+
+        public async Task RunClientSessionAsync(MqttClientConnectedEventArgs eventArgs)
+        {
+            try
+            {
+                var connectPacket = await eventArgs.ClientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout) as MqttConnectPacket;
+                if (connectPacket == null)
+                {
+                    throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1].");
+                }
+
+                var connectReturnCode = ValidateConnection(connectPacket);
+                if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
+                {
+                    await eventArgs.ClientAdapter.SendPacketAsync(new MqttConnAckPacket
+                    {
+                        ConnectReturnCode = connectReturnCode
+                    }, _options.DefaultCommunicationTimeout);
+
+                    return;
+                }
+
+                var clientSession = GetOrCreateClientSession(connectPacket);
+
+                await eventArgs.ClientAdapter.SendPacketAsync(new MqttConnAckPacket
+                {
+                    ConnectReturnCode = connectReturnCode,
+                    IsSessionPresent = clientSession.IsExistingSession
+                }, _options.DefaultCommunicationTimeout);
+
+                await clientSession.Session.RunAsync(eventArgs.Identifier, connectPacket.WillMessage, eventArgs.ClientAdapter);
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttServer), exception, exception.Message);
+            }
+            finally
+            {
+                await eventArgs.ClientAdapter.DisconnectAsync();
+            }
+        }
+
+        public void Clear()
+        {
+            lock (_syncRoot)
+            {
+                _clientSessions.Clear();
+            }
+        }
+
+        public IList<string> GetConnectedClients()
+        {
+            lock (_syncRoot)
+            {
+                return _clientSessions.Where(s => s.Value.IsConnected).Select(s => s.Key).ToList();
+            }
+        }
+
+        private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
+        {
+            if (_options.ConnectionValidator != null)
+            {
+                return _options.ConnectionValidator(connectPacket);
+            }
+
+            return MqttConnectReturnCode.ConnectionAccepted;
+        }
+
+        private GetOrCreateClientSessionResult GetOrCreateClientSession(MqttConnectPacket connectPacket)
+        {
+            lock (_syncRoot)
+            {
+                MqttClientSession clientSession;
+                var isSessionPresent = _clientSessions.TryGetValue(connectPacket.ClientId, out clientSession);
+                if (isSessionPresent)
+                {
+                    if (connectPacket.CleanSession)
+                    {
+                        _clientSessions.Remove(connectPacket.ClientId);
+                        clientSession.Dispose();
+                        clientSession = null;
+                        MqttTrace.Verbose(nameof(MqttClientSessionsManager), $"Disposed existing session of client '{connectPacket.ClientId}'.");
+                    }
+                    else
+                    {
+                        MqttTrace.Verbose(nameof(MqttClientSessionsManager), $"Reusing existing session of client '{connectPacket.ClientId}'.");
+                    }
+                }
+
+                var isExistingSession = true;
+                if (clientSession == null)
+                {
+                    isExistingSession = false;
+
+                    clientSession = new MqttClientSession(_options, DispatchPublishPacket);
+                    _clientSessions[connectPacket.ClientId] = clientSession;
+
+                    MqttTrace.Verbose(nameof(MqttClientSessionsManager), $"Created a new session for client '{connectPacket.ClientId}'.");
+                }
+
+                return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
+            }
+        }
+
+        private void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        {
+            foreach (var clientSession in _clientSessions.Values.ToList())
+            {
+                clientSession.EnqueuePublishPacket(senderClientSession, publishPacket);
+            }
+        }
+    }
+}

MQTTnet.Core/Server/MqttServer.cs

@@ -1,15 +1,15 @@
 using System;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
 using MQTTnet.Core.Diagnostics;
-using MQTTnet.Core.Internal;
 
 namespace MQTTnet.Core.Server
 {
     public class MqttServer
     {
-        private readonly MqttClientSessionManager _clientSessionManager;
+        private readonly MqttClientSessionsManager _clientSessionsManager;
         private readonly IMqttServerAdapter _adapter;
         private readonly MqttServerOptions _options;
 
@@ -20,7 +20,12 @@ namespace MQTTnet.Core.Server
             _options = options ?? throw new ArgumentNullException(nameof(options));
             _adapter = adapter ?? throw new ArgumentNullException(nameof(adapter));
             
-            _clientSessionManager = new MqttClientSessionManager(options);
+            _clientSessionsManager = new MqttClientSessionsManager(options);
+        }
+
+        public IList<string> GetConnectedClients()
+        {
+            return _clientSessionsManager.GetConnectedClients();
         }
 
         public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
@@ -54,7 +59,7 @@ namespace MQTTnet.Core.Server
             _adapter.ClientConnected -= OnClientConnected;
             _adapter.Stop();
 
-            _clientSessionManager.Clear();
+            _clientSessionsManager.Clear();
 
             MqttTrace.Information(nameof(MqttServer), "Stopped.");
         }
@@ -64,7 +69,7 @@ namespace MQTTnet.Core.Server
             MqttTrace.Information(nameof(MqttServer), $"Client '{eventArgs.Identifier}': Connected.");
             ClientConnected?.Invoke(this, eventArgs);
 
-            Task.Run(async () => await _clientSessionManager.RunClientSessionAsync(eventArgs), _cancellationTokenSource.Token).Forget();
+            Task.Run(() => _clientSessionsManager.RunClientSessionAsync(eventArgs), _cancellationTokenSource.Token);
         }
     }
 }

MQTTnet.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet</id>
-    <version>2.1.1.0</version>
+    <version>2.1.2.0</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,7 +10,11 @@
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>MQTTnet is a .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
-    <releaseNotes>* Added MQTT server.</releaseNotes>
+    <releaseNotes>* [Server] Added support for complex client IDs.
+      * [Server] Fixed an issue with not correctly removed old client sessions.
+      * [Server] Several minor performance improvements.
+      * [Server] An existing client session is no longer closed if a new client connection is invalid.
+      * [Client] Added support for sending "CleanSession" flag.</releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2017</copyright>
     <tags>MQTT MQTTClient MQTTServer MQTTBroker Broker</tags>
   </metadata>

Tests/MQTTnet.TestMqttClient/Program.cs

@@ -33,7 +33,9 @@ namespace MQTTnet.TestMqttClient
             {
                 var options = new MqttClientOptions
                 {
-                    Server = "localhost"
+                    Server = "localhost",
+                    ClientId = "XYZ",
+                    CleanSession = true
                 };
 
                 var client = new MqttClientFactory().CreateMqttClient(options);