我将如何使用响应式编程编写此代码？

Question

我刚开始使用响应式编程，我知道的只是编写代码，但还不足以弄清楚当我没有得到我期望的结果时会发生什么。除了博客文章，我真的没有可用的导师。我还没有找到一个很好的解决方案来解决我遇到的情况，我很好奇正确的方法。

问题：

我需要获取一个 Foo，它部分由 Bar 对象数组组成。我从 Web 服务中获取 Bar 对象。因此，我将每个 Web 服务调用表示为一个 IObservable，我希望在完成之前从其中包含 0 或 1 个元素。我想做一个 IObservable，它将：

• 订阅每个 IObservable 实例。
• 等待最多 2 秒超时。
• 当两个序列都完成或超时发生时：
  • 使用生成的任何 Bar 对象创建一个数组（可能有 0 个。）
  • 使用该 Bar[] 生成 Foo 对象。

我用这段代码完成了这个：

public Foo CreateFoo() {

        var producer1 = webService.BarGenerator()
                                  .Timeout(TimeSpan.FromSeconds(2), Observable.Empty<Bar>());
        var producer2 = // similar to above

        var pipe = producer1.Concat(producer2);

        Bar[] result = pipe.ToEnumerable().ToArray();

        ...
}

这似乎不对，原因有很多。最明显的是 Concat() 将串行而不是并行启动序列，所以这是一个 4 秒的超时。我真的不在乎它是否阻塞，它实际上对我正在使用的架构很方便。我很满意这种方法成为 IObservable 的生成器，但是这里有一些额外的注意事项，当我尝试时，这似乎会带来挑战：

1. 如果生产者 1 和生产者 2 的结果都产生结果，我需要最后一个数组来按顺序排列它们。
2. 我想使用 TestScheduler 来验证超时但还没有成功，我显然根本不懂调度程序。
3. 归根结底，这是一个拉动模型，无论 Foo 在一个不同的点需要它，并且“即时”接收它没有任何价值。也许这会倾斜“不要使用 Rx”的答案。老实说，我被卡住了，我切换到了基于任务的 API。但我想看看如何使用 Rx 来解决这个问题，因为我想学习。

Answer 1

    var pipe = producer1
        .Merge(producer2)
        .Buffer(Observable.Timer(TimeSpan.FromSeconds(2), testScheduler))
        .Take(1);

    var subscription = pipe
        .Select(list => new Foo(list.ToArray())
        .Subscribe(foo => {} /* Do whatever you want with your foo here.*/);

Buffer 获取窗口期间发出的所有元素（在我们的例子中为两秒），并输出一个列表。

如果您想坚持使用拉取模型，而不是订阅，您可以这样做：

var list = await pipe;
var foo = new Foo(list.ToArray());

//....