HttpWebRequest 异步调用填满了堆栈，并剥离了 Windows Phone 上的响应式扩展

Question

我正在编写一些将文件从网络资源下载到磁盘的代码。从网络读取和写入都是异步完成的。我观察到一个问题，即异步调用实际上是同步调用的，因此每次新的迭代都会在堆栈上创建一个新的函数调用。这是代码：

private IsolatedStorageFileStream fileStream = null; 
private HttpWebRequest webRequest = null; 
private Stream responseStream = null; 
private long responsePosition = 0; 
private static int BufferSize = 4096; 
private byte[] bufferRead = new byte[BufferSize]; 
private void button4_Click(object sender, RoutedEventArgs e) 
{ 
    string fileName = "TestFile.mp3"; 
    using( var store = IsolatedStorageFile.GetUserStoreForApplication() ) 
    { 
        if( store.FileExists(fileName) ) 
        { 
            store.DeleteFile(fileName); 
        } 
        fileStream = store.OpenFile(fileName, FileMode.CreateNew, FileAccess.Write, FileShare.Read); 
        webRequest = WebRequest.Create(new Uri(mpsUri.Text)) as HttpWebRequest; 
        var observableRequest = Observable.FromAsyncPattern<WebResponse>(webRequest.BeginGetResponse, webRequest.EndGetResponse); 
        Observable.Timeout(observableRequest.Invoke(), TimeSpan.FromMinutes(2)) 
            .Subscribe(response => { ResponseCallback(response); }, exception => { TimeoutCallback(); }); 
    } 
} 
private void TimeoutCallback() 
{ 
    webRequest.Abort(); 
    MessageBox.Show("Request timed-out"); 
} 
private void ResponseCallback(WebResponse webResponse) 
{ 
    if( (webResponse as HttpWebResponse).StatusCode != HttpStatusCode.OK ) 
    { 
        MessageBox.Show("Download error1"); 
    } 
    else 
    { 
        responseStream = webResponse.GetResponseStream(); 
        if( responsePosition != 0 ) 
        { 
            responseStream.Position = responsePosition; 
        } 
        IAsyncResult readResult = responseStream.BeginRead(bufferRead, 0, BufferSize, new AsyncCallback(ReadCallback), null); 
        return; 
    } 
    webResponse.Close(); 
    MessageBox.Show("Download error2"); 
} 
private void ReadCallback(IAsyncResult asyncResult) 
{ 
    int bytes = responseStream.EndRead(asyncResult); 
    DLog.Info("store:{0}, current size:{1}", bytes, fileStream.Length); 
    if( bytes > 0 ) 
    { 
        fileStream.BeginWrite(bufferRead, 0, bytes, WriteCallback, null); 
        return; 
    } 
    responseStream.Close(); 
    MessageBox.Show("Download error3"); 
} 
private void WriteCallback(IAsyncResult asyncResult) 
{ 
    DLog.Info("Stored!"); 
    responseStream.BeginRead(bufferRead, 0, BufferSize, new AsyncCallback(ReadCallback), null); 
}

这是（略微编辑的）堆栈的片段：

(...)
MainPage.ReadCallback(System.IAsyncResult asyncResult) Line 190 + 0x1b bytes    C#
Stream.BeginRead(byte[] buffer, int offset, int count, System.AsyncCallback callback, object state) + 0x6c bytes    
MainPage.WriteCallback(System.IAsyncResult asyncResult) Line 200 + 0x1f bytes    C#
Stream.BeginWrite(byte[] buffer, int offset, int count, System.AsyncCallback callback, object state) + 0x64 bytes    
FileStream.BeginWrite(byte[] array, int offset, int numBytes, System.AsyncCallback userCallback, object stateObject) + 0x7c bytes    
IsolatedStorageFileStream.BeginWrite(byte[] buffer, int offset, int numBytes, System.AsyncCallback userCallback, object stateObject) + 0x1a bytes    
MainPage.ReadCallback(System.IAsyncResult asyncResult) Line 190 + 0x1b bytes    C#
Stream.BeginRead(byte[] buffer, int offset, int count, System.AsyncCallback callback, object state) + 0x6c bytes    
MainPage.WriteCallback(System.IAsyncResult asyncResult) Line 200 + 0x1f bytes    C#
Stream.BeginWrite(byte[] buffer, int offset, int count, System.AsyncCallback callback, object state) + 0x64 bytes    
FileStream.BeginWrite(byte[] array, int offset, int numBytes, System.AsyncCallback userCallback, object stateObject) + 0x7c bytes    
IsolatedStorageFileStream.BeginWrite(byte[] buffer, int offset, int numBytes, System.AsyncCallback userCallback, object stateObject) + 0x1a bytes    
MainPage.ReadCallback(System.IAsyncResult asyncResult) Line 190 + 0x1b bytes    C#
Stream.BeginRead(byte[] buffer, int offset, int count, System.AsyncCallback callback, object state) + 0x6c bytes    
MainPage.WriteCallback(System.IAsyncResult asyncResult) Line 200 + 0x1f bytes    C#
Stream.BeginWrite(byte[] buffer, int offset, int count, System.AsyncCallback callback, object state) + 0x64 bytes    
FileStream.BeginWrite(byte[] array, int offset, int numBytes, System.AsyncCallback userCallback, object stateObject) + 0x7c bytes    
IsolatedStorageFileStream.BeginWrite(byte[] buffer, int offset, int numBytes, System.AsyncCallback userCallback, object stateObject) + 0x1a bytes    
MainPage.ReadCallback(System.IAsyncResult asyncResult) Line 190 + 0x1b bytes    C#
Stream.BeginRead(byte[] buffer, int offset, int count, System.AsyncCallback callback, object state) + 0x6c bytes    
MainPage.WriteCallback(System.IAsyncResult asyncResult) Line 200 + 0x1f bytes    C#
Stream.BeginWrite(byte[] buffer, int offset, int count, System.AsyncCallback callback, object state) + 0x64 bytes    
FileStream.BeginWrite(byte[] array, int offset, int numBytes, System.AsyncCallback userCallback, object stateObject) + 0x7c bytes    
IsolatedStorageFileStream.BeginWrite(byte[] buffer, int offset, int numBytes, System.AsyncCallback userCallback, object stateObject) + 0x1a bytes    
(...)

此堆栈显示异步回调会立即调用，并且在可能完成所有迭代之前，执行不会返回到调用函数。我希望当一个函数异步调用另一个函数时，在调用函数完成并从堆栈中删除之前不会调用回调。但是，当前行为会导致堆栈在几百次迭代后溢出。

我正在尝试通过使用反应式扩展来解决此限制（或者这不应该真正称为错误吗？所以我试图用可观察的模式替换读取网络流的迭代，例如：

var readerFunc = Observable.FromAsyncPattern<byte[], int, int, int>(responseStream.BeginRead, responseStream.EndRead);

但这里的问题是，响应式库的Windows Phone版本已被剥离，仅支持两个参数和返回参数：

Observable.FromAsyncPattern<T1, T2, TResult>

所以我不能按照上面想要的方式定义读取函数，因为这需要三个参数。即使是库的可下载版本也没有提供更多参数。

最后，我的问题是：

1. 除了使用反应式扩展之外，还有其他方法可以解决异步调用并填满堆栈的原始问题吗？

2. 如果没有，那么如何使用 Windows Phone 上可用的有限版本的反应式扩展异步从网络流读取和写入文件流？

任何帮助非常感谢！

Answer 1

真的很简单...您需要检查IAsyncResult.CompletedSynchronously是否属实。如果是这样，请采取措施分解调用堆栈。也许你可以ThreadPool.QueueUserWorkItem或WP7等价物？

Answer 2

下面是一个示例。是的，Windows Phone 7确实有一个有限版本的Rx，只接受两个参数和一个结果（而在实际版本中，您可以使用29个重载和14个可能的参数）。