如何拆分和合并 RxJS 发出的数组值？

Question

我有一个 RxJS Observable，它发出类型Uint8Array值的二进制数据。但并非每个发出的值都包含一个完整的数据对象，可以自行处理。

完整数据对象的数据格式由起始字节 ( 0xAA)、中间的一些可变长度数据和结束字节 ( 0xFF) 组成。中间的数据是 BCD 编码的，这意味着它主要不包含开始或结束字节，而只包含从0x00到 的二进制值0x99。

下面是一个例子：

// This is a mock of the source observable which emits values:
const source$ = from([
  // Case 1: One complete data object with start (0xAA) and end byte (0xFF)
  new Uint8Array([0xAA, 0x01, 0x05, 0x95, 0x51, 0xFF,]),

  // Case 2: Two complete data objects in a single value emit
  new Uint8Array([0xAA, 0x12, 0x76, 0xFF, 0xAA, 0x83, 0x43, 0xFF,]),

  // Case 3: Two uncomplete value emits which form a single data object
  new Uint8Array([0xAA, 0x61, 0x85, 0x43, 0x67]),
  new Uint8Array([0x82, 0x73, 0x44, 0x28, 0x85, 0xFF]),

  // Case 4: A combination of Cases 2 and 3
  new Uint8Array([0xAA, 0x61, 0x85, 0x43, 0x67]),
  new Uint8Array([0x55, 0x81, 0xFF, 0xAA, 0x73, 0x96]),
  new Uint8Array([0x72, 0x23, 0x11, 0x95, 0xFF]),
])

source$.subscribe((x) => {
  console.log('Emitted value as Hexdump:')
  console.log(hexdump(x.buffer))
})

目标是只接收完整的数据对象。也许作为一个转化的新 observable？

上面的例子应该是这样的：

const transformedSource$ = from([
  // Case 1
  new Uint8Array([0xAA, 0x01, 0x05, 0x95, 0x51, 0xFF,]),

  // Case 2
  new Uint8Array([0xAA, 0x12, 0x76, 0xFF,]),
  new Uint8Array([0xAA, 0x83, 0x43, 0xFF,]),

  // Case 3
  new Uint8Array([0xAA, 0x61, 0x85, 0x43, 0x67, 0x82, 0x73, 0x44, 0x28, 0x85, 0xFF]),

  // Case 4
  new Uint8Array([0xAA, 0x61, 0x85, 0x43, 0x67, 0x55, 0x81, 0xFF]),
  new Uint8Array([0xAA, 0x73, 0x96, 0x72, 0x23, 0x11, 0x95, 0xFF]),
])

1. 哪些 RxJS 方法或运算符适合于此？
2. 我想先在 at 上进行拆分，0xFF然后再进行合并。这该怎么做？非常感谢有 RxJS 经验的人的想法。

Answer 1

您可以尝试拆分块以一个一个地处理字节，将它们添加到缓冲区，直到0xff字节出现在流中并返回所有缓冲元素并为下一个块重置缓冲区：

let buffer = new Uint8Array();

transformedSource$.pipe(
        mergeAll(), // this splits your array and emits the single bytes into the stream
        mergeMap((next) => {
            buffer = new Uint8Array([...buffer, next]);
            if (next === 0xff) { (
                const result = buffer;
                buffer = new Uint8Array(); // resets the buffer
                return of(result); // will emit a completed chunk
            }
            return EMPTY; // won't emit anything 
        })
    )
    .subscribe(console.log);

如果您不是全局变量的粉丝，或者您想将该代码也用于其他流，这里有一个带有自定义运算符的替代解决方案：

const mergeChunks = () => {
    let buffer = new Uint8Array();

    return (source$) =>
        source$.pipe(
            mergeMap((next) => {
                buffer = new Uint8Array([...buffer, next]);
                if (next === 0xff) {
                    const result = buffer;
                    buffer = new Uint8Array();
                    return of(result);
                }
                return EMPTY;
            })
        );
} 

transformedSource$.pipe(
        mergeAll(),
        mergeChunks(),
    )
    .subscribe(console.log);