Synchronized dispose and subscription initialization within the Messa…

…geBus to avoid memory leaks

SignalR#1993

src/Microsoft.AspNet.SignalR.Core/Messaging/MessageBus.cs

@@ -241,7 +241,8 @@ public virtual IDisposable Subscribe(ISubscriber subscriber, string cursor, Func
             subscriber.EventKeyRemoved += _removeEvent;
             subscriber.WriteCursor = subscription.WriteCursor;
 
-            var disposable = new DisposableAction(_disposeSubscription, subscriber);
+            var subscriptionState = new SubscriptionState(subscriber);
+            var disposable = new DisposableAction(_disposeSubscription, subscriptionState);
 
             // When the subscription itself is disposed then dispose it
             subscription.Disposable = disposable;
@@ -254,6 +255,8 @@ public virtual IDisposable Subscribe(ISubscriber subscriber, string cursor, Func
                 topic.AddSubscription(subscription);
             }
 
+            subscriptionState.Initialized.Set();
+
             // If there's a cursor then schedule work for this subscription
             if (!String.IsNullOrEmpty(cursor))
             {
@@ -491,7 +494,8 @@ private void RemoveEvent(ISubscriber subscriber, string eventKey)
         [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes", Justification = "Failure to invoke the callback should be ignored")]
         private void DisposeSubscription(object state)
         {
-            var subscriber = (ISubscriber)state;
+            var subscriptionState = (SubscriptionState)state;
+            var subscriber = subscriptionState.Subscriber;
 
             // This will stop work from continuting to happen
             subscriber.Subscription.Dispose();
@@ -507,6 +511,8 @@ private void DisposeSubscription(object state)
                 // so the terminal message isn't required.
             }
 
+            subscriptionState.Initialized.Wait();
+
             subscriber.EventKeyAdded -= _addEvent;
             subscriber.EventKeyRemoved -= _removeEvent;
             subscriber.WriteCursor = null;
@@ -517,5 +523,17 @@ private void DisposeSubscription(object state)
                 RemoveEvent(subscriber, eventKey);
             }
         }
+
+        private class SubscriptionState
+        {
+            public ISubscriber Subscriber { get; private set; }
+            public ManualResetEventSlim Initialized { get; private set; }
+
+            public SubscriptionState(ISubscriber subscriber)
+            {
+                Initialized = new ManualResetEventSlim();
+                Subscriber = subscriber;
+            }
+        }
     }
 }