使用无限数量集压缩 Rx IObservable答案

【问题标题】：Zipping Rx IObservable with infinite number set使用无限数量集压缩 Rx IObservable
【发布时间】：2010-07-30 01:59:22
【问题描述】：

我有一个来自 Reactive 扩展框架的 IObservable [在下面的示例中命名的行]，我想为它观察到的每个对象添加索引号。

我已经尝试使用 Zip 函数来实现这一点：

rows.Zip(Enumerable.Range(1, int.MaxValue), (row, index) => 
    new { Row = row, Index = index })
    .Subscribe(a => ProcessRow(a.Row, a.Index), () => Completed());

.. 但不幸的是，这会抛出

ArgumentOutOfRangeException: 指定的参数超出了有效值的范围。参数名称：一次性用品

是我对 Zip 函数的理解有误还是我的代码有问题？

代码的 Range 部分似乎没有问题，并且 IObservable 尚未接收任何事件。

【问题讨论】：

我没有得到这个异常......行的类型是什么，IEnumerable、List、IObservable？我发布了我的代码，因为评论无法接受......我的代码中缺少什么导致你的抛出？

标签： linq operating-system c#-4.0 system.reactive

【解决方案1】：

.Select 有一个包含索引的重载：

rows.Select((row, index) => new { row, index });

【讨论】：

【解决方案2】：

显然，Zip 扩展方法将原始自定义 IObservable 转换为匿名 observable，并且订阅它会创建一个 System.Collections.Generic.AnonymousObserver，它不实现 IDisposable。因此，您无法以正常方式实现 Subscribe 方法（至少我见过它使用的方式），即

public IDisposable Subscribe(IObserver<T> observer) {
  // ..add to observer list..
  return observer as IDisposable
}

更可能的正确答案是：

return Disposable.Create(() => Observers.Remove(observer));

您应该注意，collction 可能会在 Completed-method 期间被修改，因此在处理它们之前创建列表的副本：

public void Completed()
{
    foreach (var observer in Observers.ToList())
    {
        observer.OnCompleted();
    }
 }

【讨论】：

【解决方案3】：

我不确定您的问题是什么，这对您有用吗（以及您在做什么这里缺少什么？）：

    static void Main(string[] args)
    {
        var rows = new List<int> { 4,5,1,2,5 }.ToObservable();
        rows.Zip(Enumerable.Range(1, int.MaxValue), (row, index) =>
            new { Row = row, Index = index })
            .Subscribe(a => ProcessRow(a.Row, a.Index), () => Completed());

        Console.ReadLine();
    }
    static void ProcessRow(int row, int index) {
        Console.WriteLine("Row {0}, Index {1}", row, index);
    }
    static void Completed() {
    }

【讨论】：

问题在于我的 IDisposable 取消订阅者创建。我愚蠢地从某个地方复制/粘贴了一个样本，这对我的情况来说是错误的。如上所述，我应该使用 Disposable.Create 函数，但我使用它的方式返回了一个空 IDisposable，这就是它抛出一个奇怪的异常的原因。