c# - HttpWebRequest 异步调用填充堆栈并剥离 Windows Phone 上的响应式(Reactive)扩展

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    
(...)

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

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

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

但这里的问题是，Windows Phone 版本的响应式(Reactive)库已被精简为仅支持两个参数和返回参数:

Observable.FromAsyncPattern<T1, T2, TResult>

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

最后，我的问题是:

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

2. 如果没有，那么如何使用 Windows Phone 上提供的有限版本的响应式扩展从网络流中异步读取数据并写入文件流？

非常感谢任何帮助!

Answer 1

真的非常简单...您需要检查IAsyncResult.CompletedSynchronously是否为true。如果是这样，请采取措施分解调用堆栈。也许您可以ThreadPool.QueueUserWorkItem或WP7等效项？