如何在可观察的反应性扩展中记住以前的有效载荷？

Question

我有一个 IObservable boolObservable，发布者就像开关一样。还有具有某些值的变量 currentItem 和包含预定义不可变值的 defaultItem。每当发布 'true' 时，应记住 currentItem 的值，并将 currentItem 设置为 defaultItem。当 'false' 发布时，currentItem 的值应更改为前一个。实现此目的的最简单方法是使用局部变量。

IObservable<bool> boolObservable;
string currentItem;
string defaultItem;

string local = null;
boolObservable.Subscribe(x =>
  {
    if (x) {
      local = currentItem;
      currentItem = defaultItem;
    }
    else
      currentItem = local;
  });

当我不得不重复多次并每次都不同地命名用于存储的局部变量时，问题就开始了。有没有一种方法可以使用 Rx 运算符来 "remember" 不仅是最新的有效载荷，还包括之前的有效载荷？

UPD: 下面@ibebbs 的回答是绝对正确的，Scan 运算符提供了必要的行为：

var switchSource = new BehaviorSubject<bool>(false);
string currentItem = "FromDb";
string defaultItem = "Default";

switchSource
    .Skip(1)
    .Scan(string.Empty, (local, b) =>
    {
        if (b)
        {
            local = currentItem;
            currentItem = defaultItem;
        }
        else
            currentItem = local;

        return local;
    })
    .Subscribe();

Console.WriteLine($"Initial value: {currentItem}"); // FromDb
switchSource.OnNext(true);
Console.WriteLine($"Default value: {currentItem}"); // Default
switchSource.OnNext(false);
Console.WriteLine($"Initial value again: {currentItem}"); // FromDb
currentItem = "UserChanged";
Console.WriteLine($"UserChanged value: {currentItem}"); // UserChanged
switchSource.OnNext(true);
Console.WriteLine($"Default value: {currentItem}"); // Default
switchSource.OnNext(false);
Console.WriteLine($"UserChanged value again: {currentItem}"); // UserChanged

Answer 1

您可以使用 Scan 通过可观察的序列对状态进行线程化。例如：

public static IObservable<string> EitherItemOrDefault(IObservable<bool> switchSource, IObservable<string> itemSource, string defaultValue)
{
    return switchSource
        .WithLatestFrom(itemSource.StartWith(string.Empty), (switchValue, itemValue) => (Switch: switchValue, Item: itemValue))
        .Scan(
            (Current: string.Empty, Local: string.Empty),
            (seed, tuple) => tuple.Switch 
                            ? (Current: defaultValue, Local: tuple.Item) 
                            : (Current: seed.Local, Local: seed.Local))
        .Select(tuple => tuple.Current);
}

这需要一个 IObservable<bool> 作为你的 "switches" 和一个 IObservable<string> 作为项目（currentItem 在你的例子中）和 returns 一个 IObservable<string> 当 switchSource 发出 False 时将包含 defaultValue，当 switchSource 发出 True 时将包含最后一个 currentItem 值。

注意：我可能把 current 和 local 弄反了。已经晚了，你的例子清楚地说明了为什么应该避免全局状态......推理起来太棘手了。

如何在可观察的反应性扩展中记住以前的有效载荷？

How to memorize previous payload in reactive extensions observable?

c#

system.reactive