使用 Observable SelectMany 处理反应式扩展错误答案

【问题标题】：Reactive Extensions error handling with Observable SelectMany使用 Observable SelectMany 处理反应式扩展错误
【发布时间】：2019-12-27 11:06:57
【问题描述】：

我正在尝试使用响应式扩展库在某个文件夹上编写文件观察程序这个想法是监视硬盘驱动器文件夹中的新文件，等到文件完全写入并将事件推送给订阅者。我不想使用FileSystemWatcher，因为它会为同一个文件引发两次 Changed 事件。

所以我用“反应式”（我希望）这样写了它：

var provider = new MessageProviderFake();
var source = Observable.Interval(TimeSpan.FromSeconds(2), NewThreadScheduler.Default).SelectMany(_ => provider.GetFiles());
using (source.Subscribe(_ => Console.WriteLine(_.Name), () => Console.WriteLine("completed to Console")))
{
    Console.WriteLine("press Enter to stop");
    Console.ReadLine();
}

但是我找不到“反应式”来处理错误。例如，文件目录可能位于外部驱动器上，但由于连接问题而变得不可用。所以我添加了GetFilesSafe，它将处理来自响应式扩展的异常错误：

static IEnumerable<MessageArg> GetFilesSafe(IMessageProvider provider)
{
    try
    {
        return provider.GetFiles();
    }
    catch (Exception e)
    {
        Console.WriteLine(e.Message);
        return new MessageArg[0];
    }
}

并像使用它

var source = Observable.Interval(TimeSpan.FromSeconds(2), NewThreadScheduler.Default).SelectMany(_ => GetFilesSafe(provider));

有没有更好的方法让SelectMany 调用provider.GetFiles()，即使引发了异常？在这种情况下，我使用错误计数器重复读取操作 N 次，然后失败（终止进程）。

Reactive Extensions 中是否有“尝试 N 次并在尝试之间等待 Q 秒”？

GetFilesSafe 也有一个问题：它返回 IEnumerable<MessageArg> 用于懒惰阅读，但是它可以在迭代时引发，异常将在 SelectMany 的某处抛出

【问题讨论】：

GetFiles 有什么作用？枚举文件夹的所有文件？
@TheodorZoulias 是的，它返回名称属性中具有文件名的对象列表。
这听起来效率很低，尤其是在文件夹包含大量文件的情况下。在修正了它的缺点之后，我更喜欢使用FileSystemWatcher。
@TheodorZoulias 实际上的想法是扫描新文件，处理它们并移动到存档。它不应该一次又一次地扫描同一个庞大的文件集

标签： c# observable system.reactive

【解决方案1】：

有一个Retry 扩展，如果当前一个错误，它只是再次订阅可观察对象，但听起来这不会提供你想要的灵活性。

你可以使用Catch 构建一些东西，如果外部发生错误，它会订阅你给它的 observable。类似于以下内容（未经测试）：

IObservable<Thing> GetFilesObs(int times, bool delay) {
    return Observable
        .Return(0)
        .Delay(TimeSpan.FromSeconds(delay ? <delay_time> : 0))
        .SelectMany(_ => Observable.Defer(() => GetFilesErroringObservable()))
        .Catch(Observable.Defer(() => GetFilesObs(times - 1, true)));
}

// call with:
GetFilesObs(<number_of_tries>, false);

正如所写，除了触发重试之外，这对错误没有任何作用。特别是，当发生足够多的错误时，它会在没有错误的情况下完成，这可能不是您想要的。

【讨论】：