Make sure the queue is started before draining.

- Added unit tests for shifting from the initial state. SignalR#1832
YakhontovYaroslav · Apr 8, 2013 · 9cfc0d8 · 9cfc0d8
1 parent 2fcc7d6
commit 9cfc0d8
Show file tree

Hide file tree

Showing 2 changed files with 37 additions and 0 deletions.
diff --git a/src/Microsoft.AspNet.SignalR.Core/Messaging/ScaleoutTaskQueue.cs b/src/Microsoft.AspNet.SignalR.Core/Messaging/ScaleoutTaskQueue.cs
@@ -127,6 +127,9 @@ public void Close()
             {
                 if (ChangeState(QueueState.Closed))
                 {
+                    // Ensure the queue is started
+                    EnsureQueueStarted();
+
                     // Drain the queue to stop all sends
                     task = Drain(_queue);
                 }
@@ -170,6 +173,8 @@ private void InitializeCore()
 
         private Task DrainQueue()
         {
+            EnsureQueueStarted();
+
             _taskCompletionSource = new TaskCompletionSource<object>();
 
             if (_queue != null)
@@ -182,6 +187,14 @@ private Task DrainQueue()
             return _taskCompletionSource.Task;
         }
 
+        private void EnsureQueueStarted()
+        {
+            if (_taskCompletionSource != null)
+            {
+                _taskCompletionSource.TrySetResult(null);
+            }
+        }
+
         private bool ChangeState(QueueState newState)
         {
             if (_state != newState)

diff --git a/tests/Microsoft.AspNet.SignalR.Tests/Server/ScaleoutTaskQueueFacts.cs b/tests/Microsoft.AspNet.SignalR.Tests/Server/ScaleoutTaskQueueFacts.cs
@@ -117,5 +117,29 @@ public void SendAfterCloseThenOpenRemainsClosed()
             queue.Open();
             Assert.Throws<InvalidOperationException>(() => queue.Enqueue(_ => TaskAsyncHelper.Empty, null));
         }
+
+        [Fact]
+        public void InitialToBufferingToOpenToSend()
+        {
+            int x = 0;
+            var queue = new ScaleoutTaskQueue(new TraceSource("Queue"), "0", new ScaleoutConfiguration() { RetryOnError = true });
+            queue.SetError(new Exception());
+            queue.Open();
+            queue.Enqueue(_ =>
+            {
+                x++;
+                return TaskAsyncHelper.Empty;
+            },
+            null).Wait();
+
+            Assert.Equal(1, x);
+        }
+
+        [Fact(Timeout = 1000)]
+        public void InitialToClosed()
+        {
+            var queue = new ScaleoutTaskQueue(new TraceSource("Queue"), "0", new ScaleoutConfiguration() { RetryOnError = true });
+            queue.Close();
+        }
     }
 }