c# - 如何使用 IObservable 将 IObservable 分割成不同长度的 IObservable

Question

我有一个“值(value)观”IObservable<T>这是返回 T必须按顺序组合成可变长度数组的元素，我有一个“控制”IObservable<int>这告诉我下一个数组必须有多长。删除一个元素、重复它或将结果打乱都会使结果变得毫无意义。

这是我在 Rx.NET 中重写的串行连接机器人项目。

IObservable<char> values = new [] {'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H' }.ToObservable();
IObservable<int> control = new [] { 1, 4, 2 }.ToObservable();
IObservable<char[]> result = control.SelectMany(length => values.Take(length).ToArray());

我想看到这样的东西:

values  ----A--B--C--D--E--F--G--H-->
control --1-----4---------------2--->
result  ---[A]---------[BCDE]--[FG]->

但是到目前为止我的尝试结果是

-[A]-[AB]-[ABCD]->

Answer 1

好的，这是满足我所有需求的代码。 Progman，您的建议在使这项工作发挥作用方面发挥了重要作用。在这里，整齐地包裹着 Observable.Create并变成了 IObservable<T> 上的扩展方法, 用一次性处理压缩序列上的订阅。

    public static IObservable<T[]> Chop<T>(this IObservable<T> values, IObservable<int> control) =>
        Observable.Create<T[]>(observer => 
        {
            List<T> buffer = new List<T>();
            return values.Zip(control.SelectMany(length => Enumerable.Repeat(length, length)), 
                              (value, length) => (value, length))
                         .Subscribe(next => 
                         {
                             buffer.Add(next.value);
                             if (buffer.Count == next.length)
                             {
                                 observer.OnNext(buffer.ToArray());
                                 buffer.Clear();
                             }
                         });
        });

示例输出:

values  ----A--B--C--D--E--F--G--H--I--J--K--L--M--N--O--P-->
control --1-4-2-0-3-3--------------------------------------->
result  ---[A]---------[BCDE]-[FG]----[HIJ]----[KLM]-------->