c# - 如何根据第二个流将 Observable 流分成 block ？

Question

我以为这很容易，但我的大脑现在正在融化..

问题

鉴于以下 IObservable<int>流:
1
1
0
0
0
1
0
0
1
0
1

我想把它拆分成一个 IObservable<IEnumerable<int>>形式流

1

1 0 0 0

1 0 0

1 0

1

所以只要有 0，它就会被添加到 IEnumerable 中，当出现 1 时，就会启动一个新的 List；这是对我真正的问题的更清晰的定义。

到目前为止我的方法

我认为一个好的解决方案是先将其转换为 IObservable<IObservable<int>>通过 Window方法，然后使用 ToEnumerable , 但不知何故我无法让它工作..我用过 Zip和 Skip(1)为了得到最后一个元素的差异，我使用了 DistinctUntilChanged() ， 也。我为您保留了我尝试过的所有变体...

可能我最接近的是这段代码:

int[] ints = new[] { 1, 1, 0, 0, 0, 1, 0, 0, 1, 0, 1 };
var observable = Observable.Interval(TimeSpan.FromMilliseconds(1000)).Take(11).Select(i => ints[i]);

Subject<int> subject = new Subject<int>();
observable.Subscribe(subject);

var observableDiff = subject.Skip(1).Zip(subject, (n, p) => new { Previous = p, Next = n });
var windows = observable.Window(() => observableDiff.Where(x => x.Next == 1));

int index = 0;
windows.Subscribe(window =>
{
  Console.WriteLine(string.Format("new window [{0}] ", index++));
  window.Subscribe(number => Console.WriteLine(number));
});

这返回了很好的结果，但不幸的是它在最后崩溃了..

new window [0]
1
new window [1]
1
0
0
0
new window [2]
1
0
0
new window [3]
1
0
new window [4]
new window [5]
new window [6]
new window [7]
new window [8]
new window [9]
<-- it goes on here until window ~ [80] with a stackoverflow exception

如果我的代码中不存在那个错误，我就会实现它...

任何帮助将不胜感激。 :)

编辑:我使用 Rx-Experimental，但它没有什么不同(使用 LinqPad 进行了检查)。 也删除了主题，它没有影响任何东西。看来我的新方法(Edit2)，你需要一个主题，否则窗口的开始是完全奇怪的。

Edit2:稍微更改了问题，以更好地突出我的问题，抱歉。还更新了我的解决方案。

Answer 1

这对我有用:

var ints = (new[] { 1, 1, 0, 0, 0, 1, 0, 0, 1, 0, 1 }).ToObservable();

var result =
    ints
        .Publish(ns =>
            ns
                .Where(n => n == 1)
                .Select(n =>
                    ns.TakeWhile(m => m == 0).StartWith(n).ToArray())
        ).Merge();

我使用 Publish 来确保 ints observable 被视为“热”而不是“冷”。

我的结果是这样的: