Rx.Net - 订阅 Cold Observable 时，发布方法缺少前几项

Question

受Akavache 启发，我正在尝试创建一个为我提供IObservable<IArticle> 的解决方案。该方法基本上首先尝试获取数据库中存在的所有文章，然后尝试从 Web 服务获取更新的文章，并在从 Web 服务获取最新文章时尝试将它们保存回数据库。

由于 web 服务本质上是一个冷的 observable，我不想订阅两次，所以我使用Publish 连接到它。我的理解是我使用的是Publish 方法的正确版本，但是，很多时候该方法往往会错过GetNewsArticles 的前几篇文章。这是通过 UI 以及在下面的调用中添加的 Trace 调用观察到的。

除了解决问题之外，如果还了解如何调试/测试这段代码（除了引入 DI 注入NewsService），那就太好了。

public IObservable<IArticle> GetContents(string newsUrl, IScheduler scheduler)
{
    var newsService = new NewsService(new HttpClient());
    scheduler = scheduler ?? TaskPoolScheduler.Default;

    var fetchObject = newsService
        .GetNewsArticles(newsUrl,scheduler)
        .Do(x => Trace.WriteLine($"Parsing Articles {x.Title}"));

    return fetchObject.Publish(fetchSubject =>
    {
        var updateObs = fetchSubject
            .Do( x =>                         
            {
                // Save to database, all sync calls
            })
            .Where(x => false)
            .Catch(Observable.Empty<Article>());

        var dbArticleObs = Observable.Create<IArticle>(o =>
        {
            return scheduler.ScheduleAsync(async (ctrl, ct) =>
            {
                using (var session = dataBase.GetSession())
                {
                    var articles = await session.GetArticlesAsync(newsUrl, ct);
                    foreach (var article in articles)
                    {
                        o.OnNext(article);
                    }
                }
                o.OnCompleted();
            });
        });

        return
            dbArticleObs                // First get all the articles from dataBase cache
                .Concat(fetchSubject    // Get the latest articles from web service 
                    .Catch(Observable.Empty<Article>())
                    .Merge(updateObs))  // Update the database with latest articles
                .Do(x => Trace.WriteLine($"Displaying {x.Title}"));
    });
}

更新 - 添加了 GetArticles

public IObservable<IContent> GetArticles(string feedUrl, IScheduler scheduler)
{
    return Observable.Create<IContent>(o =>
    {
        scheduler = scheduler ?? DefaultScheduler.Instance;
        scheduler.ScheduleAsync(async (ctrl, ct) =>
        {
            try
            {
                using (var inputStream = await Client.GetStreamAsync(feedUrl))
                {
                    var settings = new XmlReaderSettings
                    {
                        IgnoreComments = true,
                        IgnoreProcessingInstructions = true,
                        IgnoreWhitespace = true,
                        Async = true
                    };

                    //var parsingState = ParsingState.Channel;
                    Article article = null;
                    Feed feed = null;

                    using (var reader = XmlReader.Create(inputStream, settings))
                    {
                        while (await reader.ReadAsync())
                        {
                            ct.ThrowIfCancellationRequested();
                            if (reader.IsStartElement())
                            {
                                switch (reader.LocalName)
                                {
                                    ...
                                    // parsing logic goes here
                                    ...
                                }
                            }
                            else if (reader.LocalName == "item" &&
                                     reader.NodeType == XmlNodeType.EndElement)
                            {
                                o.OnNext(article);
                            }
                        }
                    }

                    o.OnCompleted();
                }
            }
            catch (Exception e)
            {
                o.OnError(e);
            }

        });
        return Disposable.Empty;
    });
}

更新 2

在这里分享source code的链接。

Answer 1

我不喜欢您的代码的一些地方。我假设NewsService 是IDisposable，因为它需要HttpClient（这是一次性的）。您没有进行适当的清理。

此外，您还没有提供完整的方法 - 因为您已尝试将其缩减为问题 - 但这使得很难推断如何重写代码。

也就是说，在我看来非常可怕的一件事是Observable.Create。你能试试这个代码，看看它是否对你有用吗？

    var dbArticleObs =
        Observable
            .Using(
                () => dataBase.GetSession(),
                session =>
                    from articles in Observable.FromAsync(ct => session.GetArticlesAsync(newsUrl, ct))
                    from article in articles
                    select article);

现在，如果是这样，请尝试重写 fetchObject 以在新建 `NewService 时使用相同的 Observable.Using。

无论如何，如果您能在问题中提供GetContents、NewsService 和dataBase 代码的完整实现，那就太好了。