如何重新订阅特定点的序列？

Question

我正在尝试解决以下问题： a) 订阅者在一段时间内接收来自 IObservable 的事件。然后它取消订阅，做一些事情然后再次订阅。在这里，它应该从执行取消订阅的同一点开始接收事件。 b) 这种行为对于多订阅者模型来说是理想的。例如，当一个人取消订阅时，其他人应该继续接收事件。

RX方面有什么建议吗？

提前致谢！

Answer 1

听起来您需要一个“可暂停”流。假设一次只有 1 个订阅者会处理这些值（而其他订阅者只是等待），这个 解决方案可能就是您所需要的。

Answer 2

这是一种相当简单的 Rx 方法，可以执行您想要从 我的复制的操作回答 href="https://stackoverflow.com/questions/7620182/pause-and-resume-subscription-on-cold-iobservable">另一个问题。我创建了一个名为 Pausable 的扩展方法，它采用一个源 observable 和第二个 boolean 值的 observable，用于暂停或恢复该 observable。

public static IObservable<T> Pausable<T>(
    this IObservable<T> source,
    IObservable<bool> pauser)
{
    return Observable.Create<T>(o =>
    {
        var paused = new SerialDisposable();
        var subscription = Observable.Publish(source, ps =>
        {
            var values = new ReplaySubject<T>();
            Func<bool, IObservable<T>> switcher = b =>
            {
                if (b)
                {
                    values.Dispose();
                    values = new ReplaySubject<T>();
                    paused.Disposable = ps.Subscribe(values);
                    return Observable.Empty<T>();
                }
                else
                {
                    return values.Concat(ps);
                }
            };

            return pauser.StartWith(false).DistinctUntilChanged()
                .Select(p => switcher(p))
                .Switch();
        }).Subscribe(o);
        return new CompositeDisposable(subscription, paused);
    });
}

它可以这样使用：

var xs = Observable.Generate(
    0,
    x => x < 100,
    x => x + 1,
    x => x,
    x => TimeSpan.FromSeconds(0.1));

var bs = new Subject<bool>();

var pxs = xs.Pausable(bs);

pxs.Subscribe(x => { /* Do stuff */ });

Thread.Sleep(500);
bs.OnNext(true);
Thread.Sleep(5000);
bs.OnNext(false);
Thread.Sleep(500);
bs.OnNext(true);
Thread.Sleep(5000);
bs.OnNext(false);