可观察.每个订阅的间隔重新启动

本文关键字：重新启动 观察 | 更新日期: 2023-09-27 17:56:01

我第一次尝试定义测量计划是：

    var schedule = Observable.Concat(
                Observable.Interval(TimeSpan.FromSeconds(1)).Take(3),
                Observable.Interval(TimeSpan.FromSeconds(3)).Take(3),
                Observable.Interval(TimeSpan.FromSeconds(5)));

不幸的是，它在取消订阅并重新订阅时重新启动，这在我的情况下不是理想的行为。因此，我带来了这样的东西：

    class Schedule : IObservable<DateTime>, IDisposable
    {
        readonly ISubject<DateTime> _subject;
        readonly IDisposable _subscrption;
        public Schedule()
        {
            _subject = new BehaviorSubject<DateTime>(DateTime.UtcNow);
            _subscrption = Observable.Concat(
                Observable.Interval(TimeSpan.FromSeconds(1)).Take(3),
                Observable.Interval(TimeSpan.FromSeconds(3)).Take(3),
                Observable.Interval(TimeSpan.FromSeconds(5)))
                .Select(i => DateTime.UtcNow)
                .Subscribe(_subject);
        }
        public IDisposable Subscribe(IObserver<DateTime> observer)
        {
            return _subject.Subscribe(observer);
        }
        public void Dispose()
        {
            _subscrption.Dispose(); 
        }
    }

它可以工作，但使用后需要处理。有没有简单的方法可以在不公开 IDisposable 的情况下定义计划？

可观察.每个订阅的间隔重新启动

如果您要取消订阅所有观察者，然后重新订阅，那么确实无法保留Disposable来跟踪连接。因为您的可观察对象无论如何都不会直接知道最新的退订是否真的是最后一个。

我建议使用Generate而不是连接可观察量

IConnectableObservable<DateTime> source = Observable.Generate(
  0,
  _ => true,
  x => x + 1,
  x => DateTime.UtcNow,
  x => {
    if (x < 3) return TimeSpan.FromSeconds(1);
    else if (x < 5) return TimeSpan.FromSeconds(3);
    else return TimeSpan.FromSeconds(5);
}).Publish();

Publish 会将您的可观察量转换为可连接可观察量，并为您提供两个选项来管理源实际生成事件的时间。

//1) Explicit connect
IDisposable connection = source.Connect();
//2) RefCounted connection
IObservable<DateTime> rcSource = source.RefCount();

在第一个版本中，一旦连接，源将变为"热"，当您想要断开连接时，您应该处理连接。

在第二秒中，当源没有更多的观察者时，它将自动断开连接。

另请参阅热与冷可观察量

这是我定义的不需要处置的解决方案。不过它非常复杂，所以如果可以使用 Rx 仪器简化某些内容，我将不胜感激。

用法：

  IObservable<DateTime> schedule = new Schedule()
            .Setup(TimeSpan.FromSeconds(1), 3)
            .Setup(TimeSpan.FromSeconds(3), 3)
            .Setup(TimeSpan.FromSeconds(5));

其中时间表是：

    class Schedule : IObservable<DateTime>
    {
        readonly TimeSpan TimerPrecision = TimeSpan.FromMilliseconds(1);
        readonly IEnumerable<TimeSpan> Intervals;
        readonly IEnumerator<DateTime> Event;
        public Schedule()
            : this(new TimeSpan[0])
        {
        }
        Schedule(IEnumerable<TimeSpan> intervals)
        {
            Intervals = intervals;
            Event = Start();
            Event.MoveNext();
        }
        public Schedule Setup(TimeSpan interval)
        {
            return Setup(interval, Int32.MaxValue);
        }
        public Schedule Setup(TimeSpan interval, int repeat)
        {
            return new Schedule(
                Intervals.Concat(
                    Enumerable.Repeat(interval, repeat)));
        }
        public IDisposable Subscribe(IObserver<DateTime> observer)
        {
            var timer = new System.Timers.Timer() { AutoReset = true };
            timer.Elapsed += (s, a) => 
            {                                                
                observer.OnNext(DateTime.UtcNow);
                if (!TryArm(timer))
                    observer.OnCompleted();
            };
            if (!TryArm(timer))
                observer.OnCompleted();
            return timer;
        }
        IEnumerator<DateTime> Start()
        {
            var time = DateTime.UtcNow;
            yield return time;
            foreach (var interval in Intervals)
            {
                time += interval;
                yield return time;
            }
        }
        TimeSpan Delay()
        {
            var now = DateTime.UtcNow;
            lock (Event)
                while (Event.Current - DateTime.UtcNow < TimerPrecision)
                    Event.MoveNext();
            return Event.Current - now;
        }
        bool TryArm(System.Timers.Timer timer)
        {
            try
            {
                timer.Interval = Delay().TotalMilliseconds;
                timer.Start();
                return true;
            }            
            catch(ObjectDisposedException)   
            {
                return false;
            }
            catch(InvalidOperationException)   
            {
                return false;
            }
        }
    }
相关文章：