转换 IObservable将不规则长度字节数组转换为 IObservable具有规则长度的数组

发布于 2024-12-08 19:42:33 字数 1023 浏览 0 评论 0原文

我有一个 IObservable，它在字节数组中提供了不确定的字节数。我想知道如何从那里返回一个 IObservable，每个字节数组中都有一定数量的字节。假设我们一次需要 10 个字节。

也就是说，如果我订阅时得到以下输入：

{1, 2, 3, 4}
{5, 6}
{7, 8, 9}
{10}
{11, 12, 13, 14, 15}
{16}
{17, 18}
{19, 20}

Bytes.Subscribe(b => Console.WriteLine(b.Length));

输出将是

我想要的是将上面的输入转换为：

{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
{11, 12, 13, 14, 15, 16, 17, 18, 19, 20}

Bytes.<WhateverItTakesToDoThat>.Subscribe(b => Console.WriteLine(b.Length));

输出将是

10
10

如果有一定数量的字节进来，它也必须工作比单个输出数据包大，即：

{21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32}
{33, 34, 35, 36, 37, 38, 39, 40, 41}
{42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52}

应该变成

{21, 22, 23, 24, 25, 26, 27, 28, 29, 30}
{31, 32, 33, 34, 35, 36, 37, 38, 39, 40}
{41, 42, 43, 44, 45, 46, 47, 48, 49, 50}

（并保留 {51, 52}，等待更多输入出现）

原文

I have an IObservable<byte[]> that gives me an uncertain amount of bytes in the byte array. I want to know how I go from that, to returning an IObservable<byte[]> with a set amount of bytes in each byte array. Let's assume we want 10 bytes at a time.

That is to say, if I get the following input if I were to subscribe:

{1, 2, 3, 4}
{5, 6}
{7, 8, 9}
{10}
{11, 12, 13, 14, 15}
{16}
{17, 18}
{19, 20}

Bytes.Subscribe(b => Console.WriteLine(b.Length));

The output would be

What I would like is to convert the input above into this:

{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
{11, 12, 13, 14, 15, 16, 17, 18, 19, 20}

Bytes.<WhateverItTakesToDoThat>.Subscribe(b => Console.WriteLine(b.Length));

The output would be

10
10

It must also work if an amount of bytes come in that are larger than a single output packet, i.e.:

{21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32}
{33, 34, 35, 36, 37, 38, 39, 40, 41}
{42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52}

Should be turned into

{21, 22, 23, 24, 25, 26, 27, 28, 29, 30}
{31, 32, 33, 34, 35, 36, 37, 38, 39, 40}
{41, 42, 43, 44, 45, 46, 47, 48, 49, 50}

(and be holding on to {51, 52}, waiting for more input to come along)

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

素手挽清风 2024-12-15 19:42:33

这很容易。试试这个：

    Bytes
        .SelectMany(b => b)
        .Buffer(10)
        .Select(bs => bs.ToArray());

It's easy. Try this:

    Bytes
        .SelectMany(b => b)
        .Buffer(10)
        .Select(bs => bs.ToArray());

回复收藏 0 原文

回眸一遍 2024-12-15 19:42:33

经过一番思考和修改后，我想出了一个解决方案。下面的代码实现了我想要的功能：

Bytes.Select( b => b.ToObservable() ) // Convert input to IObservable<IObservable<byte>>
.Merge( 1 ) // Merges the IObservable<IObservable<byte>> to an IObservable<byte>
            // with the bytes in the right order
.Buffer( 4 ) // Wait until we have 4 bytes ready
.Select( bl => bl.ToArray() ) // Take these 4 bytes and turn them back into an array
.Subscribe( b => Console.WriteLine( b.Length ) );

这可能效率很低，而且我几乎可以肯定这不是最有效的方法，所以如果有人能提出更好、更有效的解决方案，我会洗耳恭听！

I came up with one solution after a bit of thinking and tinkering. The following code does what I want:

Bytes.Select( b => b.ToObservable() ) // Convert input to IObservable<IObservable<byte>>
.Merge( 1 ) // Merges the IObservable<IObservable<byte>> to an IObservable<byte>
            // with the bytes in the right order
.Buffer( 4 ) // Wait until we have 4 bytes ready
.Select( bl => bl.ToArray() ) // Take these 4 bytes and turn them back into an array
.Subscribe( b => Console.WriteLine( b.Length ) );

This is probably inefficient, and I'm almost certain it's not he most efficient way of doing this, so if somebody out there can come up with a better, more efficient solution, I'm all ears!

回复收藏 0 原文

~没有更多了~