c# - Rx 窗口、加入、GroupJoin？

Question

我已经生成/测试了两个要组合起来执行单个查询的 Observable。

一个用户可以有多个角色。每当他们的角色 ID 发生变化时，就需要更新数据。但是数据应该只在查询处于事件状态时更新(当前有一些控件需要数据)。

角色 ID 更改也可能在查询暂停时发生。当查询再次激活时，数据也应该加载。

//Tuple has the Id of the current Role and the time that the Id updated
IObservable<Tuple<Guid, DateTime>> idUpdate

//Tuple has the state of the query (true=active or false=suspended)
//and the time the state of the query updated
IObservable<Tuple<bool, DateTime>> queryStateUpdate 

我要创作

//A hot observable that pushes true whenever the query should execute
IObservable<bool> execute 

我将其分解为两个可以合并的案例，但我不知道如何创建案例可观察对象。

• 情况 a) 角色 ID 已更新且最后状态为事件
• 情况 b) 状态更新为 Active && 这是角色 ID 更新后的第一个事件状态

我浏览了视频、lee campbells 网站、初学者 TOC 等，但我似乎无法找到这个 rx join 的好例子。关于如何创建执行或案例可观察对象的任何想法？

Answer 1

鉴于所描述的问题 - 这有点含糊，因为我没有看到实际 ID (Guid) 的用途，也没有看到 DateTime 值 -我有以下查询似乎可以解决您的问题:

IObservable<bool> execute =
    idUpdate
        .Publish(_idUpdate =>
            from qsu in queryStateUpdate
            select qsu.Item1
                ? _idUpdate.Select(x => true) 
                : Observable.Empty<bool>())
        .Switch();

我已经使用以下 idUpdate 和 queryStateUpdate observables 对此进行了测试。

var rnd = new Random();

IObservable<Tuple<Guid, DateTime>> idUpdate =
    Observable
        .Generate(
            0,
            n => n < 10000,
            n => n + 1,
            n => Tuple.Create(Guid.NewGuid(), DateTime.Now),
            n => TimeSpan.FromSeconds(rnd.NextDouble() * 0.1));

IObservable<Tuple<bool, DateTime>> queryStateUpdate =
    Observable
        .Generate(
            0,
            n => n < 100,
            n => n + 1,
            n => n % 2 == 0,
            n => TimeSpan.FromSeconds(rnd.NextDouble() * 2.0))
        .StartWith(true)
        .DistinctUntilChanged()
        .Select(b => Tuple.Create(b, DateTime.Now));

如果您能就您的问题提供一些说明，我可能会提供更好的答案来满足您的需求。

---

编辑:添加了当 Id 在不活动时更改时所需的“replay(1)”行为。

请注意，我已经摆脱了使用 DateTime 元组的需要。

IObservable<Guid> idUpdate = ...
IObservable<bool> queryStateUpdate = ...

var replay = new ReplaySubject<Guid>(1);
var disposer = new SerialDisposable();
Func<bool, IObservable<bool>, IObservable<Guid>,
    IObservable<Guid>> getSwitch = (qsu, qsus, iu) =>
{
    if (qsu)
    {
        return replay.Merge(iu);
    }
    else
    {
        replay.Dispose();
        replay = new ReplaySubject<Guid>(1);
        disposer.Disposable = iu.TakeUntil(qsus).Subscribe(replay);
        return Observable.Empty<Guid>();
    }
};

var query =
    queryStateUpdate
        .DistinctUntilChanged()
        .Publish(qsus =>
            idUpdate
            .Publish(ius =>
                qsus
                    .Select(qsu =>
                        getSwitch(qsu, qsus, ius))))
        .Switch();