是否可以在按键分组的热门 observable 中对最新项目进行采样? [英] Is it possible to sample the latest item in a hot observable grouped by key?

问题描述

在 Rx.NET 中，是否可以在按 key 分组的热门 observable 中采样最新项目?

In Rx.NET, is it possible to sample the latest item in a hot observable grouped by key?

例如，如果我有一个 IObservable，其中 Price 是:

For example, if I have an IObservable<Price>, where Price is:

Price 
- Key
- Bid
- Offer

让我们假设 IObservable 链接到外部价格馈送.

Let's assume that the IObservable is linked to an external price feed.

我是否能够检索所有最新的 Price，按 Key 分组，使用 Rx 每 1 秒采样一次?

Am I able to retrieve all latest Prices, grouped by Key, sampled every 1 second using Rx?

推荐答案

假设有一些可观察的 source，这将返回最后一秒出现的所有按关键字分组和采样的价格.

Assuming some observable source, this returns all prices grouped and sampled by key that have come in in the last second.

var sampled = source
    .GroupBy(p => p.Key)
    .SelectMany(o => o.Sample(TimeSpan.FromSeconds(1)));

如果有一些价格在最后一秒没有收到消息，则不会包含在内.

If there's some price which hasn't received a message in the last second, that will not be included.

如果您希望包含旧价格，这将起作用:

If you want old prices included, this will work:

var sampled2 = source
    .Scan(ImmutableDictionary<int, Price>.Empty, (state, p) => state.SetItem(p.Key, p))
    .Replay(1)
    .RefCount();
var dummySubscription = sampled2.Subscribe();
var result = Observable.Interval(TimeSpan.FromSeconds(1))
    .SelectMany(_ => sampled2.Take(1).SelectMany(state => state.Values));

确保在处理完 result observable 后处理 DummySubscription.

Just make sure to dispose of the DummySubscription when done with the result observable.

这篇关于是否可以在按键分组的热门 observable 中对最新项目进行采样?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！