c# - 拍摄 IObservable 的快照

Question

假设我有一个服务:

public interface ICustomersService
{
    IObservable<ICustomer> Customers
    {
        get;
    }
}

执行Customers property 首先获取所有现有客户并将它们传递给观察者，之后它只传递稍后添加到系统中的客户。因此，它永远不会完成。

现在假设我想抓取当前客户的快照(作为 List<ICustomer> )，忽略将来可能添加的任何内容。我怎么做？任何调用 ToList()或其亲属将永远阻塞，因为序列永远不会完成。

我想我可以写我自己的扩展，所以我尝试了这个:

public static class RxExtensions
{
    public static List<T> ToSnapshot<T>(this IObservable<T> @this)
    {
        var list = new List<T>();

        using (@this.Subscribe(x => list.Add(x)));

        return list;
    }
}

这似乎有效。例如:

var customers = new ReplaySubject<string>();

// snapshot has nothing in it
var snapshot1 = customers.ToSnapshot();

customers.OnNext("A");
customers.OnNext("B");

// snapshot has just the two customers in it
var snapshot2 = customers.ToSnapshot();

customers.OnNext("C");

// snapshot has three customers in it
var snapshot3 = customers.ToSnapshot();

我意识到当前的实现取决于调度程序是当前线程，否则 ToSnapshot可能会在收到元素之前关闭订阅。但是，我怀疑我还可以包括 ToSnapshot覆盖需要 IScheduler并确保在结束快照之前收到安排在那里的任何项目。

我找不到 Rx 中内置的这种快照功能。我错过了什么吗？

Answer 1

您可以尝试在您的可观察对象上使用超时

source.Customers().TakeUntil(DateTime.Now).ToEnumerable();