Updated scale-out stream unit tests:

- Added some more tests to verify behavior of queuing behavior

tests/Microsoft.AspNet.SignalR.Tests/Server/ScaleoutStreamFacts.cs

@@ -1,28 +1,35 @@
 using System;
 using System.Diagnostics;
 using System.Threading.Tasks;
+using Microsoft.AspNet.SignalR.Infrastructure;
 using Microsoft.AspNet.SignalR.Messaging;
+using Moq;
 using Xunit;
+using Xunit.Extensions;
 
 namespace Microsoft.AspNet.SignalR.Tests.Server
 {
     public class ScaleoutStreamFacts
     {
         [Fact]
-        public void SendBeforeOpenDoesNotThrow()
+        public void SendBeforeOpenDoesNotThrowWhenQueuingBehaviorInitialOnly()
         {
             var perfCounters = new Microsoft.AspNet.SignalR.Infrastructure.PerformanceCounterManager();
-            var stream = new ScaleoutStream(new TraceSource("Queue"), "0", QueuingBehavior.Always, 1000, perfCounters);
+            var stream = new ScaleoutStream(new TraceSource("Queue"), "0", QueuingBehavior.InitialOnly, 1000, perfCounters);
+
+            var sendTask = stream.Send(_ => { return TaskAsyncHelper.Empty; }, null);
 
-            Assert.DoesNotThrow(() => stream.Send(_ => { return TaskAsyncHelper.Empty; }, null));
+            stream.Open();
+
+            Assert.DoesNotThrow(() => sendTask.Wait());
         }
 
         [Fact]
-        public void SendBeforeOpenQueues()
+        public void SendsBeforeOpenedRunOnceOpenedWhenQueuingBehaviorInitialOnly()
         {
             var perfCounters = new Microsoft.AspNet.SignalR.Infrastructure.PerformanceCounterManager();
-            var stream = new ScaleoutStream(new TraceSource("Queue"), "0", QueuingBehavior.Always, 1000, perfCounters);
-
+            var stream = new ScaleoutStream(new TraceSource("Queue"), "0", QueuingBehavior.InitialOnly, 1000, perfCounters);
+            
             int x = 0;
 
             stream.Send(_ =>
@@ -42,10 +49,70 @@ public void SendBeforeOpenQueues()
             stream.Open();
 
             task.Wait();
-
+            
             Assert.Equal(2, x);
         }
 
+        [Theory]
+        [InlineData(QueuingBehavior.InitialOnly, 1)]
+        [InlineData(QueuingBehavior.Always, 2)]
+        public void SendAfterOpenDoesBehavesCorrectlyForConfiguredQueuingBehavior(QueuingBehavior queuingBehavior, int expectedCounter)
+        {
+            var perfCounters = new Mock<IPerformanceCounterManager>();
+            var queueLengthCounter = new Mock<IPerformanceCounter>();
+            var counterIncrementCount = 0;
+            queueLengthCounter.Setup(c => c.Increment()).Callback(() => counterIncrementCount++);
+            perfCounters.DefaultValue = DefaultValue.Mock;
+            perfCounters.SetReturnsDefault(new NoOpPerformanceCounter());
+            perfCounters.SetupAllProperties();
+            perfCounters.Setup(pc => pc.ScaleoutSendQueueLength).Returns(queueLengthCounter.Object);
+
+            var stream = new ScaleoutStream(new TraceSource("Queue"), "0", queuingBehavior, 1000, perfCounters.Object);
+
+            stream.Send(_ => TaskAsyncHelper.Empty, null);
+
+            stream.Open();
+
+            stream.Send(_ => TaskAsyncHelper.Empty, null).Wait();
+
+            Assert.Equal(expectedCounter, counterIncrementCount);
+        }
+
+        [Fact]
+        public void SendAfterOpenAndAfterErrorThrows()
+        {
+            var perfCounters = new Microsoft.AspNet.SignalR.Infrastructure.PerformanceCounterManager();
+            var stream = new ScaleoutStream(new TraceSource("Queue"), "0", QueuingBehavior.InitialOnly, 1000, perfCounters);
+
+            int x = 0;
+
+            var firstSend = stream.Send(_ =>
+            {
+                x++;
+                return TaskAsyncHelper.Empty;
+            },
+            null);
+
+            stream.Open();
+
+            // Wait on the first send to drain
+            firstSend.Wait();
+
+            stream.SetError(new Exception("Failed"));
+
+            Assert.Throws<Exception>(() =>
+            {
+                stream.Send(_ =>
+                {
+                    x++;
+                    return TaskAsyncHelper.Empty;
+                },
+                null).Wait();
+            });
+
+            Assert.Equal(1, x);
+        }
+
         [Fact]
         public void ErrorOnSendThrowsNextTime()
         {