IObservable <IDisposable>으로 IDisposables의 누출을 중지하는 방법?

Question

내 프로덕션 코드에 버그가있어 추적하여 재현 할 테스트 사례를 만들 수있었습니다.

인스턴스를 만들고 동시에 구독 항목에 최대 하나의 항목을 유지하기 위해 연속 사용 일회용을 사용하고 있습니다. 장면에 그래픽 개체를 추가하고 업데이트가있을 때 제거 할 수있는 깔끔한 방법입니다.

그러나 다음 테스트 사례는 미묘한 버그를 보여줍니다.

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 방법. 는 IDisposable가 생성되고 가입에 전송되지 않습니다

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

     } 


    } 

} 

내가 구독을 처리

, 하나 더.

6 개의 disposables가 생성되지만 하나는 유출됩니다. 이것은 실제 시스템에서 스케줄링 지연을 모델링하기 위해 시스템에 넣은 지연 때문입니다.

내 질문은.

을 쓰는 것이 가능합니까? 위와 비슷한 IDisposables를 누설하지 않는 SubscribeDisposable입니다.

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> 

그것은 위험한 보인다 내 코드에서 제거됩니다.