在反应扩展中合并可观测项

Question

刚刚学习RX，想制作一个迭代文件系统的程序。以下是我想出的有效方法：

using System;
using System.IO;
using System.Reactive.Disposables;
using System.Reactive.Linq;
namespace ConsoleApplication8
{
    internal class Program
    {
        private static IObservable<string> GetFiles(string folder, string filePattern)
        {
            return Observable.Create<string>(
                o =>
                {
                    var files = Directory.GetFiles(folder, filePattern);
                    foreach (var file in files)
                    {
                        o.OnNext(file);
                    }
                    var folders = Directory.GetDirectories(folder);
                    foreach (var f in folders)
                    {
                        var x = GetFiles(f, filePattern);
                        x.Subscribe(p => { o.OnNext(p); });
                    }
                    o.OnCompleted();
                    return Disposable.Empty;
                });
        }
        private static void Main(string[] args)
        {
            var o = GetFiles(@"d:temp", "*.*");
            o.Subscribe(p => { Console.WriteLine(p); });
            Console.Read();
        }
    }
}

（注意递归的使用，再次调用GetFiles并订阅）

虽然它很有效，但它看起来很笨拙，我忍不住想，我应该使用像Concat这样的东西来组合序列，而不是只是将它们冒泡起来。

此外，我想将Foreach更改为Parallel.Foreach，但我不确定使用RX会产生什么后果。我似乎找不到太多文件。

关于如何使用RX更好地写这篇文章，有什么建议吗？

Answer 1

要解决这样的问题，可以先编写函数的LINQ版本。例如：

    static IEnumerable<string> GetFiles(string folder, string filePattern)
    {
        return Directory.GetFiles(folder, filePattern)
            .Concat(Directory.GetDirectories(folder).SelectMany(f => GetFilesEnumerable(f, filePattern)));
    }

然后只需将IEnumerables更改为IOobservables:

    static IObservable<string> GetFiles(string folder, string filePattern)
    {
        return Directory.GetFiles(folder, filePattern).ToObservable()
            .Concat(Directory.GetDirectories(folder).ToObservable().SelectMany(f => GetFilesEnumerable(f, filePattern)));
    }

Answer 2

事实证明，根据我对SearchOption.AllDirectories的发现（为什么我这么多年来从未注意到这一点），它可以归结为两行真正的代码：

using System;
using System.IO;
using System.Reactive.Linq;
namespace ConsoleApplication8
{
    internal class Program
    {
        private static void Main(string[] args)
        {
            var o = Directory.GetFiles(@"e:code", "*.*", SearchOption.AllDirectories).ToObservable();
            o.Subscribe(f => Console.WriteLine(f));

            Console.Read();
        }
    }
}

现在真是太简单了。需要一个新的RX问题来处理。