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

Question

我想生成一个文件的可观察对象，这样文件名的发现可以在任何时候取消。对于本例，取消将在1秒内自动完成。

下面是我当前的代码:

class Program
{
    static void Main()
    {
        try
        {
            RunAsync(@"\abcxyz").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。必须使用从外部上下文(GetFileSource参数)捕获的cts。我觉得很丑。

应该这样做吗?肯定有更好的办法。

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));

您还可以看到一个更高级的实现异步Generate方法的示例。

Answer 2

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

另外，当你在Observable.Create中执行return Disposable.Empty;时，你正在创建一个不能被正常的Rx订阅一次性阻止的可观察对象。这可能导致内存泄漏和不必要的处理。

最后，抛出异常来结束正常的计算是一个很糟糕的主意。

有一个很好的干净的解决方案似乎可以达到你的目的:

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);。你为什么要这么做?