Rx.NET如何在连接开始到Connected状态后5秒内跳过Divergence变更

改造后完整代码 #

enum Status { NotConnected, Connecting, Connected }

class Provider
{
    public Status State { get => m_state; set => this.RaiseAndSetIfChanged(ref m_state, value); }
    public double Divergence { get => m_divergence; set => this.RaiseAndSetIfChanged(ref m_divergence, value); }
}

public void Run(Provider provider)
{
    // 构造允许处理Divergence事件的信号流：true=允许响应，false=跳过
    var allowProcessSignal = provider.WhenAny(x => x.State, x => x.Value)
        .SelectMany(state => state switch
        {
            Status.NotConnected => Observable.Return(true),
            Status.Connecting => Observable.Return(false),
            Status.Connected => Observable.Return(false)
                .Concat(Observable.Return(true).Delay(TimeSpan.FromSeconds(5))),
            _ => Observable.Return(true)
        })
        .DistinctUntilChanged()
        .StartWith(true); // 初始状态为NotConnected，默认允许响应

    provider.WhenAny(viewItem => viewItem.Divergence, change => change.Sender)
        .Throttle(TimeSpan.FromMilliseconds(500))
        // 关联允许信号，过滤掉需要跳过的事件
        .WithLatestFrom(allowProcessSignal, (provider, isAllow) => (provider, isAllow))
        .Where(t => t.isAllow)
        .Select(t => t.provider)
        .Subscribe(OnDivergenceChanged);
}

private void OnDivergenceChanged(Provider provider)
{
    // 原有业务逻辑保持不变
    ...
}

实现逻辑说明 #

• 信号流allowProcessSignal跟随Provider的State状态动态变更：
  • 状态变为NotConnected时直接标记为允许响应
  • 状态变为Connecting时直接标记为跳过响应
  • 状态变为Connected时先标记为跳过响应，等待5秒后自动切换为允许响应
• SelectMany操作符会自动取消上一个状态的未完成任务：如果在Connected的5秒等待期内状态切回NotConnected，未执行的延迟任务会被直接取消，立即切换为允许状态，符合循环流转的业务要求
• 原有Divergence的500ms防抖逻辑保持不变，仅在允许信号为true时才会触发后续的业务回调

低版本C#兼容 #

如果项目使用的C#版本不支持switch表达式，可将SelectMany内的逻辑替换为传统switch写法：

.SelectMany(state => 
{
    switch (state)
    {
        case Status.NotConnected:
            return Observable.Return(true);
        case Status.Connecting:
            return Observable.Return(false);
        case Status.Connected:
            return Observable.Return(false)
                .Concat(Observable.Return(true).Delay(TimeSpan.FromSeconds(5)));
        default:
            return Observable.Return(true);
    }
})

内容的提问来源于stack exchange，提问作者Pavel Ivchenkov