Fix cross access to async events. (#1515)

fix cross access to async events.

Source/MQTTnet/Internal/AsyncEvent.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics;
 
@@ -13,7 +14,16 @@ namespace MQTTnet.Internal
     {
         readonly List<AsyncEventInvocator<TEventArgs>> _handlers = new List<AsyncEventInvocator<TEventArgs>>();
 
-        public bool HasHandlers => _handlers.Count > 0;
+        ICollection<AsyncEventInvocator<TEventArgs>> _handlersForInvoke;
+
+        public AsyncEvent()
+        {
+            _handlersForInvoke = _handlers;
+        }
+
+        // Track the existence of handlers in a separate field so that checking it all the time will not
+        // require locking the actual list (_handlers).
+        public bool HasHandlers { get; private set; }
 
         public void AddHandler(Func<TEventArgs, Task> handler)
         {
@@ -22,7 +32,13 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            _handlers.Add(new AsyncEventInvocator<TEventArgs>(null, handler));
+            lock (_handlers)
+            {
+                _handlers.Add(new AsyncEventInvocator<TEventArgs>(null, handler));
+
+                HasHandlers = true;
+                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
+            }
         }
 
         public void AddHandler(Action<TEventArgs> handler)
@@ -32,17 +48,32 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            _handlers.Add(new AsyncEventInvocator<TEventArgs>(handler, null));
+            lock (_handlers)
+            {
+                _handlers.Add(new AsyncEventInvocator<TEventArgs>(handler, null));
+
+                HasHandlers = true;
+                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
+            }
         }
 
         public async Task InvokeAsync(TEventArgs eventArgs)
         {
-            foreach (var handler in _handlers)
+            if (!HasHandlers)
+            {
+                return;
+            }
+
+            // Adding or removing handlers will produce a new list instance all the time.
+            // So locking here is not required since only the reference to an immutable list
+            // of handlers is used.
+            var handlers = _handlersForInvoke;
+            foreach (var handler in handlers)
             {
                 await handler.InvokeAsync(eventArgs).ConfigureAwait(false);
             }
         }
-        
+
         public void RemoveHandler(Func<TEventArgs, Task> handler)
         {
             if (handler == null)
@@ -50,7 +81,13 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            _handlers.RemoveAll(h => h.WrapsHandler(handler));
+            lock (_handlers)
+            {
+                _handlers.RemoveAll(h => h.WrapsHandler(handler));
+
+                HasHandlers = _handlers.Count > 0;
+                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
+            }
         }
 
         public void RemoveHandler(Action<TEventArgs> handler)
@@ -60,7 +97,13 @@ namespace MQTTnet.Internal
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            _handlers.RemoveAll(h => h.WrapsHandler(handler));
+            lock (_handlers)
+            {
+                _handlers.RemoveAll(h => h.WrapsHandler(handler));
+
+                HasHandlers = _handlers.Count > 0;
+                _handlersForInvoke = new List<AsyncEventInvocator<TEventArgs>>(_handlers.ToList());
+            }
         }
 
         public async Task TryInvokeAsync(TEventArgs eventArgs, MqttNetSourceLogger logger)