Fix AsyncQueue handling of cancelled tasks

This change fixes an issue in AsyncQueue where cancelled tasks were not
being skipped when new values are enqueued.  This primarily affected the
queueing of RunspaceHandle access in PowerShellContext.

Author: daviwil

src/PowerShellEditorServices/Utility/AsyncQueue.cs

@@ -73,18 +73,24 @@ public async Task EnqueueAsync(T item)
         {
             using (await queueLock.LockAsync())
             {
-                if (this.requestQueue.Count > 0)
-                {
-                    // There are requests waiting, immediately dispatch the item
-                    TaskCompletionSource<T> requestTaskSource = this.requestQueue.Dequeue();
-                    requestTaskSource.SetResult(item);
-                }
-                else
+                TaskCompletionSource<T> requestTaskSource = null;
+
+                // Are any requests waiting?
+                while (this.requestQueue.Count > 0)
                 {
-                    // No requests waiting, queue the item for a later request
-                    this.itemQueue.Enqueue(item);
-                    this.IsEmpty = false;
+                    // Is the next request cancelled already?
+                    requestTaskSource = this.requestQueue.Dequeue();
+                    if (!requestTaskSource.Task.IsCanceled)
+                    {
+                        // Dispatch the item
+                        requestTaskSource.SetResult(item);
+                        return;
+                    }
                 }
+                               
+                // No more requests waiting, queue the item for a later request
+                this.itemQueue.Enqueue(item);
+                this.IsEmpty = false;
             }
         }
 

test/PowerShellEditorServices.Test/Utility/AsyncQueueTests.cs

@@ -56,6 +56,26 @@ await Task.WhenAll(
             Assert.Equal(0, expectedItems.Except(outputItems).Count());
         }
 
+        [Fact]
+        public async Task AsyncQueueSkipsCancelledTasks()
+        {
+            AsyncQueue<int> inputQueue = new AsyncQueue<int>();
+
+            // Queue up a couple of tasks to wait for input
+            CancellationTokenSource cancellationSource = new CancellationTokenSource();
+            Task<int> taskOne = inputQueue.DequeueAsync(cancellationSource.Token);
+            Task<int> taskTwo = inputQueue.DequeueAsync();
+
+            // Cancel the first task and then enqueue a number
+            cancellationSource.Cancel();
+            await inputQueue.EnqueueAsync(1);
+
+            // Did the second task get the number?
+            Assert.Equal(TaskStatus.Canceled, taskOne.Status);
+            Assert.Equal(TaskStatus.RanToCompletion, taskTwo.Status);
+            Assert.Equal(1, taskTwo.Result);
+        }
+
         private async Task ConsumeItems(
             AsyncQueue<int> inputQueue,
             ConcurrentBag<int> outputItems,