From 9fddfb3e77cc032d9d496a678ae4a1badfb34631 Mon Sep 17 00:00:00 2001
From: Christian <christian.kratky@googlemail.com>
Date: Wed, 25 Nov 2020 21:08:01 +0100
Subject: [PATCH] Add UnitTests

---
 .../ManagedMqttClient_Tests.cs                | 60 +++++++++++++++++++
 1 file changed, 60 insertions(+)

diff --git a/Tests/MQTTnet.Core.Tests/ManagedMqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/ManagedMqttClient_Tests.cs
index ae671de7..636e998c 100644
--- a/Tests/MQTTnet.Core.Tests/ManagedMqttClient_Tests.cs
+++ b/Tests/MQTTnet.Core.Tests/ManagedMqttClient_Tests.cs
@@ -352,6 +352,66 @@ namespace MQTTnet.Tests
             }
         }
 
+        [TestMethod]
+        public async Task Manage_Session_MaxParallel_Subscribe()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                testEnvironment.IgnoreClientLogErrors = true;
+
+                var serverOptions = new MqttServerOptionsBuilder();
+                await testEnvironment.StartServerAsync(serverOptions);
+
+                var options = new MqttClientOptionsBuilder().WithClientId("1").WithKeepAlivePeriod(TimeSpan.FromSeconds(5));
+                
+                var hasReceive = false;
+
+                void OnReceive()
+                {
+                    if (!hasReceive)
+                    {
+                        hasReceive = true;
+                    }
+                }
+
+                var clients = await Task.WhenAll(Enumerable.Range(0, 25)
+                    .Select(i => TryConnect_Subscribe(testEnvironment, options, OnReceive)));
+
+                var connectedClients = clients.Where(c => c?.IsConnected ?? false).ToList();
+
+                Assert.AreEqual(1, connectedClients.Count);
+
+                await Task.Delay(10000); // over 1.5T
+
+                var option2 = new MqttClientOptionsBuilder().WithClientId("2").WithKeepAlivePeriod(TimeSpan.FromSeconds(10));
+                var sendClient = await testEnvironment.ConnectClientAsync(option2);
+                await sendClient.PublishAsync("aaa", "1");
+                
+                await Task.Delay(3000);
+
+                Assert.AreEqual(true, hasReceive);
+            }
+        }
+
+        private async Task<IMqttClient> TryConnect_Subscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive)
+        {
+
+            try
+            {
+                var sendClient = await testEnvironment.ConnectClientAsync(options);
+                sendClient.ApplicationMessageReceivedHandler = new MQTTnet.Client.Receiving.MqttApplicationMessageReceivedHandlerDelegate(e =>
+                {
+                    onReceive();
+                });
+                await sendClient.SubscribeAsync("aaa");
+                return sendClient;
+            }
+            catch (System.Exception)
+            {
+                return null;
+            }
+        }
+
         async Task<ManagedMqttClient> CreateManagedClientAsync(
             TestEnvironment testEnvironment,
             IMqttClient underlyingClient = null,