反应性扩展 - Abort/Cancel 一个 OnCompleted

Question

我有一个显示状态消息的控件；控件在一段时间后隐藏。这是处理它的代码：

private void ShowFor(TimeSpan? delay)
{
   Visible = true;

   if (!delay.HasValue) return; 

  // _pauseTimer is a MultipleAssignmentDisposable
  _pauseTimer.Disposable = Observable
     .Timer(delay.Value)
     .ObserveOn(SynchronizationContext.Current)
     .Subscribe(
          onNext:  _ => { /* do nothing */ },
     onCompleted: () => { Visible = false; },
         onError:  e => { /* what could possibly go wrong? */});
}

显示控件，等待 n 秒，隐藏控件。十分简单。

问题是在此计时器到时之前收到另一条消息时该怎么办。第二条消息显示，然后第一个计时器到期并过早隐藏控件。

如何"abort" 以前的定时器？处理 pauseTimer.Disposable？

Answer 1

您可以将控件的显示与隐藏分开。假设消息来自 IObservable<string> messages（如果当前不是，则很容易进行设置）然后订阅此消息并设置控件可见，并显示 OnNext.

中的消息

分别订阅同一个流，并应用节流以在您需要多长时间后隐藏消息，例如：

messages.Throttle(TimeSpan.FromSeconds(1))
        .ObserveOn(SynchronizationContext.Current)
        .Subscribe(_ => control.Visible = false);

仅当在所需的延迟时间内未看到新消息时，节流阀才会发出。

为了解决不同的延迟，请将您的消息源设为包含消息和严重性的类型。为简单起见，我将使用 Tuple<string, Timespan>，但您可以使用枚举来表示严重性并做一些更详细的事情。 Throttle 有一个过载，可以根据任何流改变它的持续时间。您可以根据消息持续时间创建 Throttle 源：

// assuming messages is `Tuple<string, Timespan>`
var delayStream = messages.Throttle(
    messages.SelectMany(x => Observable.Timer(x.Item2)))

这将根据消息的严重性创建不同的限制，并且您可以对严重警告使用非常大的（最大）超时 - 或者只为它们发出 Observable.Empty 而不是使用计时器。

请注意，在此方案下（以及您使用单个控件的描述），一条新消息将替换之前的一条消息。一个小的改变将使这更容易处理 - 如果您的控件显示多条消息并且每条消息都带有一个 id，delayStream 可用于决定需要从当前列表中删除哪一条。

Answer 2

将 _pauseTimer 更改为 SerialDisposable 而不是 MultipleAssignmentDisposable，然后每次您这样做时 _pauseTimer.Disposable = newDisposable 它都会处理其当前订阅以订阅下一个.

反应性扩展 - Abort/Cancel 一个 OnCompleted

Reactive Extensions - Abort/Cancel an OnCompleted

c#

reactive-programming

system.reactive