Don't use Task.Factory.FromAsync in StreamExtensions

- We have seen WriteAsync hang #1874
SignalR · Apr 17, 2013 · 0e0a571 · 0e0a571
1 parent eeca9f0
commit 0e0a571
Showing 1 changed file with 51 additions and 11 deletions.
diff --git a/src/Microsoft.AspNet.SignalR.Client/Infrastructure/StreamExtensions.cs b/src/Microsoft.AspNet.SignalR.Client/Infrastructure/StreamExtensions.cs
@@ -15,33 +15,73 @@ public static Task<int> ReadAsync(this Stream stream, byte[] buffer)
 #if NETFX_CORE || NET45
             return stream.ReadAsync(buffer, 0, buffer.Length);
 #else
+            return FromAsync(cb => stream.BeginRead(buffer, 0, buffer.Length, cb, null), ar => stream.EndRead(ar));
+#endif
+        }
+
+        [SuppressMessage("Microsoft.Performance", "CA1811:AvoidUncalledPrivateCode", Justification = "This is a shared class.")]
+        public static Task WriteAsync(this Stream stream, byte[] buffer)
+        {
+#if NETFX_CORE || NET45
+            return stream.WriteAsync(buffer, 0, buffer.Length);
+#else
+            return FromAsync(cb => stream.BeginWrite(buffer, 0, buffer.Length, cb, null), WrapEndRead(stream));
+#endif
+        }
+
+#if !(NETFX_CORE || NET45)
+        private static Func<IAsyncResult, object> WrapEndRead(Stream stream)
+        {
+            return ar =>
+            {
+                stream.EndWrite(ar);
+                return null;
+            };
+        }
+
+        [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes", Justification = "Exceptions are flowed back to the caller.")]
+        private static Task<T> FromAsync<T>(Func<AsyncCallback, IAsyncResult> begin, Func<IAsyncResult, T> end)
+        {
+            var tcs = new TaskCompletionSource<T>();
             try
             {
-                return Task.Factory.FromAsync((cb, state) => stream.BeginRead(buffer, 0, buffer.Length, cb, state), ar => stream.EndRead(ar), null);
+                var result = begin(ar =>
+                {
+                    if (!ar.CompletedSynchronously)
+                    {
+                        CompleteAsync(tcs, ar, end);
+                    }
+                });
+
+                if (result.CompletedSynchronously)
+                {
+                    CompleteAsync(tcs, result, end);
+                }
             }
             catch (Exception ex)
             {
-                return TaskAsyncHelper.FromError<int>(ex);
+                tcs.TrySetException(ex);
             }
-#endif
+
+            return tcs.Task;
         }
 
-        [SuppressMessage("Microsoft.Performance", "CA1811:AvoidUncalledPrivateCode", Justification = "This is a shared class.")]
         [SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes", Justification = "Exceptions are flowed back to the caller.")]
-        public static Task WriteAsync(this Stream stream, byte[] buffer)
+        private static void CompleteAsync<T>(TaskCompletionSource<T> tcs, IAsyncResult ar, Func<IAsyncResult, T> end)
         {
-#if NETFX_CORE || NET45
-            return stream.WriteAsync(buffer, 0, buffer.Length);
-#else
             try
             {
-                return Task.Factory.FromAsync((cb, state) => stream.BeginWrite(buffer, 0, buffer.Length, cb, state), ar => stream.EndWrite(ar), null);
+                tcs.TrySetResult(end(ar));
+            }
+            catch (OperationCanceledException)
+            {
+                tcs.SetCanceled();
             }
             catch (Exception ex)
             {
-                return TaskAsyncHelper.FromError(ex);
+                tcs.TrySetException(ex);
             }
-#endif
         }
+#endif
     }
 }