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; 
     } 
    } 
} 

：それは明確に観察できるソース上OnCompletedを呼んだ

 
0 
1 
2 
3 
4 
5 
6 
7 
8 
9 
10 
Done. 

このコードを試してみてください。