使用 Reactive 对 Observable 的某些部分进行求和

Question

我正在尝试使用反应式运算符来查找可观察对象发出的值的单个总和。最终目标是发出单独的总和。序列看起来像这样。我想要添加的那些作为连续组（长度不同）出现，在我想要丢弃的值之间具有不同的频率。我要添加的字段有一个 bool 类型且值为 true 的字段。

-(F,2)-(T,4)-(T,2)-(T,7)-(F,8)-(F,9)-(F,1)-(T,2)-(T,1)-(F,1)-

到目前为止我尝试了什么：

myObservable.
    .Where(x => x.IsItUseful == true)
    .Aggregate(0.0, (sum,currentItem) => sum + currentItem.Value)
    .Subscribe("NotYet")

这个返回所有被标记为真的元素的总和。

myObservable
    .SkipWhile(x => x.IsItUseful == false)
    .TakeWhile(x => x.IsItUseful == true)
    .Aggregate(0.0, (sum, currentItem) => sum + currentItem.Item3)
    .Subscribe("NotYetAgain");

这个只给出第一组的总和。

现在我正在尝试这些方法。

myObservable
    .Buffer(myObservable.DistinctUntilChanged(x => x.IsItUseful => true)
    .Subscribe("NotSure")

我对 BufferBoundary 和 BufferClosingSelector 仍然很模糊。我认为一旦我处理了一组有效值，就会打开一个新缓冲区。并且这个新缓冲区将具有从该点到另一个有效组结束的值。这意味着在第二组之前我也会选择一些无效的值。我无法在 Buffer 上找到一些使用打开和关闭选项的示例。不确定这是否也是正确的方法。

最后一个选项是我在 Buffer 上编写一个扩展方法并将我的自定义逻辑放在那里。但如果有一个开箱即用的解决方案，我会更喜欢。

Answer 1

我在这里推荐两种主要方法。一个使用Scan，另一个使用Buffer/Window。它们都有可以解决的边缘情况问题，但需要在问题方面明确。

这是Scan 解决方案：

var result = source
    .Scan((0, true), (state, value) => (value.IsItUseful ? state.Item1 + value.Value : 0, value.IsItUseful))
    .Publish(_tuples => 
        _tuples.Zip(_tuples.Skip(1), (oldTuple, newTuple) => (oldTuple, newTuple))
    )
    .Where(t => t.oldTuple.Item2 == true && t.newTuple.Item2 == false)
    .Select(t => t.oldTuple.Item1);

Scan 与Aggregate 类似，只是更有用：Aggregate 最后只会输出一个值；而Scan 发出中间值。所以我们在那里跟踪运行总和，当我们看到错误时重置为 0。下一步（Zip）将最新消息与其前身结合起来，因此我们可以确定是否必须发出：如果新标志值为假，但旧标志值为真，我们想发出。然后我们发出旧的总和。

如果最后一个标志值为真，这里有一个极端情况问题：我假设你想在OnCompleted 上发出，但目前不会发生。请说明是否需要。

这是Window 解决方案：

var result2 = source
    .Publish(_values => _values
        .Window(_values.Select(v => v.IsItUseful).DistinctUntilChanged().Where(b => b == false))
    )
    .SelectMany(o => o.Where(a => a.IsItUseful).Sum(a => a.Value));

通过明显的新错误窗口，然后将它们相加，类似于您提出的建议。

这里的边缘情况问题是，如果您以错误开头/结尾（就像您的样本集一样），您最终会得到一个前导和尾随 0。删除这些也需要进行一些清理。

仅供参考：Window 和 Buffer 实际上是相同的：它们具有相同的重载，并且每个值都归入“窗口”。 Window 将它们作为可观察的流返回，Buffer 将它们保存在一个列表中，该列表在窗口关闭时返回。更多look here.

---

如果其他人想测试这个，这里是运行器代码：

public class Message
{
    public Message(bool b, int v)
    {
        IsItUseful = b;
        Value = v;
    }

    public bool IsItUseful { get; set; }
    public int Value { get; set; }
}

var values = new List<Message>
{
    new Message(false, 2),
    new Message(true, 4),
    new Message(true, 2),
    new Message(true, 7),
    new Message(false, 8),
    new Message(false, 9),
    new Message(false, 1),
    new Message(true, 2),
    new Message(true, 1),
    new Message(false, 1),
};

var source = values.ToObservable();

var result = source
    .Scan((0, true), (state, value) => (value.IsItUseful ? state.Item1 + value.Value : 0, value.IsItUseful))
    .Publish(_tuples => 
        _tuples.Zip(_tuples.Skip(1), (oldTuple, newTuple) => (oldTuple, newTuple))
    )
    .Where(t => t.oldTuple.Item2 == true && t.newTuple.Item2 == false)
    .Select(t => t.oldTuple.Item1);

var result2 = source
    .Publish(_values => _values
        .Buffer(_values.Select(v => v.IsItUseful).DistinctUntilChanged().Where(b => b == false))
    )
    .Select(o => o.Where(a => a.IsItUseful).Sum(a => a.Value));

result.Dump();  //Linqpad
result2.Dump(); //Linqpad