Rx Observable Window 带参数的关闭函数

Question

我正在尝试将可观察对象分离到窗口中（或者出于我的目的，缓冲区也可以），同时能够在自定义位置关闭窗口/缓冲区。

例如我有一个 observable，它产生从 1 开始向上移动的整数。我想在每个可被 7 整除的数字处关闭一个窗口。在这种情况下，我的关闭函数需要将项目作为参数。

Window 方法重载：

Window<TSource, TWindowClosing>(IObservable<TSource>, Func<IObservable<TWindowClosing>>)

要么无法使用此重载来完成，要么我无法绕过它。 Documentation describes that it does exactly what I want 但没有显示示例。此外，它还展示了一个非确定性关闭的示例，这取决于关闭可观察集合发出项目的时间。

Window 操作符将一个可观察的序列分解成连续的 非重叠窗口。当前窗口的结束和开始 下一个窗口由一个可观察的序列控制，即 作为输入传递的 windowClosingSelect 函数的结果 操作员的参数。运算符可用于对集合进行分组 事件进入一个窗口。例如，事务的状态可能是 被观察到的主要序列。这些状态可能包括： 准备、准备、活动和承诺/中止。主要序列 可以包括所有这些状态，如果它们按顺序出现的话。这 windowClosingSelect 函数可以返回一个可观察的序列 仅在 Committed 或 Abort 状态下产生值。这个会 关闭代表特定交易事件的窗口 交易。

我在想像下面这样的东西可以完成这项工作，但我必须自己实现它：

Window<TSource, TWindowClosing>(IObservable<TSource>, Func<TSource, bool>)

• 是否可以使用内置函数进行这种窗口化（我知道我可以自己构建一个）？
• 是否可以根据发射的项目关闭窗口，或者仅在非确定性的情况下，从可观察窗口中发射项目后？

Answer 1

使用带有Where 子句的原始序列作为结束序列。如果您的源序列很冷，请使用Publish 和RefCount 使其正常工作。

var source = ...;
var sharedSource = source.Publish().RefCount();
var closingSignal = sharedSource.Where(i => (i % 7) == 0);
var windows = sharedSource.Window(() => closingSignal);