Reactive Extensions ToEnumerable 如果不迭代所有内容，如何处理可观察状态

Question

如果我在IObservable 上使用ToEnumerable 扩展，则用户可能不会迭代所有元素。在那种情况下，Observable.Create 中声明的IDisposable 该如何正确处理？

为了争论，假设不能直接将IObservable返回给用户（在这种情况下，用户可以自己实现取消）。

 private IObservable<Object> MakeObservable()
 {
   return Observable.Create(async (observer, cancelToken) =>
   {
     using(SomeDisposable somedisposable = new SomeDisposable())
     {
        while(true)
        {
          Object result = somedisposable.GetNextObject();
          if(result == null)
          {
            break;
          }
          observer.OnNext(result);
        }       
     }
   }
 }

 public IEnumerable<Object> GetObjects()
 {
   return MakeObservable().ToEnumerable();
 }

 public void Test()
 {
   IEnumerable<Object> e = GetObjects();
   int i = 0;
   foreach(Object o in e)
   {
     if(i++ == 10)
        break;
   }
   //somedisposable is not disposed here!!!
 }

Answer 1

问题是您对Create 的定义没有返回，因此无法停止。如果您将可观察的源更改为基于计时器的东西，那么它可以正常工作。试试这个代码：

public IEnumerable<long> GetObjects()
{
    return Observable
        .Interval(TimeSpan.FromSeconds(1.0))
        .Finally(() => Console.WriteLine("Done."))
        .ToEnumerable();
}

public void Test()
{
    foreach (long i in GetObjects())
    {
        Console.WriteLine(i);
        if (i == 10)
        {
            break;
        }
    }
}

当你运行它时，你会得到这个输出：

0 1 2 3 4 5 6 7 8 9 10 完毕。

它显然在源 observable 上调用 OnCompleted。