fix: Add thread safeties to subscriptions list (#247)

Author: SaschaBa

Source/HiveMQtt/Client/HiveMQClient.cs

@@ -66,6 +66,8 @@ public HiveMQClient(HiveMQClientOptions? options = null)
     /// <inheritdoc />
     public List<Subscription> Subscriptions { get; } = new();
 
+    private SemaphoreSlim SubscriptionsSemaphore { get; } = new(1, 1);
+
     /// <inheritdoc />
     public bool IsConnected() => this.Connection.State == ConnectState.Connected;
 
@@ -479,8 +481,16 @@ void TaskHandler(object? sender, OnSubAckReceivedEventArgs args)
                     subscription.MessageReceivedHandler = handler.Value;
                 }
             }
+        }
 
-            this.Subscriptions.Add(subscription);
+        try
+        {
+            await this.SubscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
+            this.Subscriptions.AddRange(subscribeResult.Subscriptions);
+        }
+        finally
+        {
+            _ = this.SubscriptionsSemaphore.Release();
         }
 
         // Fire the corresponding event
@@ -508,9 +518,17 @@ public async Task<UnsubscribeResult> UnsubscribeAsync(string topic)
     /// <inheritdoc />
     public async Task<UnsubscribeResult> UnsubscribeAsync(Subscription subscription)
     {
-        if (!this.Subscriptions.Contains(subscription))
+        try
         {
-            throw new HiveMQttClientException("No such subscription found.  Make sure to take subscription(s) from HiveMQClient.Subscriptions[] or HiveMQClient.GetSubscriptionByTopic().");
+            await this.SubscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
+            if (!this.Subscriptions.Contains(subscription))
+            {
+                throw new HiveMQttClientException("No such subscription found.  Make sure to take subscription(s) from HiveMQClient.Subscriptions[] or HiveMQClient.GetSubscriptionByTopic().");
+            }
+        }
+        finally
+        {
+            _ = this.SubscriptionsSemaphore.Release();
         }
 
         var unsubOptions = new UnsubscribeOptionsBuilder()
@@ -523,11 +541,22 @@ public async Task<UnsubscribeResult> UnsubscribeAsync(Subscription subscription)
     /// <inheritdoc />
     public async Task<UnsubscribeResult> UnsubscribeAsync(List<Subscription> subscriptions)
     {
-        for (var i = 0; i < subscriptions.Count; i++)
+        HashSet<Subscription> currentSubscriptions;
+        try
         {
-            if (!this.Subscriptions.Contains(subscriptions[i]))
+            await this.SubscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
+            currentSubscriptions = this.Subscriptions.ToHashSet();
+        }
+        finally
+        {
+            _ = this.SubscriptionsSemaphore.Release();
+        }
+
+        foreach (var sub in subscriptions)
+        {
+            if (!currentSubscriptions.Contains(sub))
             {
-                throw new HiveMQttClientException("No such subscription found.  Make sure to take subscription(s) from HiveMQClient.Subscriptions[] or HiveMQClient.GetSubscriptionByTopic().");
+                throw new HiveMQttClientException("No such subscription found. Make sure to take subscription(s) from HiveMQClient.Subscriptions[] or HiveMQClient.GetSubscriptionByTopic().");
             }
         }
 
@@ -589,13 +618,28 @@ void TaskHandler(object? sender, OnUnsubAckReceivedEventArgs args)
         };
 
         var counter = 0;
+        var subscriptionsToRemove = new List<Subscription>();
         foreach (var reasonCode in unsubAck.ReasonCodes)
         {
             unsubscribeResult.Subscriptions[counter].UnsubscribeReasonCode = reasonCode;
             if (reasonCode == UnsubAckReasonCode.Success)
             {
-                // Remove the subscription from the client
-                this.Subscriptions.Remove(unsubscribeResult.Subscriptions[counter]);
+                // Collect subscriptions which need to be removed
+               subscriptionsToRemove.Add(unsubscribeResult.Subscriptions[counter]);
+            }
+        }
+
+        if (subscriptionsToRemove.Count != 0)
+        {
+            try
+            {
+                // remove subscriptions from client while locking them
+                await this.SubscriptionsSemaphore.WaitAsync().ConfigureAwait(false);
+                _ = this.Subscriptions.RemoveAll(subscriptionsToRemove.Contains);
+            }
+            finally
+            {
+                _ = this.SubscriptionsSemaphore.Release();
             }
         }
 

Source/HiveMQtt/Client/HiveMQClientEvents.cs

@@ -266,26 +266,46 @@ internal virtual void OnMessageReceivedEventLauncher(PublishPacket packet)
             messageHandled = true;
         }
 
+        if (packet.Message.Topic is null)
+        {
+            return;
+        }
+
         // Per Subscription Event Handler
-        foreach (var subscription in this.Subscriptions)
+        // use ToList, so the iteration goes through a copy and changes at the list make not problems
+        // otherwise it would be necessary to lock the Subscriptions with the semaphore of HiveMQClient
+        List<Subscription> tempList;
+        try
+        {
+            this.SubscriptionsSemaphore.Wait();
+            tempList = this.Subscriptions.ToList();
+        }
+        finally
+        {
+            _ = this.SubscriptionsSemaphore.Release();
+        }
+
+        var matchingSubscriptions = tempList.Where(sub =>
+            sub.MessageReceivedHandler is not null &&
+            MatchTopic(sub.TopicFilter.Topic, packet.Message.Topic));
+
+        foreach (var subscription in matchingSubscriptions)
         {
-            if (packet.Message.Topic != null && MatchTopic(subscription.TopicFilter.Topic, packet.Message.Topic))
+            // We have a per-subscription message handler.
+            _ = Task.Run(() =>
             {
-                if (subscription.MessageReceivedHandler != null && subscription.MessageReceivedHandler.GetInvocationList().Length > 0)
+                try
                 {
-                    // We have a per-subscription message handler.
-                    _ = Task.Run(() => subscription.MessageReceivedHandler?.Invoke(this, eventArgs)).ContinueWith(
-                        t =>
-                        {
-                            if (t.IsFaulted)
-                            {
-                                Logger.Error($"per-subscription MessageReceivedEventLauncher faulted ({packet.Message.Topic}): {t.Exception?.Message}");
-                            }
-                        }, TaskScheduler.Default);
-
-                    messageHandled = true;
+                    subscription.MessageReceivedHandler?.Invoke(this, eventArgs);
                 }
-            }
+                catch (Exception e)
+                {
+                    Logger.Error(
+                        $"per-subscription MessageReceivedEventLauncher faulted ({packet.Message.Topic}): {e.Message}");
+                }
+            });
+
+            messageHandled = true;
         }
 
         if (!messageHandled)

Source/HiveMQtt/Client/HiveMQClientUtil.cs

@@ -31,20 +31,18 @@ public partial class HiveMQClient : IDisposable, IHiveMQClient
     /// <returns>A boolean indicating whether the subscription exists.</returns>
     internal bool SubscriptionExists(Subscription subscription)
     {
-        if (this.Subscriptions.Contains(subscription))
+        List<Subscription> tempList;
+        try
         {
-            return true;
+            this.SubscriptionsSemaphore.Wait();
+            tempList = this.Subscriptions.ToList();
         }
-
-        foreach (var candidate in this.Subscriptions)
+        finally
         {
-            if (candidate.TopicFilter.Topic == subscription.TopicFilter.Topic)
-            {
-                return true;
-            }
+            _ = this.SubscriptionsSemaphore.Release();
         }
 
-        return false;
+        return tempList.Any(s => s.TopicFilter.Topic == subscription.TopicFilter.Topic);
     }
 
     /// <summary>
@@ -54,15 +52,18 @@ internal bool SubscriptionExists(Subscription subscription)
     /// <returns>The subscription or null if not found.</returns>
     internal Subscription? GetSubscriptionByTopic(string topic)
     {
-        foreach (var subscription in this.Subscriptions)
+        List<Subscription> tempList;
+        try
         {
-            if (subscription.TopicFilter.Topic == topic)
-            {
-                return subscription;
-            }
+            this.SubscriptionsSemaphore.Wait();
+            tempList = this.Subscriptions.ToList();
+        }
+        finally
+        {
+            _ = this.SubscriptionsSemaphore.Release();
         }
 
-        return null;
+        return tempList.FirstOrDefault(s => s.TopicFilter.Topic == topic);
     }
 
     /// <summary>

Tests/HiveMQtt.Test/HiveMQClient/ThreadSafeSubscribeUnsubscribeTest.cs

@@ -0,0 +1,94 @@
+namespace HiveMQtt.Test.HiveMQClient;
+
+using System.Collections.Concurrent;
+using Client;
+using Client.Options;
+using MQTT5.ReasonCodes;
+using MQTT5.Types;
+using Xunit;
+using Xunit.Abstractions;
+
+public class ThreadSafeSubscribeUnsubscribeTest
+{
+    private readonly ITestOutputHelper testOutputHelper;
+
+    public ThreadSafeSubscribeUnsubscribeTest(ITestOutputHelper testOutputHelper)
+    {
+        this.testOutputHelper = testOutputHelper;
+    }
+
+    [Fact]
+    public async Task SubscribeUnsubscribe_InManyThreadsAsync()
+    {
+        const int workerCount = 100;
+        const int iterationsPerWorker = 100;
+        const int topicsPerIteration = 10;
+        const int publishesPerIteration = 10;
+        const int totalExpectedSuccesses = workerCount * iterationsPerWorker;
+
+        var options = new HiveMQClientOptionsBuilder().WithClientId("ConcurrentSubscribeUnsubscribeAndPublish").Build();
+        options.ResponseTimeoutInMs = 20000;
+        var client = new HiveMQClient(options);
+        Assert.NotNull(client);
+
+        var connectResult = await client.ConnectAsync().ConfigureAwait(false);
+        Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);
+        Assert.True(client.IsConnected());
+
+        client.OnMessageReceived += (_, args) => { };
+        _ = await client.SubscribeAsync("/test/#").ConfigureAwait(false);
+
+        var exceptionMessages = new ConcurrentBag<string>();
+        var successCount = 0;
+        var tasks = new List<Task>();
+
+        foreach (var workerId in Enumerable.Range(0, workerCount))
+        {
+            tasks.Add(Task.Run(async () =>
+            {
+                for (var i = 0; i < iterationsPerWorker; i++)
+                {
+                    var topicPrefix = $"/test/topic/{workerId}/{i}";
+
+                    var topicsToManage = Enumerable.Range(0, topicsPerIteration)
+                                                   .Select(j => $"{topicPrefix}/{(char)('a' + j)}")
+                                                   .ToList();
+
+                    try
+                    {
+                        var topicFilters = topicsToManage.Select(topic => new TopicFilter(topic, QualityOfService.ExactlyOnceDelivery)).ToList();
+                        var subscribeOptions = new SubscribeOptions { TopicFilters = topicFilters };
+                        _ = await client.SubscribeAsync(subscribeOptions).ConfigureAwait(false);
+
+                        var publishTasks = new List<Task>(publishesPerIteration * 3);
+                        for (var j = 0; j < publishesPerIteration; j++)
+                        {
+                            publishTasks.Add(client.PublishAsync(topicsToManage.First(), "Hello World"));
+                            publishTasks.Add(client.PublishAsync(topicsToManage.Last(), "Hello World"));
+                            publishTasks.Add(client.PublishAsync("/unknown/topic", "Hello World"));
+                        }
+
+                        await Task.WhenAll(publishTasks).ConfigureAwait(false);
+
+                        var subscriptions = topicsToManage.Select(topic => new Subscription(new TopicFilter(topic))).ToList();
+                        var unsubscribeOptions = new UnsubscribeOptions { Subscriptions = subscriptions };
+                        _ = await client.UnsubscribeAsync(unsubscribeOptions).ConfigureAwait(false);
+
+                        _ = Interlocked.Increment(ref successCount);
+                    }
+                    catch (Exception e)
+                    {
+                        var errorMessage = $"Worker {workerId}, Iteration {i}: {e}";
+                        exceptionMessages.Add(errorMessage);
+                        this.testOutputHelper.WriteLine(errorMessage);
+                    }
+                }
+            }));
+        }
+
+        await Task.WhenAll(tasks).ConfigureAwait(false);
+
+        Assert.Equal(totalExpectedSuccesses, successCount);
+        Assert.Empty(exceptionMessages);
+    }
+}