c# - Rx 框架 : execute an action on timeout without interrupting the original observable sequence

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 菜鸟):

1. Buffer (即使它不是那么可读)
2. 玩转 this Timeout overload ;

3. 在没有任何变化时返回一些特殊的“服务端”(而不是使用 DistinctUntilChanged)并在 UI 代码上处理它:

   服务.GetObservableDeviceStates() .订阅(状态=> viewModel.CurrentState = state.Special ? “空闲”:state.ToString());

编辑:据报道in the answer ，解决方案是:

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

EDIT2(警告)

如果 onNext 和 onTimeout 更新 UI 组件，为避免 CrossThreadExceptions 两个 ObserveOn(uiSynchronizationContext) 是必需的，因为 Throttle 在另一个线程上工作!

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

Answer 1

Timeout 或多或少适用于表示单个异步操作的可观察对象 - 例如，如果所述可观察对象在一定时间内未通知您，则返回默认值或 OnError。

您要找的接线员是Throttle ，即使一开始看起来不像。 Throttle(p) 为您提供一个流，当源流在 p 期间未生成值时生成一个值。

与现有代码并行，您可以使用 source.Throttle(period).Do(...side effect)。