Rx.NET 生成可取消的可观察文件名的方法是什么？

Question

我想生成一个可观察的文件，以便可以随时取消文件名的发现。在本示例中，取消会在 1 秒内自动发生。

这是我当前的代码：

class Program
{
    static void Main()
    {
        try
        {
            RunAsync(@"\\abc\xyz").GetAwaiter().GetResult();
        }
        catch (Exception exc)
        {
            Console.Error.WriteLine(exc);
        }
        Console.Write("Press Enter to exit");
        Console.ReadLine();
    }

    private static async Task RunAsync(string path)
    {
        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(1));
        await GetFileSource(path, cts);
    }

    private static IObservable<string> GetFileSource(string path, CancellationTokenSource cts)
    {
        return Observable.Create<string>(obs => Task.Run(async () =>
        {
            Console.WriteLine("Inside Before");
            foreach (var file in Directory.EnumerateFiles(path, "*", SearchOption.AllDirectories).Take(50))
            {
                cts.Token.ThrowIfCancellationRequested();
                obs.OnNext(file);
                await Task.Delay(100);
            }
            Console.WriteLine("Inside After");
            obs.OnCompleted();
            return Disposable.Empty;
        }, cts.Token))
        .Do(Console.WriteLine);
    }
}

我不喜欢我的实现的两个方面（如果有更多 - 请随时指出）：

1. 我有一个可枚举的文件，但我手动迭代每个文件。我可以以某种方式使用ToObservable 扩展吗？
2. 我不知道如何使用传递给Task.Run 的cts.Token。必须使用从外部上下文捕获的cts（GetFileSource 参数）。我觉得很难看。

这是应该怎么做的吗？一定是更好的方法。

Answer 1

我仍然不相信这真的是一个反应性问题，你要求对生产者施加背压，这确实违背了反应性应该如何工作。

话虽如此，如果你打算这样做，你应该意识到非常细粒度的时间操作应该几乎总是委派给Scheduler，而不是尝试与Tasks和CancellationTokens进行协调.所以我会重构成这样：

public static IObservable<string> GetFileSource(string path, Func<string, Task<string>> processor, IScheduler scheduler = null) {

  scheduler = scheduler ?? Scheduler.Default;

  return Observable.Create<string>(obs => 
  {
    //Grab the enumerator as our iteration state.
    var enumerator = Directory.EnumerateFiles(path, "*", SearchOption.AllDirectories)
                              .GetEnumerator();
    return scheduler.Schedule(enumerator, async (e, recurse) =>
    {
      if (!e.MoveNext())
      {
         obs.OnCompleted();
         return;
      }

      //Wait here until processing is done before moving on
      obs.OnNext(await processor(e.Current));

      //Recursively schedule
      recurse(e);
    });
  });

}

然后，不要传入取消令牌，而是使用TakeUntil：

var source = GetFileSource(path, x => {/*Do some async task here*/; return x; })
 .TakeUntil(Observable.Timer(TimeSpan.FromSeconds(1));

您还可以查看更高级的示例，以实现 async Generate method。

Answer 2

当您可以使用其他运算符时，我建议您避免使用Observable.Create。

此外，当您在 Observable.Create 中执行 return Disposable.Empty; 时，您正在创建一个无法被普通 Rx 订阅一次性停止的 observable。这可能会导致内存泄漏和不必要的处理。

最后，抛出异常来结束正常计算是个坏主意。

有一个很好的干净解决方案似乎可以满足您的需求：

private static IObservable<string> GetFileSource(string path, CancellationTokenSource cts)
{
    return
        Directory
            .EnumerateFiles(path, "*", SearchOption.AllDirectories)
            .ToObservable()
            .Take(50)
            .TakeWhile(f => !cts.IsCancellationRequested);
}

我唯一没有包括的是Task.Delay(100);。你为什么要这么做？