4

我正在尝试拼凑一个工作原理如下的 Rx 管道：

  1. 我编写了一个函数，它接受一个 IObservable，为我提供包含有关公司信息的配置文件
  2. 我查询各种数据源以查找可能相关的公司资料，所有这些都是并行的。我将其合并到一个 IObservable 的公司简介中。
  3. 当我取回这些可能相关的配置文件时，我会将它们与我已经观察到的配置文件进行比较，如果它们的相关性 > 80% 并且与我已经观察到的任何配置文件都不相同，我认为它们是匹配的。
  4. 我想将匹配的公司返回到第 1 步，以便我可以搜索这些新匹配配置文件的相关数据。

我使用一些已知的良好配置文件引导该过程。

最终，没有更多尚未看到的匹配配置文件，因此该过程结束。

我在编程时遇到了麻烦。如果我使用 Subject 允许管道的尾端将其配置文件发送到工作流的开头，那么没有人会调用 OnCompleted 并且我永远不会发现该过程已经结束。如果我用递归来开发它，我似乎总是以堆栈溢出结束，因为我试图用自己的返回值调用一个函数。

谁能帮助我以我可以确定该过程已结束的方式完成此任务？

4

1 回答 1

8

听起来您想要这样的数据流：

seed profiles --> source --> get related --> output
                     ^                    |
                     |                    v
                     -<--- transform <-----

这似乎是解决一般问题比解决特定问题更容易或更容易的情况，因此我将提出一个通用的“反馈”函数，它应该为您提供所需的构建块：

编辑：固定功能完成

IObservable<TResult> Feedback<T, TResult>(this IObservable<T> seed,
                                          Func<T, IObservable<TResult>> produce,
                                          Func<TResult, IObservable<T>> feed)
    {
        return Observable.Create<TResult>(
                obs =>
                {
                    var ret = new CompositeDisposable();
                    Action<IDisposable> partComplete = 
                        d =>
                        {
                            ret.Remove(d);
                            if (ret.Count == 0) obs.OnCompleted();
                        };
                    Action<IObservable<T>, Action<T>> ssub =
                        (o, n) =>
                        {
                            var disp = new SingleAssignmentDisposable();
                            ret.Add(disp);
                            disp.Disposable = o.Subscribe(n, obs.OnError, () => partComplete(disp));
                        };
                    Action<IObservable<TResult>, Action<TResult>> rsub =
                        (o, n) =>
                        {
                            var disp = new SingleAssignmentDisposable();
                            ret.Add(disp);
                            disp.Disposable = o.Subscribe(n, obs.OnError, () => partComplete(disp));
                        };

                    Action<T> recurse = null;
                    recurse = s =>
                              {
                                  rsub(produce(s),
                                       r => 
                                       {
                                           obs.OnNext(r);
                                           ssub(feed(r), recurse);
                                       });
                              };

                    ssub(seed, recurse);
                    return ret;
                });
    }

在您的情况下，T并且TResult看起来是相同feed的，身份功能也是如此。 produce将是用于实现步骤 2 和 3 的函数。

我测试了这个函数的一些示例代码：

void Main()
{
    var seed = new int[] { 1, 2, 3, 4, 5, 6 };
    var found = new HashSet<int>();
    var mults = seed.ToObservable()
                    .Feedback(i =>
                              {
                                  return Observable.Range(0, 5)
                                         .Select(r => r * i)
                                         .TakeWhile(v => v < 100)
                                         .Where(v => found.Add(v));
                              },
                              i => Observable.Return(i));

    using (var disp = mults.Dump())
    {
        Console.WriteLine("Press any key to stop");
        Console.ReadKey();
    }
    Console.WriteLine("Press any key to exit");
    Console.ReadKey();
}

static IDisposable Dump<T>(this IObservable<T> source)
{
    return source.Subscribe(item => Console.WriteLine(item),
                            ex => Console.WriteLine("Error occurred in dump observable: " + ex.ToString()),
                            () => Console.WriteLine("Dump completed"));
}
于 2013-03-09T04:21:32.643 回答