如何实现仅在收到N个初始项后生效的Rx超时操作符

问题背景 #

我有一个来自仪器的测量值Observable序列，仅当值变化达到一定幅度时才触发测量事件。底层数据源是.NET事件而非IObservable<T>流，我通过Observable.FromEventPattern将其转换为IObservable<T>——该操作会在订阅时注册事件处理器并启动仪器测量，释放时注销处理器并停止测量，因此必须严格管理Observable的生命周期。

现有仪器Observable的实现如下：

// 带测量启停生命周期及订阅时立即返回当前值的FromEventPattern示例
public static IObservable<Double> MeasurementObservable(this MyInstrument instrument)
{
    return Observable.Defer(
      () => Observable
        .FromEventPattern<NewMeasurementEventHandler, Double>(
            h =>
            {
                instrument.NewMeasurementEvent += h;
                instrument.StartMeasuring();
            },
            h =>
            {
                instrument.StopMeasuring();
                instrument.NewMeasurementEvent -= h;
            })
        .Select(eventPattern => eventPattern.EventArgs)
        // 立即返回当前值，之后在事件触发时返回新值
        .Prepend(instrument.CurrentValue)
     );
}

应用场景是：用户启动测量后手动启动被测过程，应用无法直接获知过程的开始/结束，只能通过收到大量变化事件判断过程启动，通过停止收到事件判断过程结束。

我尝试编写一个Reactive操作符，逻辑是：先等待收到至少N个测量值（确认过程运行），再启用超时策略——当新值停止接收时，优雅完成测量并取消订阅底层源。当前实现如下：

public static class ObservableExtensions
{
    /// <summary>
    /// 在Observable产生初始一批项后，实现超时策略：若指定时间内无新项则完成Observable。
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="source"></param>
    /// <param name="initialItems">启用超时策略前需产生的项数。</param>
    /// <param name="dueTime">触发流完成的超时时长。</param>
    /// <returns></returns>
    public static IObservable<T> CompleteWhenUpdatesStop<T>(this IObservable<T> source, Int32 initialItems, TimeSpan dueTime)
    {
        var refCounted = source.Publish().RefCount();
        return refCounted.Take(initialItems)
            .Concat(refCounted.Timeout(dueTime, Observable.Empty<T>()));
    }
}

但这个实现存在问题：尽管用了Publish().RefCount()，但Take(initialItems)完成后会导致RefCount计数下降，若此时Concat还未完成第二个流的订阅，底层订阅会被释放，随后Concat会重新订阅底层Observable。这既破坏了仪器的生命周期管理（重复启停测量），还会因为冷Observable的特性重复输出初始值。我需要消费者能看到底层Observable发出的每个值仅一次，请问如何正确实现「收集N个初始项后才生效的超时逻辑」？

正确实现 #

核心思路是全程仅维护底层源的单个订阅，在订阅内部跟踪已接收的项数，待达到指定数量后再启用超时逻辑，且每次收到新值时重置超时定时器。

实现代码 #

public static class ObservableExtensions
{
    /// <summary>
    /// 在Observable产生指定数量的初始项后，若指定时间内无新项则完成流。
    /// 全程共享底层源的单个订阅，避免重复启停仪器，每个值仅推送一次。
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="source">底层测量值Observable</param>
    /// <param name="initialItems">启用超时前需接收的项数</param>
    /// <param name="dueTime">超时无数据时触发流完成的时长</param>
    /// <returns></returns>
    public static IObservable<T> CompleteWhenUpdatesStop<T>(this IObservable<T> source, int initialItems, TimeSpan dueTime)
    {
        return Observable.Create<T>(observer =>
        {
            int receivedCount = 0;
            IDisposable timeoutTimer = Disposable.Empty;

            // 订阅底层源，全程仅一次
            var sourceSubscription = source.Subscribe(
                value =>
                {
                    // 推送值给观察者，确保每个值仅发送一次
                    observer.OnNext(value);
                    receivedCount++;

                    // 仅当收集够初始项数后，才启用超时逻辑
                    if (receivedCount >= initialItems)
                    {
                        // 重置超时定时器：每次收到新值就取消之前的定时器，重新计时
                        timeoutTimer.Dispose();
                        timeoutTimer = Observable.Timer(dueTime)
                            .Subscribe(_ => observer.OnCompleted());
                    }
                },
                // 转发错误
                observer.OnError,
                // 转发底层源的完成信号
                observer.OnCompleted
            );

            // 组合订阅资源，确保取消订阅时清理所有资源
            return new CompositeDisposable(sourceSubscription, timeoutTimer);
        });
    }
}

实现说明 #

1. 单订阅生命周期：通过Observable.Create直接订阅底层源，全程仅创建一个订阅，完美适配仪器的StartMeasuring/StopMeasuring生命周期，不会重复启停。
2. 计数与超时触发：用receivedCount跟踪已接收的项数，仅当计数达到initialItems后才启动超时逻辑。每次收到新值时都会重置超时定时器，确保只有当真正停止接收数据时才触发流完成。
3. 值的唯一性：每个底层源推送的值都会直接转发给观察者，仅发送一次，避免冷Observable重复输出初始值的问题。
4. 资源清理：用CompositeDisposable管理底层订阅和超时定时器，确保取消订阅时所有资源都被正确释放。

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