IObservable <IDisposable>でIDisposablesの漏出を止めるには？

Question

生産コードにバグがあり、これを追跡して再現するテストケースを作成することができました。私は

IObservable<IDisposable> 

インスタンスを作成し、一度に生きている最大1つのアイテムを維持するために、サブスクリプションにシリアル使い捨てを使用しています。グラフィックオブジェクトをシーンに追加し、更新が利用可能になったときにそれらを削除するためのきれいな方法です。

ただし、次のテストケースでは微妙なバグが示されています。

using System; 
using System.Reactive.Concurrency; 
using System.Reactive.Disposables; 
using System.Reactive.Linq; 
using FluentAssertions; 
using Microsoft.Reactive.Testing; 
using Xunit; 

namespace WeinCadSW.Spec 
{ 
    /// <summary> 
    /// This test case demonstrates problems with streams of IDisposables. 
    /// http://stackoverflow.com/questions/37936537/how-to-stop-leaking-idisposables-with-an-iobservableidisposable 
    /// </summary> 
    public class ObservableDisposableSpec : ReactiveTest 
    { 
     TestScheduler _Scheduler = new TestScheduler(); 
     [Fact] 
     public void ShouldWork() 
     { 
      var o = _Scheduler.CreateHotObservable 
       (OnNext(100, "A") 
       , OnNext(200, "B") 
       , OnNext(250, "C") 
       , OnNext(255, "D") 
       , OnNext(258, "E") 
       , OnNext(600, "F") 
       ); 

      var disposablesCreated = 0; 
      var disposabledDisposed = 0; 
      var oo = o.Select 
       (s => 
       { 
        disposablesCreated++; 
        return Disposable.Create(() => disposabledDisposed++); 
       }) 
       .Delay(TimeSpan.FromTicks(10), _Scheduler); 


      IDisposable sub = Disposable.Empty; 
      _Scheduler.ScheduleAbsolute(null, 0, (Func<IScheduler, object, IDisposable>)((scheduler, state) => 
      { 
       sub = oo.SubscribeDisposable(); 
       return Disposable.Empty; 
      })); 
      _Scheduler.ScheduleAbsolute(null, 605, (Func<IScheduler, object, IDisposable>)((scheduler, state) => 
      { 
       sub.Dispose(); 
       return Disposable.Empty; 
      })); 

      _Scheduler.Start(); 

      // This test will fail here because 6 disposables are created. 
      disposablesCreated.Should().Be(6); 
      disposabledDisposed.Should().Be(6); // but is actually 5 

     } 

    } 

と、問題の核心であるSubscribeDisposable方法。

public static class Extensions 
    { 

     public static IDisposable SubscribeDisposable (this IObservable<IDisposable> o) 
     { 
      var d = new SerialDisposable(); 

      var s = o.Subscribe(v => 
      { 
       d.Disposable = v; 
      }); 

      return new CompositeDisposable(s, d); 

     } 


    } 

} 

サブスクリプションを廃棄すると、もう1つのIDisposableが生成され、サブスクリプションに送信されません。

6ディスポーザブルが生成されますが、1つはリークされます。これは、実際のシステムでスケジューリング遅延をモデル化するためにシステムに入れた遅延のためです。

私の質問はそうです。

サブスクリプションと同様のIDisposablesを漏らさないように書くことは可能ですか？

Answer 1

私はRXが強制する契約のために上記の問題に対する解決策がないと確信しています。上記のパターンを避けるためにコードを更新しました。代わりに私は使用します。それは危険なようだとして

/// <summary> 
    /// Subscribes to the observable sequence and manages the disposables 
    /// with a serial disposable. That 
    /// is before the function is called again the previous disposable is disposed. 
    /// </summary> 

    public static IDisposable SubscribeDisposable<T> 
    (this IObservable<T> o, Func<T, IDisposable> fn, Action<Exception> errHandler) 
    { 
     var d = new SerialDisposable(); 

     var s = o.Subscribe(v => 
     { 
       d.Disposable = Disposable.Empty; 
       d.Disposable = fn(v) ?? Disposable.Empty; 
     }, onError:errHandler); 

     return new CompositeDisposable(s,d); 

    } 

したがって

IObservable<IDisposable> 

は私のコードから削除されます。