如何使用RxJS从arrayBuffer拆分数据框？

问题描述：

我使用websocket从硬件接收数据帧。数据帧的定义如下：如何使用RxJS从arrayBuffer拆分数据框？

0xbb(head) ---data--- 0xee(tail)

所接收的数据是存储在Uint8Array，可以存在多个帧：

var buffer = new Uint8Array([0xbb,0,0,0,0xee,0xbb,1,1,1,0xee,0xbb,3,3,3,0xee]);

，我可以在阵列转换为可观察到的：

var obs= Rx.Observable.from(buffer);

RxMarbles：

--0xbb--0--0--0--0xee--0xbb--1--1--1--0xee--0xbb--2--2--2--0xee------ 
------------------000------------------111------------------222------

如何使用RxJS来拆分observable？要使用哪些运营商？ RxJS是这种情况的最佳实践吗？

你的意思是你想将它转换为'Observables'数组？ – CozyAzure

是的，我想要将每个数据帧转换为可观察的数字 – rhyttr

@rhyttr什么定义了每个Observable中的项目数？ – martin

答

const source = Rx.Observable 
 
    .from(['0xbb','0','0','0','0xee','0xbb','1','1','1','0xee','0xbb','3','3','3','0xee']) 
 
    .concatMap(i => Rx.Observable.of(i).delay(1)); 
 

 
source 
 
    .filter(i => i != '0xee' && i != '0xbb') 
 
    .buffer(source.filter(i => i === '0xee')) 
 
    .subscribe(val => console.log(val));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.2/Rx.js"></script>

我们需要将这些值转换通过，否则缓冲区的closingNotifier运行速度更快，然后值的摄取利用concatMap(.. delay(1))以异步和你结束了三个空数组。但是，既然你从websocket收到这些数据包，你已经是异步的了。

此代码不是100％万无一失的，例如当外部设备不发射时会发生什么0xee？我们最终将下一条消息连接到前一条消息。

感谢您的回答，您提醒我字节缓冲区需要转换为时间流，因此RxJS可以运行。但延迟concatMap（.. delay（1））会在快速和大量的通信时发生性能问题吗？ – rhyttr

你也可以使用'observeOn（async）'而不是'delay（1）'。 'async'来自'rxjs/scheduler/async'的import {async};' – martin

答

我想你可以达到你想要的东西与scan()：

const buffer = new Uint8Array([0xbb,0,0,0,0xee,0xbb,1,1,1,0xee,0xbb,3,3,3,0xee]); 
const obs = Observable.from(buffer); 

obs.scan((acc, v) => { 
    if (v === 0xbb) { 
     return [v]; 
    } else { 
     acc.push(v); 
     return acc; 
    } 
    }, []) 
    .filter(acc => acc[acc.length - 1] === 0xee) 
    .subscribe(console.log);

这将打印以下阵列：

[ 187, 0, 0, 0, 238 ] 
[ 187, 1, 1, 1, 238 ] 
[ 187, 3, 3, 3, 238 ]

不过，如果你需要发出与每个数据帧观测量然后按照马克的回答只需使用window()而不是buffer()。

是的，我需要frameable的可观察性，但是谢谢你的回答，这是一种选择方式！ – rhyttr

如何使用RxJS从arrayBuffer拆分数据框？

相关推荐