RX：サンプルの\ Interval \ Switchパイプラインをテストします

Question

イベントストリームの非アクティブをテストするRXコードがありますが、ストリームのアクティビティは非アクティブトリガの間隔をリセットします。

本当の問題 - （ObserveInactivityは、ちょうど最後のアクティビティのタイムスタンプに基づいてタイムアウトとリセット前のサンプルなければならない、私は間隔が毎秒再作成することはキル上だと思いますが）必要に応じて

public interface IReportActivity 
{ 
    event EventHandler Activity; 
} 

public interface IInactivityMonitor 
{ 
    IObservable<Unit> ObserveInactivity(TimeSpan inactivityTimeout); 
} 

public class InactivityMonitor : IInactivityMonitor 
{ 
    private readonly ISchedulerProvider _schedulerProvider; 
    private readonly IReportActivity _activitySource; 
    private IObservable<Unit> _inactivityObservable; 

    public InactivityMonitor(IRaiseActivity activitySource, ISchedulerProvider schedulerProvider) 
    { 
     _activitySource = activitySource; 
     _schedulerProvider = schedulerProvider; 
    } 

    public IObservable<Unit> ObserveInactivity(TimeSpan inactivityTimeout) 
    { 
     return GetInactivityObservable() 
      .Select(_ => Observable.Interval(inactivityTimeout, _schedulerProvider.NewThread) 
        .Timestamp() 
        .Select(__ => Unit.Default)) 
      .Switch(); 
    } 

    public IObservable<Unit> GetInactivityObservable() 
    { 
     return _inactivityObservable = _inactivityObservable ?? 
      Observable.FromEventPattern<EventHandler<EventArgs>, EventArgs>(
       h => _activitySource.Activity += h, 
       h => _activitySource.Activity -= h) 
       .Sample(TimeSpan.FromSeconds(1), _schedulerProvider.NewThread) 
       .Select(_ => Unit.Default) 
       .Publish() 
       .RefCount(); 
    } 
} 

コードは動作します私はこのコードをテストしようとしています。

[TestFixture] 
public class InactivityMonitorTests 
{ 
    private TestSchedulers _testSchedulers; 
    private InactivityMonitor _sut; 
    private AutoMock _moqqer; 

    protected override void Setup() 
    { 
     base.Setup(); 

     _moqqer = new AutoMock() 
     _testSchedulers = new TestSchedulers(); 
     _moqqer.Use<ISchedulerProvider>(_testSchedulers); 
     _sut = Moqqer.CreateInstance<InactivityMonitor>(); 
    } 

    // this test passes 
    [Test] 
    public void GetInactivityObservable_ActivityDetected_ReportsActivity() 
    { 
     var activityObserved = false; 

     _sut.GetInactivityObservable() 
      .Subscribe(x => activityObserved = true); 

     RaiseActivityEvent(); 

     _testSchedulers.NewThread.AdvanceBy(TimeSpan.FromSeconds(11).Ticks); 

     activityObserved.Should().BeTrue(); 
    } 

    private void RaiseActivityEvent() 
    { 
     _moqqer.GetMock<IReportActivty>() 
      .Raise(m => m.Activity += null, EventArgs.Empty); 
    } 

    // this test fails, The interval never appears to get set up via the tests. 
    [Test] 
    public void ObserveActivity_InactivtyTimeoutExceeded_NotificationReceived() 
    { 
     var inactivityObserved = false; 

     _sut.ObserveInactivity(TimeSpan.FromSeconds(10)) 
      .Subscribe(x => inactivityObserved = true); 

     _testSchedulers.NewThread.AdvanceBy(TimeSpan.FromSeconds(11).Ticks); 

     inactivityObserved.Should().BeTrue(); 
    } 
} 

public interface ISchedulerProvider 
{ 
    IScheduler CurrentThread { get; } 
    IScheduler Dispatcher { get; } 
    IScheduler Immediate { get; } 
    IScheduler NewThread { get; } 
    IScheduler ThreadPool { get; } 

    IScheduler TaskPool { get; } 
} 

public sealed class TestSchedulers : ISchedulerProvider 
{ 
    private readonly TestScheduler _currentThread = new TestScheduler(); 
    private readonly TestScheduler _dispatcher = new TestScheduler(); 
    private readonly TestScheduler _immediate = new TestScheduler(); 
    private readonly TestScheduler _newThread = new TestScheduler(); 
    private readonly TestScheduler _threadPool = new TestScheduler(); 
    private readonly TestScheduler _taskPool = new TestScheduler(); 
    #region Explicit implementation of ISchedulerService 
    IScheduler ISchedulerProvider.CurrentThread { get { return _currentThread; } } 
    IScheduler ISchedulerProvider.Dispatcher { get { return _dispatcher; } } 
    IScheduler ISchedulerProvider.Immediate { get { return _immediate; } } 
    IScheduler ISchedulerProvider.NewThread { get { return _newThread; } } 
    IScheduler ISchedulerProvider.ThreadPool { get { return _threadPool; } } 
    IScheduler ISchedulerProvider.TaskPool { get { return _taskPool; } } 

    #endregion 
    public TestScheduler CurrentThread { get { return _currentThread; } } 
    public TestScheduler Dispatcher { get { return _dispatcher; } } 
    public TestScheduler Immediate { get { return _immediate; } } 
    public TestScheduler NewThread { get { return _newThread; } } 
    public TestScheduler ThreadPool { get { return _threadPool; } } 
    public TestScheduler TaskPool { get { return _taskPool; } } 
} 

私が加入していないとした後、別のスケジューラ上で購読しようとする前に、サブスクライブすることを開始するが、何の成功の前に、スケジューラを起動するためのさまざまな方法を試してみました。

EDIT：デバッグでは、ObserveInactivityメソッドで間隔が決して生成されないことが示されます。

誰かが私を正しい方向に向けることができます。

おかげ

Answer 1

私は_activitySource.Activityからトリガされるの代わりにので、あなたの非アクティブクエリを再編成するでしょうし、そのために時間を追加して、あなたの代わりにXXX期間中に故障/沈黙を期待し、そのためのnotifcationを上げます。アクティビティがあった場合、そのクエリをキャンセル/再起動するようにアップグレードします。

すなわち代わりRxはすでに遅延評価されるよう周りに

public IObservable<Unit> GetInactivityObservable() 
{ 
    return _inactivityObservable = _inactivityObservable ?? 
     Observable.FromEventPattern<EventHandler<EventArgs>, EventArgs>(
      h => _activitySource.Activity += h, 
      h => _activitySource.Activity -= h) 
      .Select(_=>Unit.Default) 
      .StartWith(Unit.Default) 
      .Sample(TimeSpan.FromSeconds(1), _schedulerProvider.NewThread) 
      .Select(a=>Observable.Timer(TimeSpan.FromSeconds(1), _schedulerProvider.NewThread).Select(_=>Unit.Default)) 
      .Switch() 
      .Publish() 
      .RefCount(); 
} 

そしてが、私はこの方法で、静的および非副作用を作成し、それを呼び出します

public IObservable<Unit> GetInactivityObservable() 
{ 
    return _inactivityObservable = _inactivityObservable ?? 
     Observable.FromEventPattern<EventHandler<EventArgs>, EventArgs>(
      h => _activitySource.Activity += h, 
      h => _activitySource.Activity -= h) 
      .Sample(TimeSpan.FromSeconds(1), _schedulerProvider.NewThread) 
      .Select(_ => Unit.Default) 
      .Publish() 
      .RefCount(); 
} 

変化あなたのコンストラクタ、または単にそれをインライン化します。

public class InactivityMonitor : IInactivityMonitor 
{ 
    private readonly ISchedulerProvider _schedulerProvider; 
    private readonly IObservable<Unit> _inactivityObservable; 
    private static readonly TimeSpan SilencePeriod = TimeSpan.FromSeconds(1); 

public InactivityMonitor(IRaiseActivity activitySource, ISchedulerProvider schedulerProvider) 
{ 
    _schedulerProvider = schedulerProvider; 
    _inactivityObservable = Observable.FromEventPattern<EventHandler<EventArgs>, EventArgs>(
      h => _activitySource.Activity += h, 
      h => _activitySource.Activity -= h) 
      .StartWith(null) 
      .Select(a=>Observable.Timer(TimeSpan.FromSeconds(1), _schedulerProvider.NewThread)) 
      .Switch() 
      .Select(_=>Unit.Default) 
      .Publish() 
      .RefCount(); 
} 

//... 

}

Answer 2

だから、戻ってこの問題に取得する機会を持つ私は私の方法でエラーを見ました。

Observable.Intervalは、アクティビティが報告されていない（名前のない非アクティブな観察可能オブジェクトによって）ため、開始されません。アプリを実行すると、少なくとも1つのイベントが発生している可能性が最も高いです。これを修正するために、私は.StartWith(Unit.Default)を以下のように追加しました。これにより、オリジナルの実装で明らかに潜在的なバグであったオブザーバごとにすぐにタイムアウトを開始することができます。

public IObservable<Unit> ObserveInactivity(TimeSpan inactivityTimeout) 
{ 
    return GetInactivityObservable() 
     .StartWith(Unit.Default) // <-- 
     .Select(_ => Observable.Interval(inactivityTimeout, _schedulerProvider.NewThread) 
       .Timestamp() 
       .Select(__ => Unit.Default)) 
     .Switch(); 
}