拍摄 ReplaySubject<T> 缓冲区的快照答案

【问题标题】：Taking a snapshot of ReplaySubject<T> buffer拍摄 ReplaySubject<T> 缓冲区的快照
【发布时间】：2013-03-05 06:33:07
【问题描述】：

我有一个大的 ReplaySubject，它允许订阅者正常接收重播缓冲区和未来的通知。另外，我希望能够拍一张“快照” 的当前缓冲区并将其作为列表同步返回，而无需订阅。

有没有办法做到这一点？

谢谢

【问题讨论】：

标签： .net system.reactive

【解决方案1】：

您可以不只是订阅，接收项目，然后取消订阅吗？

public static List<T> Snapshot<T>(ReplaySubject<T> subject)
{
    List<T> snapshot = new List<T>();
    using (subject.Subscribe(item => snapshot.Add(item))
    {
        // Deliberately empty; subscribing will add everything to the list.
    }
    return snapshot;
}

当然，这是假设订阅ReplaySubject<T> 会同步调用元素处理程序。你会想检查一下，但这是我所期望的。

您还应该考虑是否要以某种方式处理错误/完成。

【讨论】：

我正要描述这是错误的，因为主题可能还没有完成，但它实际上是唯一的方法（并且没有错）。聪明的！话虽如此，我可能会从 ReplaySubject 复制源代码并添加一个显式方法来执行此操作
谢谢大家，这很好，我只是假设我可能还缺少其他东西。

【解决方案2】：

仅因为@PaulBetts 说只有一种方法可以做到这一点:)

注意：我不推荐这种方式；使用飞碟的方法，你以后会感谢自己的。

所以ReplaySubject<T> 的神奇之处在于它将通过OnNext 接收到的任何值排队到内部Queue<TimeInterval<T>> 上。所以我们可以编写一个包装器，在回放主题的私有细节中摆弄以获取该信息：

public class FixedReplaySubject<T> : ISubject<T>
{
    private ReplaySubject<T> _inner;
    private Func<Queue<TimeInterval<T>>> _snapshotGetter;

    public FixedReplaySubject(ReplaySubject<T> source)
    {
        _inner = source;
        var expr = Expression.Lambda(
            typeof(Func<Queue<TimeInterval<T>>>), 
            Expression.Field(
                Expression.Constant(source), 
                source.GetType()
                    .GetField("_queue", BindingFlags.NonPublic|BindingFlags.Instance)));
        _snapshotGetter = (Func<Queue<TimeInterval<T>>>)expr.Compile();
    }

    public IEnumerable<TimeInterval<T>> Snapshot()
    {
        return _snapshotGetter();
    }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        return _inner.Subscribe(observer);
    }
    public void OnNext(T value)
    {
        _inner.OnNext(value);
    }
    public void OnCompleted()
    {
        _inner.OnCompleted();
    }
    public void OnError(Exception error)
    {
        _inner.OnError(error);
    }
    public void Dispose()
    {
        _inner.Dispose();
    }
}

测试台：

void Main()
{
    var src = new ReplaySubject<int>();
    src.OnNext(1);
    src.OnNext(2);
    src.OnNext(3);
    src.OnNext(4);
    src.OnNext(5);
    src.OnNext(6);
    var heh = new FixedReplaySubject<int>(src);
    heh.Snapshot().Dump();
}

结果：（TimeInterval<T> 只是价值 + 时间）

1 00:00:00.0010265 
2 00:00:00.0010278 
3 00:00:00.0010278 
4 00:00:00.0010282 
5 00:00:00.0010282 
6 00:00:00.0010286

【讨论】：