Rxのフレームワークは：...ポーリング低レベルのデバイスの状態（の変化）によって生成された観察光源与えられた原観察シーケンス

Question

を中断することなく

// observable source metacode: 
IObservable<DeviceState> source = Observable.Interval(TimeSpan.FromSeconds(0.5)) 
    .Select(tick => new DeviceState(_device.ReadValue())) 
    .DistinctUntilChanged(); 

をタイムアウトにアクションを実行します.. 。そしてUIを更新し、消費者...

// UI metacode: 
service.GetObservableDeviceStates() 
    .Subscribe(state => viewModel.CurrentState = state.ToString()); 

...私はソースにサブスクリプションを中断することなく、ソースの「非アクティブ」のx秒後にカスタムアクションを実行する必要があります。このようなもの：

// UI metacode: 
service.GetObservableDeviceStates() 
    .DoOnTimeout(TimeSpan.FromSeconds(x),() => viewModel.CurrentState = "Idle") 
    .Subscribe(state => viewModel.CurrentState = state.ToString()); 

ベストプラクティスは何ですか？心に来る可能性のある解決策は、（私はのRxのnoobだ）、次のとおりです。

1. Buffer（それはそう読めるない場合でも）
2. this Timeout overload遊ん。

3. 何が（代わりにDistinctUntilChangedを使用しての）変化しない何か特別な "サービス側" を返すと、UIのコードでそれを扱う：

   service.GetObservableDeviceStates（） .Subscribe（状態=> viewModel.CurrentState = state.Special？ "Idle"：state.ToString（））;

EDIT：in the answerが報告されているように、溶液である：

 service.GetObservableDeviceStates() 
      .Do(onNext) 
      .Throttle(TimeSpan.FromSeconds(x)) 
      .Subscribe(onTimeout); 

EDIT2（警告）

onNextとOnTimeoutパス更新UIコンポーネント場合、CrossThreadExceptionsを回避する ObserveOn（uiSynchronizationContext） Throttleは別のスレッドで動作するため、必要です。一定の時間であなたを通知していない観察できると述べている場合、例えばのためにデフォルト値を返すことかOnError -

 service.GetObservableDeviceStates() 
      .ObserveOn(uiSynchronizationContext) 
      .Do(onNext) 
      .Throttle(TimeSpan.FromSeconds(x)) 
      .ObserveOn(uiSynchronizationContext) 
      .Subscribe(onTimeout); 

Answer 1

タイムアウトは、多かれ少なかれ、単一の非同期操作を表す観測するためのものです。

あなたが探しているオペレータは、最初はそれほど見えないかもしれませんが、Throttleです。 Throttle(p)は、ソースストリームが期間pの値を生成していないときに値を生成するストリームを提供します。

既存のコードと並行して、source.Throttle(period).Do(...side effect)を使用できます。

Answer 2

私は個人的にこれを行うためのDoメソッドを避けます。この例ではコードをかなり簡単にしていますが、すぐに「Do」をコードベースに入れてしまえばすぐにスパゲッティを見つけることができます。

Amb、Timer、TakeUntil、Throttleなどの組み合わせを使用して、探している結果を取得してMonad *を維持することも検討できます。簡単に言えば、ステータス値のシーケンスを順番通りに持ち、コードにタイマーを入れる（つまり、サービスにロードする）必要がないことが理想です。

public IObservable<DeviceStatus> GetObservableDeviceStates(TimeSpan silencePeriod) 
{ 
    return Observable.Create<DeviceStatus>(
    o=> 
    { 
     var idle = Observable.Timer(silencePeriod).Select(_=>new DeviceStatus("Idle")); 

     var polledStatus = Observable.Interval(TimeSpan.FromSeconds(0.5)) 
         .Select(tick => new DeviceStatus(_device.ReadValue())) 
         .DistinctUntilChanged() 
         .Publish(); 

     var subscription = (from status in polledStatus 
          from cont in Observable.Return(status).Concat(idle.TakeUntil(polledStatus)) 
          select cont) 
        .Subscribe(o); 

     return new CompositeDisposable(subscription, polledStatus.Connect()); 
    }); 
} 

このコードでは、指定した変更の無音期間が発生すると、サービスでアイドル状態値が返されるようになりました。

これはあなたのUIメタコードはシンプルとどまり、DeviceStatusに関連するロジックは、それが

// UI metacode: 
service.GetObservableDeviceStates(TimeSpan.FromSeconds(2)) 
    .Subscribe(state => viewModel.CurrentState = state.ToString());