Observable 和 Webclient 获取 csv

Question

我的 lightswitch 应用程序中有一个函数，可以从网站下载 csv 文件，我想使用 Rx 框架重写该文件，并提供同步调用它的可能性。

下面提供了旧函数和新函数的代码片段。然而，新函数不起作用，对 ParseCSV 的调用永远不会发生。我想知道为什么，如果有更好的解决方案，请随时提供。

旧代码：

private void ObservableCollection<Data> collection;
public ObservableCollection<Data> GetData(string url, ObservableCollection<Data> targetCollection)
{

collection = targetCollection;
if (!string.IsNullOrEmpty(url))
{
    WebClient wc = new WebClient();
    wc.OpenReadCompleted += new OpenReadCompletedEventHandler(OpenReadCompleted_ParseCSV);
    wc.OpenReadAsync(new Uri(url));
}
return collection;
}

private void OpenReadCompleted_ParseCSV(object sender, OpenReadCompletedEventArgs e)
{

if (e.Error != null) return;

var webClient = sender as WebClient;
if (webClient == null) return;

try
{
    using (StreamReader reader = new StreamReader(e.Result))
    {
        string contents = reader.ReadToEnd();
        ...
    }
}
catch (Exception ex)
{
    System.Diagnostics.Debug.WriteLine("Error parsing CSV!\n" + ex.Message);
}

}

新代码（带 Rx）：

private void ObservableCollection<Data> collection;
public ObservableCollection<Data> GetData(string url, ObservableCollection<Data> targetCollection)
{
collection = targetCollection;
if (!string.IsNullOrEmpty(url))
{
    var result = Observable.FromEventPattern<OpenReadCompletedEventHandler, OpenReadCompletedEventArgs>
                 (
                    ev => webClient.OpenReadCompleted += ev,
                    ev => webClient.OpenReadCompleted -= ev
                 )
                 .Select(o => o.EventArgs.Result)
                 .FirstOrDefault()
                 .ParseCSV();

    // Call the Async method
    webClient.OpenReadAsync(new Uri(url));
}
return collection;
}

private void ParseCSV(this Stream stream)
{
try
{
    using (StreamReader reader = new StreamReader(e.Result))
    {
        string contents = reader.ReadToEnd();
        ...
    }
}
catch (Exception ex)
{
    System.Diagnostics.Debug.WriteLine("Unable to get history data!\n" + ex.Message);
}
}

Answer 1

很难知道你想要什么（我认为你试图简化在 StackOverflow 上发布的代码，并且在翻译中丢失了很多内容），但我认为对此我没有什么可以改变的。

我注意到的一件事是您正在主线程上解析结果。您可能会这样做是有原因的，但您可能会考虑这一点：

//Note the void here. Is your intention to return a new collection or contribute
//to an existing one? I assumed the latter and changed the method to be more clear
//that this method causes side effects.
public void GetData(string url, ObservableCollection<Data> targetCollection)
{
        var result = Observable
            .FromEventPattern<OpenReadCompletedEventHandler, OpenReadCompletedEventArgs>
            (
                ev => webClient.OpenReadCompleted += ev,
                ev => webClient.OpenReadCompleted -= ev
            )
            .Select(o => ParseCSV(o.EventArgs.Result));

    result.Subscribe(targetCollection.Add);
    webClient.OpenReadAsync(new Uri(url));
}

//This method now returns a Data object read from a Stream
private static Data ParseCSV(Stream stream)
{
    try
    {
        using (StreamReader reader = new StreamReader(stream))
        {
            string contents = reader.ReadToEnd();
            //...
            return data;
        }
    }
    catch (Exception ex)
    {
        //Use Exception.ToString(). You get error and stack trace information
        //For this error as well as any inner exceptions. Nice!
        System.Diagnostics.Debug.WriteLine("Unable to get history data!\n" + ex.ToString());
    }
    return null;
}

在这里，对于从 webClient 请求返回的每个值（只有一个），我们将结果投影到您的 Data 类，而不是在可观察流之外进行转换。

我对你的方法做了一些小的修改。我不太喜欢这样带有副作用的代码（传递集合来贡献似乎是错误的向量），但我会允许它。除此之外，我认为这应该效果很好。

Answer 2

WPF 的改进答案

创建一个新项目并将此代码粘贴到您的主窗口中。添加一个名为 XBStart 的按钮，如果将单击处理程序连接到 XBStart_Click，则一切准备就绪。运行项目并单击按钮！

public partial class MainWindow : Window
{
    public MainWindow()
    {
        InitializeComponent();
        RX.DataReady += new RX.OnData(RX_DataReady);
    }

    private void RX_DataReady(ObservableCollection<string> Data)
    {
        Debugger.Break();
    }

    private void XBStart_Click(object sender, RoutedEventArgs e)
    {
        RX.GetData("http://www.yahoo.com");
    }
}

public static class RX
{
    public delegate void OnData(ObservableCollection<string> Data);

    public static event OnData DataReady;

    private static WebClient webClient;

    private static ObservableCollection<string> TheData { get; set; }

    private static void Notify()
    {
        if (DataReady != null)
        {
            DataReady(TheData);
        }
    }

    public static void GetData(string url)
    {
        webClient = new WebClient();
        TheData = new ObservableCollection<string>();
        var result = Observable
            .FromEventPattern<OpenReadCompletedEventHandler, 
                              OpenReadCompletedEventArgs>
            (
                ev => webClient.OpenReadCompleted += ev,
                ev => webClient.OpenReadCompleted -= ev
            )
            .Select(o => Parse.CSV(o.EventArgs.Result));

        result.Subscribe<string>(p =>
        {
            TheData.Add(p);
            Notify();
        });
        webClient.OpenReadAsync(new Uri(url));
    }
}

public static class Parse
{
    //This method now returns a Data object read from a Stream
    public static string CSV(Stream stream)
    {
        try
        {
            using (StreamReader reader = new StreamReader(stream))
            {
                string contents = reader.ReadToEnd();
                //...
                return contents;
            }
        }
        catch (Exception ex)
        {
            //Use Exception.ToString(). 
            //You get error and stack trace information
            //For this error as well as any inner exceptions. Nice!
            System.Diagnostics.Debug.WriteLine("Unable to get history data!\n" +
                                               ex.ToString());
        }
        return null;
    }
}