Rx可观测-链接

Question

我有一个从web下载csv文件的函数。我需要扩展它，以便它一个接一个地对2个网站进行2次调用。但是我不知道该怎么做…

功能如下：

// Define other methods and classes here
private void GetCSVData(string url1, string url2)
{
    WebClient webClient = null;
    try
    {
        webClient = new WebClient();
        var task = Observable.FromEventPattern
            <OpenReadCompletedEventHandler, OpenReadCompletedEventArgs>
        (
            ev => webClient.OpenReadCompleted += ev,
            ev => webClient.OpenReadCompleted -= ev
        );
        // needs to be redone
        task.Subscribe(t => ParseCSV1(t.EventArgs.Result));
        // call ParseCSV1()
        // then call ParseCSV2()
        // needs redone, 2 calls to 2 website
        webClient.OpenReadAsync(new Uri(url1));     
    }
    catch (WebException wex)
    {
        System.Diagnostics.Debug.WriteLine(wex.ToString());
    }
    catch (Exception ex)
    {
        System.Diagnostics.Debug.WriteLine(ex.ToString());
    }
}
private void ParseCSV1(Stream stream)
{
    // Parse steps...
}
private void ParseCSV2(Stream stream)
{
    // Parse steps...
}

Answer 1

您所拥有的应该是可以的。如果您再次调用OpenReadAsync，则会触发第二个事件（每个URL一个），您的订阅将同时处理这两个事件。

您应该能够简单地执行以下操作：

// Call both, will generate two events.
webClient.OpenReadAsync(new Uri(url1));
webClient.OpenReadAsync(new Uri(url2));

然而，我怀疑您在使用WebClient时遇到了一些终身问题（它从未得到正确处理）。

Answer 2

我。。。个人可能不会在这里使用Rx。实际上，这听起来更适合任务库；这是我可能会做的事情：

void Main()
{
    var webClient = new System.Net.WebClient();
    Task.Factory
        .StartNew(() => GetUrl("http://www.google.com"))
        .ContinueWith(tsk => ParseCsv(CheckAndHandleError(tsk)))
        .ContinueWith(tsk => 
            { 
                if(CheckAndHandleError(tsk)) 
                    return GetUrl("http://www.yahoo.com"); 
                else
                    return string.Empty;
            })
        .ContinueWith(tsk => ParseCsv(CheckAndHandleError(tsk)));
}
private T CheckAndHandleError<T>(Task<T> task)
{
    if(task.IsFaulted)
    {
        // you'll need to handle the tsk.Exception here
        throw new NotImplementedException();
    }
    else
    {
        return task.Result;
    }   
}
private string GetUrl(string address)
{
    using(var client = new WebClient())
    {
        var stream = client.OpenRead(address);
        using(var rdr = new StreamReader(stream))
        {
            return rdr.ReadToEnd();
        }
    }
}
private bool ParseCsv(string csvText)
{
    // whatever this is
    Console.WriteLine(csvText);
    return true;
}