Merge pull request #191 from Sonarr/signalr-1-2-2

Upgraded SignalR to 1.2.2

src/Microsoft.AspNet.SignalR.Core/AuthorizeAttribute.cs

@@ -122,7 +122,7 @@ namespace Microsoft.AspNet.SignalR
         {
             if (user == null)
             {
-                throw new ArgumentNullException("user");
+                return false;
             }
 
             if (!user.Identity.IsAuthenticated)

src/Microsoft.AspNet.SignalR.Core/Hosting/HostDependencyResolverExtensions.cs

@@ -20,8 +20,12 @@ namespace Microsoft.AspNet.SignalR.Hosting
                 throw new ArgumentNullException("instanceName");
             }
 
+            // Performance counters are broken on mono so just skip this step
+            if (!MonoUtility.IsRunningMono)
+            {
                 // Initialize the performance counters
                 resolver.InitializePerformanceCounters(instanceName, hostShutdownToken);
+            }
 
             // Dispose the dependency resolver on host shut down (cleanly)
             resolver.InitializeResolverDispose(hostShutdownToken);
@@ -41,12 +45,11 @@ namespace Microsoft.AspNet.SignalR.Hosting
             // TODO: Guard against multiple calls to this
 
             // When the host triggers the shutdown token, dispose the resolver
-            hostShutdownToken.Register(state =>
+            hostShutdownToken.SafeRegister(state =>
             {
                 ((IDependencyResolver)state).Dispose();
             },
-            resolver,
-            useSynchronizationContext: false);
+            resolver);
         }
     }
 }

src/Microsoft.AspNet.SignalR.Core/Hosting/IResponse.cs

@@ -17,6 +17,11 @@ namespace Microsoft.AspNet.SignalR.Hosting
         /// </summary>
         CancellationToken CancellationToken { get; }
 
+        /// <summary>
+        /// Gets or sets the status code of the response.
+        /// </summary>
+        int StatusCode { get; set; }
+
         /// <summary>
         /// Gets or sets the content type of the response.
         /// </summary>

src/Microsoft.AspNet.SignalR.Core/Hosting/IWebSocket.cs

@@ -16,9 +16,9 @@ namespace Microsoft.AspNet.SignalR.Hosting
         Action<string> OnMessage { get; set; }
 
         /// <summary>
-        /// Invoked when the websocket gracefully closes
+        /// Invoked when the websocket closes
         /// </summary>
-        Action<bool> OnClose { get; set; }
+        Action OnClose { get; set; }
 
         /// <summary>
         /// Invoked when there is an error

src/Microsoft.AspNet.SignalR.Core/Hosting/IWebSocketRequest.cs

@@ -11,6 +11,7 @@ namespace Microsoft.AspNet.SignalR.Hosting
         /// Accepts an websocket request using the specified user function.
         /// </summary>
         /// <param name="callback">The callback that fires when the websocket is ready.</param>
-        Task AcceptWebSocketRequest(Func<IWebSocket, Task> callback);
+        /// <param name="initTask">The task that completes when the websocket transport is ready.</param>
+        Task AcceptWebSocketRequest(Func<IWebSocket, Task> callback, Task initTask);
     }
 }

src/Microsoft.AspNet.SignalR.Core/Hubs/HubDispatcher.cs

@@ -358,7 +358,7 @@ namespace Microsoft.AspNet.SignalR.Hubs
         private Task ExecuteHubEvent(IRequest request, string connectionId, Func<IHub, Task> action)
         {
             var hubs = GetHubs(request, connectionId).ToList();
-            var operations = hubs.Select(instance => action(instance).Catch().OrEmpty()).ToArray();
+            var operations = hubs.Select(instance => action(instance).OrEmpty().Catch()).ToArray();
 
             if (operations.Length == 0)
             {

src/Microsoft.AspNet.SignalR.Core/Infrastructure/BinaryTextWriter.cs

@@ -0,0 +1,34 @@
+using System;
+using Microsoft.AspNet.SignalR.Hosting;
+
+namespace Microsoft.AspNet.SignalR.Infrastructure
+{
+    /// <summary>
+    /// A buffering text writer that supports writing binary directly as well
+    /// </summary>
+    internal unsafe class BinaryTextWriter : BufferTextWriter, IBinaryWriter
+    {
+        public BinaryTextWriter(IResponse response) :
+            base((data, state) => ((IResponse)state).Write(data), response, reuseBuffers: true, bufferSize: 128)
+        {
+
+        }
+
+        public BinaryTextWriter(IWebSocket socket) :
+            base((data, state) => ((IWebSocket)state).SendChunk(data), socket, reuseBuffers: false, bufferSize: 1024)
+        {
+
+        }
+
+
+        public BinaryTextWriter(Action<ArraySegment<byte>, object> write, object state, bool reuseBuffers, int bufferSize) :
+            base(write, state, reuseBuffers, bufferSize)
+        {
+        }
+
+        public void Write(ArraySegment<byte> data)
+        {
+            Writer.Write(data);
+        }
+    }
+}

src/Microsoft.AspNet.SignalR.Core/Infrastructure/BufferTextWriter.cs

@@ -13,7 +13,7 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
     /// we don't need to write to a long lived buffer. This saves massive amounts of memory
     /// as the number of connections grows.
     /// </summary>
-    internal unsafe class BufferTextWriter : TextWriter, IBinaryWriter
+    internal abstract unsafe class BufferTextWriter : TextWriter
     {
         private readonly Encoding _encoding;
 
@@ -31,13 +31,13 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
         }
 
         public BufferTextWriter(IWebSocket socket) :
-            this((data, state) => ((IWebSocket)state).SendChunk(data), socket, reuseBuffers: false, bufferSize: 128)
+            this((data, state) => ((IWebSocket)state).SendChunk(data), socket, reuseBuffers: false, bufferSize: 1024 * 4)
         {
 
         }
 
         [SuppressMessage("Microsoft.Globalization", "CA1305:SpecifyIFormatProvider", MessageId = "System.IO.TextWriter.#ctor", Justification = "It won't be used")]
-        public BufferTextWriter(Action<ArraySegment<byte>, object> write, object state, bool reuseBuffers, int bufferSize)
+        protected BufferTextWriter(Action<ArraySegment<byte>, object> write, object state, bool reuseBuffers, int bufferSize)
         {
             _write = write;
             _writeState = state;
@@ -46,7 +46,7 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
             _bufferSize = bufferSize;
         }
 
-        private ChunkedWriter Writer
+        protected internal ChunkedWriter Writer
         {
             get
             {
@@ -79,17 +79,12 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
             Writer.Write(value);
         }
 
-        public void Write(ArraySegment<byte> data)
-        {
-            Writer.Write(data);
-        }
-
         public override void Flush()
         {
             Writer.Flush();
         }
 
-        private class ChunkedWriter
+        internal class ChunkedWriter
         {
             private int _charPos;
             private int _charLen;

src/Microsoft.AspNet.SignalR.Core/Infrastructure/CancellationTokenExtensions.cs

@@ -1,20 +1,25 @@
 // Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.md in the project root for license information.
 
 using System;
+using System.Diagnostics.CodeAnalysis;
+using System.Linq.Expressions;
+using System.Reflection;
 using System.Threading;
 
 namespace Microsoft.AspNet.SignalR.Infrastructure
 {
     internal static class CancellationTokenExtensions
     {
+        private delegate CancellationTokenRegistration RegisterDelegate(ref CancellationToken token, Action<object> callback, object state);
+
+        private static readonly RegisterDelegate _tokenRegister = ResolveRegisterDelegate();
+
         public static IDisposable SafeRegister(this CancellationToken cancellationToken, Action<object> callback, object state)
         {
             var callbackWrapper = new CancellationCallbackWrapper(callback, state);
 
             // Ensure delegate continues to use the C# Compiler static delegate caching optimization.
-            CancellationTokenRegistration registration = cancellationToken.Register(s => Cancel(s),
-                                                                                    callbackWrapper,
-                                                                                    useSynchronizationContext: false);
+            CancellationTokenRegistration registration = _tokenRegister(ref cancellationToken, s => InvokeCallback(s), callbackWrapper);
 
             var disposeCancellationState = new DiposeCancellationState(callbackWrapper, registration);
 
@@ -22,7 +27,7 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
             return new DisposableAction(s => Dispose(s), disposeCancellationState);
         }
 
-        private static void Cancel(object state)
+        private static void InvokeCallback(object state)
         {
             ((CancellationCallbackWrapper)state).TryInvoke();
         }
@@ -32,6 +37,56 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
             ((DiposeCancellationState)state).TryDispose();
         }
 
+        [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes", Justification = "This method should never throw since it runs as part of field initialzation")]
+        private static RegisterDelegate ResolveRegisterDelegate()
+        {
+            // The fallback is just a normal register that capatures the execution context.
+            RegisterDelegate fallback = (ref CancellationToken token, Action<object> callback, object state) =>
+            {
+                return token.Register(callback, state);
+            };
+
+#if NETFX_CORE || PORTABLE
+            return fallback;
+#else
+
+            MethodInfo methodInfo = null;
+
+            try
+            {
+                // By default we don't want to capture the execution context,
+                // since this is internal we need to create a delegate to this up front
+                methodInfo = typeof(CancellationToken).GetMethod("InternalRegisterWithoutEC",
+                                                                 BindingFlags.NonPublic | BindingFlags.Instance,
+                                                                 binder: null,
+                                                                 types: new[] { typeof(Action<object>), typeof(object) },
+                                                                 modifiers: null);
+            }
+            catch
+            {
+                // Swallow this exception. Being extra paranoid, we don't want anything to break in case this dirty
+                // reflection hack fails for any reason
+            }
+
+            // If the method was removed then fallback to the regular method
+            if (methodInfo == null)
+            {
+                return fallback;
+            }
+
+            try
+            {
+
+                return (RegisterDelegate)Delegate.CreateDelegate(typeof(RegisterDelegate), null, methodInfo);
+            }
+            catch
+            {
+                // If this fails for whatever reason just fallback to normal register
+                return fallback;
+            }
+#endif
+        }
+
         private class DiposeCancellationState
         {
             private readonly CancellationCallbackWrapper _callbackWrapper;
@@ -48,9 +103,15 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
                 // This normally waits until the callback is finished invoked but we don't care
                 if (_callbackWrapper.TrySetInvoked())
                 {
-                    // Bug #1549, .NET 4.0 has a bug where this throws if the CTS
+                    try
+                    {
                         _registration.Dispose();
                     }
+                    catch (ObjectDisposedException)
+                    {
+                        // Bug #1549, .NET 4.0 has a bug where this throws if the CTS is disposed.
+                    }
+                }
             }
         }
 

src/Microsoft.AspNet.SignalR.Core/Infrastructure/Connection.cs

@@ -144,7 +144,7 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
         {
             using (var stream = new MemoryStream(128))
             {
-                var bufferWriter = new BufferTextWriter((buffer, state) =>
+                var bufferWriter = new BinaryTextWriter((buffer, state) =>
                 {
                     ((MemoryStream)state).Write(buffer.Array, buffer.Offset, buffer.Count);
                 },
@@ -236,8 +236,7 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
 
                                                       if (command == null)
                                                       {
-                                                          var platform = (int)Environment.OSVersion.Platform;
-                                                          if (platform == 4 || platform == 6 || platform == 128)
+                                                          if (MonoUtility.IsRunningMono)
                                                           {
                                                               return;
                                                           }

src/Microsoft.AspNet.SignalR.Core/Infrastructure/MonoUtility.cs

@@ -0,0 +1,31 @@
+using System;
+using System.Diagnostics.CodeAnalysis;
+
+namespace Microsoft.AspNet.SignalR.Infrastructure
+{
+    internal static class MonoUtility
+    {
+        private static readonly Lazy<bool> _isRunningMono = new Lazy<bool>(() => CheckRunningOnMono());
+
+        internal static bool IsRunningMono
+        {
+            get
+            {
+                return _isRunningMono.Value;
+            }
+        }
+
+        [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes", Justification = "This should never fail")]
+        private static bool CheckRunningOnMono()
+        {
+            try
+            {
+                return Type.GetType("Mono.Runtime") != null;
+            }
+            catch
+            {
+                return false;
+            }
+        }
+    }
+}

src/Microsoft.AspNet.SignalR.Core/Infrastructure/TaskQueue.cs

@@ -35,7 +35,7 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
             _maxSize = maxSize;
         }
 
-#if !CLIENT_NET45
+#if !CLIENT_NET45 && !CLIENT_NET4 && !NETFX_CORE && !SILVERLIGHT
         [SuppressMessage("Microsoft.Performance", "CA1811:AvoidUncalledPrivateCode", Justification = "This is shared code.")]
         public IPerformanceCounter QueueSizeCounter { get; set; }
 #endif
@@ -62,19 +62,16 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
 
                 if (_maxSize != null)
                 {
-                    if (Interlocked.Read(ref _size) == _maxSize)
+                    // Increment the size if the queue
+                    if (Interlocked.Increment(ref _size) > _maxSize)
                     {
-                        // REVIEW: Do we need to make the contract more clear between the
-                        // queue full case and the queue drained case? Should we throw an exeception instead?
+                        Interlocked.Decrement(ref _size);
 
                         // We failed to enqueue because the size limit was reached
                         return null;
                     }
 
-                    // Increment the size if the queue
-                    Interlocked.Increment(ref _size);
-                    
-#if !CLIENT_NET45
+#if !CLIENT_NET45 && !CLIENT_NET4 && !NETFX_CORE && !SILVERLIGHT
                     var counter = QueueSizeCounter;
                     if (counter != null)
                     {
@@ -93,7 +90,7 @@ namespace Microsoft.AspNet.SignalR.Infrastructure
                             // Decrement the number of items left in the queue
                             Interlocked.Decrement(ref queue._size);
 
-#if !CLIENT_NET45
+#if !CLIENT_NET45 && !CLIENT_NET4 && !NETFX_CORE && !SILVERLIGHT
                             var counter = QueueSizeCounter;
                             if (counter != null)
                             {

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

@@ -33,8 +33,10 @@ namespace Microsoft.AspNet.SignalR.Messaging
             _escapedKey = minifiedKey;
         }
 
-        public static void WriteCursors(TextWriter textWriter, IList<Cursor> cursors)
+        public static void WriteCursors(TextWriter textWriter, IList<Cursor> cursors, string prefix)
         {
+            textWriter.Write(prefix);
+
             for (int i = 0; i < cursors.Count; i++)
             {
                 if (i > 0)
@@ -48,7 +50,7 @@ namespace Microsoft.AspNet.SignalR.Messaging
             }
         }
 
-        private static void WriteUlongAsHexToBuffer(ulong value, TextWriter textWriter)
+        internal static void WriteUlongAsHexToBuffer(ulong value, TextWriter textWriter)
         {
             // This tracks the length of the output and serves as the index for the next character to be written into the pBuffer.
             // The length could reach up to 16 characters, so at least that much space should remain in the pBuffer.
@@ -114,17 +116,17 @@ namespace Microsoft.AspNet.SignalR.Messaging
             return sb.ToString();
         }
 
-        public static List<Cursor> GetCursors(string cursor)
+        public static List<Cursor> GetCursors(string cursor, string prefix)
         {
-            return GetCursors(cursor, s => s);
+            return GetCursors(cursor, prefix, s => s);
         }
 
-        public static List<Cursor> GetCursors(string cursor, Func<string, string> keyMaximizer)
+        public static List<Cursor> GetCursors(string cursor, string prefix, Func<string, string> keyMaximizer)
         {
-            return GetCursors(cursor, (key, state) => ((Func<string, string>)state).Invoke(key), keyMaximizer);
+            return GetCursors(cursor, prefix, (key, state) => ((Func<string, string>)state).Invoke(key), keyMaximizer);
         }
 
-        public static List<Cursor> GetCursors(string cursor, Func<string, object, string> keyMaximizer, object state)
+        public static List<Cursor> GetCursors(string cursor, string prefix, Func<string, object, string> keyMaximizer, object state)
         {
             // Technically GetCursors should never be called with a null value, so this is extra cautious
             if (String.IsNullOrEmpty(cursor))
@@ -132,6 +134,14 @@ namespace Microsoft.AspNet.SignalR.Messaging
                 throw new FormatException(Resources.Error_InvalidCursorFormat);
             }
 
+            // If the cursor does not begin with the prefix stream, it isn't necessarily a formatting problem.
+            // The cursor with a different prefix might have had different, but also valid, formatting.
+            // Null should be returned so new cursors will be generated
+            if (!cursor.StartsWith(prefix, StringComparison.Ordinal))
+            {
+                return null;
+            }
+
             var signals = new HashSet<string>();
             var cursors = new List<Cursor>();
             string currentKey = null;
@@ -143,8 +153,10 @@ namespace Microsoft.AspNet.SignalR.Messaging
             var sbEscaped = new StringBuilder();
             Cursor parsedCursor;
 
-            foreach (var ch in cursor)
+            for (int i = prefix.Length; i < cursor.Length; i++)
             {
+                var ch = cursor[i];
+
                 // escape can only be true if we are consuming the key
                 if (escape)
                 {

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

@@ -3,7 +3,9 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
+using System.Globalization;
 using System.IO;
+using System.Security.Cryptography;
 using System.Threading.Tasks;
 using Microsoft.AspNet.SignalR.Infrastructure;
 
@@ -11,6 +13,8 @@ namespace Microsoft.AspNet.SignalR.Messaging
 {
     internal class DefaultSubscription : Subscription
     {
+        internal static string _defaultCursorPrefix = GetCursorPrefix();
+
         private List<Cursor> _cursors;
         private List<Topic> _cursorTopics;
 
@@ -36,7 +40,7 @@ namespace Microsoft.AspNet.SignalR.Messaging
             else
             {
                 // Ensure delegate continues to use the C# Compiler static delegate caching optimization.
-                _cursors = Cursor.GetCursors(cursor, (k, s) => UnminifyCursor(k, s), stringMinifier) ?? GetCursorsFromEventKeys(EventKeys, topics);
+                _cursors = Cursor.GetCursors(cursor, _defaultCursorPrefix, (k, s) => UnminifyCursor(k, s), stringMinifier) ?? GetCursorsFromEventKeys(EventKeys, topics);
             }
 
             _cursorTopics = new List<Topic>();
@@ -126,7 +130,7 @@ namespace Microsoft.AspNet.SignalR.Messaging
         {
             lock (_cursors)
             {
-                Cursor.WriteCursors(textWriter, _cursors);
+                Cursor.WriteCursors(textWriter, _cursors, _defaultCursorPrefix);
             }
         }
 
@@ -196,6 +200,22 @@ namespace Microsoft.AspNet.SignalR.Messaging
             return list;
         }
 
+        private static string GetCursorPrefix()
+        {
+            using (var rng = new RNGCryptoServiceProvider())
+            {
+                var data = new byte[4];
+                rng.GetBytes(data);
+
+                using (var writer = new StringWriter(CultureInfo.InvariantCulture))
+                {
+                    var randomValue = (ulong)BitConverter.ToUInt32(data, 0);
+                    Cursor.WriteUlongAsHexToBuffer(randomValue, writer);
+                    return "d-" + writer.ToString() + "-";
+                }
+            }
+        }
+
         private static ulong GetMessageId(TopicLookup topics, string key)
         {
             Topic topic;

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

@@ -233,7 +233,7 @@ namespace Microsoft.AspNet.SignalR.Messaging
                     }
                     finally
                     {
-                        if (!subscription.UnsetQueued() || workTask.IsFaulted)
+                        if (!subscription.UnsetQueued() || workTask.IsFaulted || workTask.IsCanceled)
                         {
                             // If we don't have more work to do just make the subscription null
                             subscription = null;
@@ -271,7 +271,7 @@ namespace Microsoft.AspNet.SignalR.Messaging
                     Trace.TraceEvent(TraceEventType.Error, 0, "Work failed for " + subscription.Identity + ": " + task.Exception.GetBaseException());
                 }
 
-                if (moreWork && !task.IsFaulted)
+                if (moreWork && !task.IsFaulted && !task.IsCanceled)
                 {
                     PumpImpl(taskCompletionSource, subscription);
                 }
@@ -295,10 +295,7 @@ namespace Microsoft.AspNet.SignalR.Messaging
 
                     Trace.TraceEvent(TraceEventType.Verbose, 0, "Dispoing the broker");
 
-                    //Check if OS is not Windows and exit
-                    var platform = (int)Environment.OSVersion.Platform;
-
-                    if ((platform == 4) || (platform == 6) || (platform == 128))
+                    if (MonoUtility.IsRunningMono)
                     {
                         return;
                     }

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

@@ -15,6 +15,9 @@ namespace Microsoft.AspNet.SignalR.Messaging
         private static readonly List<ArraySegment<Message>> _emptyList = new List<ArraySegment<Message>>();
         public readonly static MessageResult TerminalMessage = new MessageResult(terminal: true);
 
+        /// <summary>
+        /// Gets an <see cref="T:IList{Message}"/> associated with the result.
+        /// </summary>
         [SuppressMessage("Microsoft.Design", "CA1006:DoNotNestGenericTypesInMemberSignatures", Justification = "This is an optimization to avoid allocations.")]
         public IList<ArraySegment<Message>> Messages { get; private set; }
 

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

@@ -12,6 +12,8 @@ namespace Microsoft.AspNet.SignalR.Messaging
 {
     public class ScaleoutSubscription : Subscription
     {
+        private const string _scaleoutCursorPrefix = "s-";
+
         private readonly IList<ScaleoutMappingStore> _streams;
         private readonly List<Cursor> _cursors;
 
@@ -40,10 +42,15 @@ namespace Microsoft.AspNet.SignalR.Messaging
             }
             else
             {
-                cursors = Cursor.GetCursors(cursor);
+                cursors = Cursor.GetCursors(cursor, _scaleoutCursorPrefix);
 
+                // If the cursor had a default prefix, "d-", cursors might be null
+                if (cursors == null)
+                {
+                    cursors = new List<Cursor>();
+                }
                 // If the streams don't match the cursors then throw it out
-                if (cursors.Count != _streams.Count)
+                else if (cursors.Count != _streams.Count)
                 {
                     cursors.Clear();
                 }
@@ -63,7 +70,7 @@ namespace Microsoft.AspNet.SignalR.Messaging
 
         public override void WriteCursor(TextWriter textWriter)
         {
-            Cursor.WriteCursors(textWriter, _cursors);
+            Cursor.WriteCursors(textWriter, _cursors, _scaleoutCursorPrefix);
         }
 
         [SuppressMessage("Microsoft.Design", "CA1002:DoNotExposeGenericLists", Justification = "The list needs to be populated")]

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

@@ -120,15 +120,9 @@ namespace Microsoft.AspNet.SignalR.Messaging
 
             WorkImpl(tcs);
 
-            // Fast Path
-            if (tcs.Task.IsCompleted)
-            {
             return tcs.Task;
         }
 
-            return FinishAsync(tcs);
-        }
-
         public bool SetQueued()
         {
             return Interlocked.Increment(ref _state) == State.Working;
@@ -140,19 +134,6 @@ namespace Microsoft.AspNet.SignalR.Messaging
             return Interlocked.CompareExchange(ref _state, State.Idle, State.Working) != State.Working;
         }
 
-        private static Task FinishAsync(TaskCompletionSource<object> tcs)
-        {
-            return tcs.Task.ContinueWith(task =>
-            {
-                if (task.IsFaulted)
-                {
-                    return TaskAsyncHelper.FromError(task.Exception);
-                }
-
-                return TaskAsyncHelper.Empty;
-            }).FastUnwrap();
-        }
-
         [SuppressMessage("Microsoft.Usage", "CA2202:Do not dispose objects multiple times", Justification = "We have a sync and async code path.")]
         [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes", Justification = "We want to avoid user code taking the process down.")]
         private void WorkImpl(TaskCompletionSource<object> taskCompletionSource)
@@ -199,10 +180,17 @@ namespace Microsoft.AspNet.SignalR.Messaging
                         }
                     }
                     catch (Exception ex)
+                    {
+                        if (ex.InnerException is TaskCanceledException)
+                        {
+                            taskCompletionSource.TrySetCanceled();
+                        }
+                        else
                         {
                             taskCompletionSource.TrySetUnwrappedException(ex);
                         }
                     }
+                }
                 else
                 {
                     WorkImplAsync(callbackTask, taskCompletionSource);
@@ -233,6 +221,10 @@ namespace Microsoft.AspNet.SignalR.Messaging
                 {
                     taskCompletionSource.TrySetUnwrappedException(task.Exception);
                 }
+                else if (task.IsCanceled)
+                {
+                    taskCompletionSource.TrySetCanceled();
+                }
                 else if (task.Result)
                 {
                     WorkImpl(taskCompletionSource);

src/Microsoft.AspNet.SignalR.Core/Microsoft.AspNet.SignalR.Core.csproj

@@ -71,9 +71,11 @@
     <Compile Include="Infrastructure\AckHandler.cs" />
     <Compile Include="Configuration\DefaultConfigurationManager.cs" />
     <Compile Include="Infrastructure\ArraySegmentTextReader.cs" />
+    <Compile Include="Infrastructure\BinaryTextWriter.cs" />
     <Compile Include="Infrastructure\ConnectionManager.cs" />
     <Compile Include="ConnectionMessage.cs" />
     <Compile Include="Infrastructure\DefaultProtectedData.cs" />
+    <Compile Include="Infrastructure\MonoUtility.cs" />
     <Compile Include="Infrastructure\DiffPair.cs" />
     <Compile Include="Infrastructure\DiffSet.cs" />
     <Compile Include="GlobalHost.cs" />

src/Microsoft.AspNet.SignalR.Core/PersistentConnection.cs

@@ -6,6 +6,7 @@ using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.Globalization;
 using System.Linq;
+using System.Text;
 using System.Threading.Tasks;
 using Microsoft.AspNet.SignalR.Configuration;
 using Microsoft.AspNet.SignalR.Hosting;
@@ -165,7 +166,7 @@ namespace Microsoft.AspNet.SignalR
 
             if (Transport == null)
             {
-                throw new InvalidOperationException(String.Format(CultureInfo.CurrentCulture, Resources.Error_ProtocolErrorUnknownTransport));
+                return FailResponse(context.Response, String.Format(CultureInfo.CurrentCulture, Resources.Error_ProtocolErrorUnknownTransport));
             }
 
             string connectionToken = context.Request.QueryString["connectionToken"];
@@ -173,10 +174,17 @@ namespace Microsoft.AspNet.SignalR
             // If there's no connection id then this is a bad request
             if (String.IsNullOrEmpty(connectionToken))
             {
-                throw new InvalidOperationException(String.Format(CultureInfo.CurrentCulture, Resources.Error_ProtocolErrorMissingConnectionToken));
+                return FailResponse(context.Response, String.Format(CultureInfo.CurrentCulture, Resources.Error_ProtocolErrorMissingConnectionToken));
             }
 
-            string connectionId = GetConnectionId(context, connectionToken);
+            string connectionId;
+            string message;
+            int statusCode;
+            
+            if (!TryGetConnectionId(context, connectionToken, out connectionId, out message, out statusCode))
+            {
+                return FailResponse(context.Response, message, statusCode);
+            }
 
             // Set the transport's connection id to the unprotected one
             Transport.ConnectionId = connectionId;
@@ -227,10 +235,21 @@ namespace Microsoft.AspNet.SignalR
         }
 
         [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes", Justification = "We want to catch any exception when unprotecting data.")]
-        internal string GetConnectionId(HostContext context, string connectionToken)
+        internal bool TryGetConnectionId(HostContext context,
+                                           string connectionToken,
+                                           out string connectionId,
+                                           out string message,
+                                           out int statusCode)
         {
             string unprotectedConnectionToken = null;
 
+            // connectionId is only valid when this method returns true
+            connectionId = null;
+
+            // message and statusCode are only valid when this method returns false
+            message = null;
+            statusCode = 400;
+
             try
             {
                 unprotectedConnectionToken = ProtectedData.Unprotect(connectionToken, Purposes.ConnectionToken);
@@ -242,21 +261,24 @@ namespace Microsoft.AspNet.SignalR
 
             if (String.IsNullOrEmpty(unprotectedConnectionToken))
             {
-                throw new InvalidOperationException(String.Format(CultureInfo.CurrentCulture, Resources.Error_ConnectionIdIncorrectFormat));
+                message = String.Format(CultureInfo.CurrentCulture, Resources.Error_ConnectionIdIncorrectFormat);
+                return false;
             }
 
             var tokens = unprotectedConnectionToken.Split(SplitChars, 2);
 
-            string connectionId = tokens[0];
+            connectionId = tokens[0];
             string tokenUserName = tokens.Length > 1 ? tokens[1] : String.Empty;
             string userName = GetUserIdentity(context);
 
             if (!String.Equals(tokenUserName, userName, StringComparison.OrdinalIgnoreCase))
             {
-                throw new InvalidOperationException(String.Format(CultureInfo.CurrentCulture, Resources.Error_UnrecognizedUserIdentity));
+                message = String.Format(CultureInfo.CurrentCulture, Resources.Error_UnrecognizedUserIdentity);
+                statusCode = 403;
+                return false;
             }
 
-            return connectionId;
+            return true;
         }
 
         [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes", Justification = "We want to prevent any failures in unprotecting")]
@@ -477,6 +499,12 @@ namespace Microsoft.AspNet.SignalR
             return context.Response.End(data);
         }
 
+        private static Task FailResponse(IResponse response, string message, int statusCode = 400)
+        {
+            response.StatusCode = statusCode;
+            return response.End(message);
+        }
+
         private static bool IsNegotiationRequest(IRequest request)
         {
             return request.Url.LocalPath.EndsWith("/negotiate", StringComparison.OrdinalIgnoreCase);

src/Microsoft.AspNet.SignalR.Core/Scripts/hubs.js

@@ -1,5 +1,5 @@
 /*!
- * ASP.NET SignalR JavaScript Library v1.1.3
+ * ASP.NET SignalR JavaScript Library v1.2.2
  * http://signalr.net/
  *
  * Copyright Microsoft Open Technologies, Inc. All rights reserved.
@@ -10,7 +10,7 @@
 
 /// <reference path="..\..\SignalR.Client.JS\Scripts\jquery-1.6.4.js" />
 /// <reference path="jquery.signalR.js" />
-(function ($, window) {
+(function ($, window, undefined) {
     /// <param name="$" type="jQuery" />
     "use strict";
 

src/Microsoft.AspNet.SignalR.Core/TaskAsyncHelper.cs

@@ -1,9 +1,13 @@
 // Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.md in the project root for license information.
 
 using System;
+using System.Collections;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
+using System.Globalization;
 using System.Linq;
+using System.Reflection;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNet.SignalR.Infrastructure;
@@ -159,7 +163,7 @@ namespace Microsoft.AspNet.SignalR
         {
             // observe Exception
 #if !WINDOWS_PHONE && !SILVERLIGHT && !NETFX_CORE
-            Trace.TraceError("SignalR exception thrown by Task: {0}", exception);
+            Trace.TraceWarning("SignalR exception thrown by Task: {0}", exception);
 #endif
             handler(exception, state);
         }

src/Microsoft.AspNet.SignalR.Core/Transports/ForeverTransport.cs

@@ -162,7 +162,7 @@ namespace Microsoft.AspNet.SignalR.Transports
         }
 
         [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes", Justification = "Exceptions are flowed to the caller.")]
-        private Task ProcessReceiveRequest(ITransportConnection connection)
+        protected Task ProcessReceiveRequest(ITransportConnection connection)
         {
             Func<Task> initialize = null;
 
@@ -273,7 +273,7 @@ namespace Microsoft.AspNet.SignalR.Transports
         {
             var context = (MessageContext)state;
 
-            response.TimedOut = context.Transport.IsTimedOut;
+            response.Reconnect = context.Transport.HostShutdownToken.IsCancellationRequested;
 
             // If we're telling the client to disconnect then clean up the instantiated connection.
             if (response.Disconnect)
@@ -282,7 +282,7 @@ namespace Microsoft.AspNet.SignalR.Transports
                 return context.Transport.Send(response).Then(c => OnDisconnectMessage(c), context)
                                         .Then(() => TaskAsyncHelper.False);
             }
-            else if (response.TimedOut || response.Aborted)
+            else if (context.Transport.IsTimedOut || response.Aborted)
             {
                 context.Registration.Dispose();
 

src/Microsoft.AspNet.SignalR.Core/Transports/LongPollingTransport.cs

@@ -3,6 +3,7 @@
 using System;
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNet.SignalR.Hosting;
 using Microsoft.AspNet.SignalR.Infrastructure;
@@ -252,7 +253,7 @@ namespace Microsoft.AspNet.SignalR.Transports
         {
             var context = (MessageContext)state;
 
-            response.TimedOut = context.Transport.IsTimedOut;
+            response.Reconnect = context.Transport.HostShutdownToken.IsCancellationRequested;
 
             Task task = TaskAsyncHelper.Empty;
 

src/Microsoft.AspNet.SignalR.Core/Transports/PersistentResponse.cs

@@ -20,7 +20,7 @@ namespace Microsoft.AspNet.SignalR.Transports
         private readonly Action<TextWriter> _writeCursor;
 
         public PersistentResponse()
-            : this(message => true, writer => { })
+            : this(message => false, writer => { })
         {
 
         }
@@ -61,9 +61,10 @@ namespace Microsoft.AspNet.SignalR.Transports
         public bool Aborted { get; set; }
 
         /// <summary>
-        /// True if the connection timed out.
+        /// True if the client should try reconnecting.
         /// </summary>
-        public bool TimedOut { get; set; }
+        // This is set when the host is shutting down.
+        public bool Reconnect { get; set; }
 
         /// <summary>
         /// Signed token representing the list of groups. Updates on change.
@@ -106,7 +107,7 @@ namespace Microsoft.AspNet.SignalR.Transports
                 jsonWriter.WriteValue(1);
             }
 
-            if (TimedOut)
+            if (Reconnect)
             {
                 jsonWriter.WritePropertyName("T");
                 jsonWriter.WriteValue(1);

src/Microsoft.AspNet.SignalR.Core/Transports/TransportDisconnectBase.cs

@@ -130,6 +130,14 @@ namespace Microsoft.AspNet.SignalR.Transports
             }
         }
 
+        protected CancellationToken HostShutdownToken
+        {
+            get
+            {
+                return _hostShutdownToken;
+            }
+        }
+
         public bool IsTimedOut
         {
             get
@@ -186,7 +194,7 @@ namespace Microsoft.AspNet.SignalR.Transports
 
         protected virtual TextWriter CreateResponseWriter()
         {
-            return new BufferTextWriter(Context.Response);
+            return new BinaryTextWriter(Context.Response);
         }
 
         protected void IncrementErrors()

src/Microsoft.AspNet.SignalR.Core/Transports/WebSocketTransport.cs

@@ -19,7 +19,7 @@ namespace Microsoft.AspNet.SignalR.Transports
         private bool _isAlive = true;
 
         private readonly Action<string> _message;
-        private readonly Action<bool> _closed;
+        private readonly Action _closed;
         private readonly Action<Exception> _error;
 
         public WebSocketTransport(HostContext context,
@@ -73,6 +73,12 @@ namespace Microsoft.AspNet.SignalR.Transports
         }
 
         public override Task ProcessRequest(ITransportConnection connection)
+        {
+            if (IsAbortRequest)
+            {
+                return connection.Abort(ConnectionId);
+            }
+            else
             {
                 var webSocketRequest = _context.Request as IWebSocketRequest;
 
@@ -82,6 +88,9 @@ namespace Microsoft.AspNet.SignalR.Transports
                     throw new InvalidOperationException(Resources.Error_WebSocketsNotSupported);
                 }
 
+                Connection = connection;
+                InitializePersistentState();
+
                 return webSocketRequest.AcceptWebSocketRequest(socket =>
                 {
                     _socket = socket;
@@ -89,13 +98,15 @@ namespace Microsoft.AspNet.SignalR.Transports
                     socket.OnMessage = _message;
                     socket.OnError = _error;
 
-                return ProcessRequestCore(connection);
-            });
+                    return ProcessReceiveRequest(connection);
+                },
+                InitializeTcs.Task);
+            }
         }
 
         protected override TextWriter CreateResponseWriter()
         {
-            return new BufferTextWriter(_socket);
+            return new BinaryTextWriter(_socket);
         }
 
         public override Task Send(object value)
@@ -113,6 +124,11 @@ namespace Microsoft.AspNet.SignalR.Transports
             return Send((object)response);
         }
 
+        protected internal override Task InitializeResponse(ITransportConnection connection)
+        {
+            return _socket.Send("{}");
+        }
+
         private static Task PerformSend(object state)
         {
             var context = (WebSocketTransportContext)state;
@@ -131,18 +147,11 @@ namespace Microsoft.AspNet.SignalR.Transports
             }
         }
 
-        private void OnClosed(bool clean)
+        private void OnClosed()
         {
-            Trace.TraceInformation("CloseSocket({0}, {1})", clean, ConnectionId);
-
-            // If we performed a clean disconnect then we go through the normal disconnect routine.  However,
-            // If we performed an unclean disconnect we want to mark the connection as "not alive" and let the
-            // HeartBeat clean it up.  This is to maintain consistency across the transports.
-            if (clean)
-            {
-                Abort();
-            }
+            Trace.TraceInformation("CloseSocket({0})", ConnectionId);
 
+            // Require a request to /abort to stop tracking the connection. #2195
             _isAlive = false;
         }
 

src/Microsoft.AspNet.SignalR.Owin/Handlers/CallHandler.cs

@@ -2,6 +2,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Security.Principal;
 using System.Threading.Tasks;
 using Microsoft.AspNet.SignalR.Hosting;
 using Microsoft.AspNet.SignalR.Owin.Infrastructure;
@@ -64,12 +65,22 @@ namespace Microsoft.AspNet.SignalR.Owin
             _connection.Initialize(_configuration.Resolver, hostContext);
 
             if (!_connection.Authorize(serverRequest))
+            {
+                IPrincipal user = hostContext.Request.User;
+                if (user != null && user.Identity.IsAuthenticated)
                 {
                     // If we failed to authorize the request then return a 403 since the request
                     // can't do anything
                     return EndResponse(environment, 403, "Forbidden");
                 }
                 else
+                {
+                    // If we failed to authorize the request and the user is not authenticated
+                    // then return a 401
+                    return EndResponse(environment, 401, "Unauthorized");
+                }
+            }
+            else
             {
                 return _connection.ProcessRequest(hostContext);
             }

src/Microsoft.AspNet.SignalR.Owin/ServerRequest.cs

@@ -7,6 +7,7 @@ using System.Security.Principal;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNet.SignalR.Owin.Infrastructure;
+using Microsoft.AspNet.SignalR.Hosting;
 
 namespace Microsoft.AspNet.SignalR.Owin
 {
@@ -138,15 +139,17 @@ namespace Microsoft.AspNet.SignalR.Owin
         }
 
 #if NET45
-        public Task AcceptWebSocketRequest(Func<IWebSocket, Task> callback)
+        public Task AcceptWebSocketRequest(Func<IWebSocket, Task> callback, Task initTask)
         {
             var accept = _environment.Get<Action<IDictionary<string, object>, WebSocketFunc>>(OwinConstants.WebSocketAccept);
             if (accept == null)
             {
-                throw new InvalidOperationException(Resources.Error_NotWebSocketRequest);
+                var response = new ServerResponse(_environment);
+                response.StatusCode = 400;
+                return response.End(Resources.Error_NotWebSocketRequest);
             }
 
-            var handler = new OwinWebSocketHandler(callback);
+            var handler = new OwinWebSocketHandler(callback, initTask);
             accept(null, handler.ProcessRequestAsync);
             return TaskAsyncHelper.Empty;
         }

src/Microsoft.AspNet.SignalR.Owin/ServerResponse.cs

@@ -27,6 +27,18 @@ namespace Microsoft.AspNet.SignalR.Owin
             get { return _callCancelled; }
         }
 
+        public int StatusCode 
+        {
+            get
+            {
+                return _environment.Get<int>(OwinConstants.ResponseStatusCode);
+            }
+            set
+            {
+                _environment[OwinConstants.ResponseStatusCode] = value;
+            }
+        }
+
         public string ContentType
         {
             get { return ResponseHeaders.GetHeader("Content-Type"); }

src/NzbDrone.SignalR/NoOpPerformanceCounterManager.cs

@@ -1,105 +0,0 @@
-using System.Diagnostics;
-using System.Threading;
-using Microsoft.AspNet.SignalR.Infrastructure;
-
-namespace NzbDrone.SignalR
-{
-    public class NoOpPerformanceCounterManager : IPerformanceCounterManager
-    {
-        private static readonly IPerformanceCounter noOpCounter = new NoOpPerformanceCounter();
-
-        public void Initialize(string instanceName, CancellationToken hostShutdownToken)
-        {
-
-        }
-
-        public IPerformanceCounter LoadCounter(string categoryName, string counterName, string instanceName, bool isReadOnly)
-        {
-            return noOpCounter;
-        }
-
-        public IPerformanceCounter ConnectionsConnected { get { return noOpCounter; } }
-        public IPerformanceCounter ConnectionsReconnected { get { return noOpCounter; } }
-        public IPerformanceCounter ConnectionsDisconnected { get { return noOpCounter; } }
-        public IPerformanceCounter ConnectionsCurrent { get { return noOpCounter; } }
-        public IPerformanceCounter ConnectionMessagesReceivedTotal { get { return noOpCounter; } }
-        public IPerformanceCounter ConnectionMessagesSentTotal { get { return noOpCounter; } }
-        public IPerformanceCounter ConnectionMessagesReceivedPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter ConnectionMessagesSentPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter MessageBusMessagesReceivedTotal { get { return noOpCounter; } }
-        public IPerformanceCounter MessageBusMessagesReceivedPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter ScaleoutMessageBusMessagesReceivedPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter MessageBusMessagesPublishedTotal { get { return noOpCounter; } }
-        public IPerformanceCounter MessageBusMessagesPublishedPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter MessageBusSubscribersCurrent { get { return noOpCounter; } }
-        public IPerformanceCounter MessageBusSubscribersTotal { get { return noOpCounter; } }
-        public IPerformanceCounter MessageBusSubscribersPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter MessageBusAllocatedWorkers { get { return noOpCounter; } }
-        public IPerformanceCounter MessageBusBusyWorkers { get { return noOpCounter; } }
-        public IPerformanceCounter MessageBusTopicsCurrent { get { return noOpCounter; } }
-        public IPerformanceCounter ErrorsAllTotal { get { return noOpCounter; } }
-        public IPerformanceCounter ErrorsAllPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter ErrorsHubResolutionTotal { get { return noOpCounter; } }
-        public IPerformanceCounter ErrorsHubResolutionPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter ErrorsHubInvocationTotal { get { return noOpCounter; } }
-        public IPerformanceCounter ErrorsHubInvocationPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter ErrorsTransportTotal { get { return noOpCounter; } }
-        public IPerformanceCounter ErrorsTransportPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter ScaleoutStreamCountTotal { get { return noOpCounter; } }
-        public IPerformanceCounter ScaleoutStreamCountOpen { get { return noOpCounter; } }
-        public IPerformanceCounter ScaleoutStreamCountBuffering { get { return noOpCounter; } }
-        public IPerformanceCounter ScaleoutErrorsTotal { get { return noOpCounter; } }
-        public IPerformanceCounter ScaleoutErrorsPerSec { get { return noOpCounter; } }
-        public IPerformanceCounter ScaleoutSendQueueLength { get { return noOpCounter; } }
-    }
-
-    public class NoOpPerformanceCounter : IPerformanceCounter
-    {
-        public string CounterName
-        {
-            get
-            {
-                return this.GetType().Name;
-            }
-        }
-
-        public long RawValue
-        {
-            get
-            {
-                return 0L;
-            }
-            set
-            {
-            }
-        }
-
-        public long Decrement()
-        {
-            return 0L;
-        }
-
-        public long Increment()
-        {
-            return 0L;
-        }
-
-        public long IncrementBy(long value)
-        {
-            return 0L;
-        }
-
-        public void Close()
-        {
-        }
-
-        public void RemoveInstance()
-        {
-        }
-
-        public CounterSample NextSample()
-        {
-            return CounterSample.Empty;
-        }
-    }
-}

src/NzbDrone.SignalR/NzbDrone.SignalR.csproj

@@ -50,7 +50,6 @@
     <Compile Include="..\NzbDrone.Common\Properties\SharedAssemblyInfo.cs">
       <Link>Properties\SharedAssemblyInfo.cs</Link>
     </Compile>
-    <Compile Include="NoOpPerformanceCounterManager.cs" />
     <Compile Include="NzbDronePersistentConnection.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="Serializer.cs" />

src/NzbDrone.SignalR/SignalrDependencyResolver.cs

@@ -1,6 +1,5 @@
 using System;
 using Microsoft.AspNet.SignalR;
-using Microsoft.AspNet.SignalR.Infrastructure;
 using NzbDrone.Common.Composition;
 
 namespace NzbDrone.SignalR
@@ -16,7 +15,6 @@ namespace NzbDrone.SignalR
 
         private SignalrDependencyResolver(IContainer container)
         {
-            container.RegisterSingleton(typeof(IPerformanceCounterManager), typeof(NoOpPerformanceCounterManager));
             _container = container;
         }