如何使用 RxJS 编写简单的元流答案

【问题标题】：How to code simple meta-stream with RxJS如何使用 RxJS 编写简单的元流
【发布时间】：2017-07-19 16:05:26
【问题描述】：

背景

这是我第一次尝试响应式编程。我有一个不时接收数据缓冲区的流。

当缓冲区以02 开始时，表示消息的开始，而当缓冲区以03 结束时，表示消息的结束。

例子：

- - - 02 53 44 5a - - - 52 6a 72 7a - - - 62 32 62 0d - 0a 03 - - - - >

目标

我的目标是每当我检测到一条消息完成时触发一个事件，并将整个消息作为参数。

研究

看了几个tutorials on reactive programming和RxJS之后，我知道我要创建一个hot observable。

通过我的研究，我相信我想创建一个“meta-observable”或“meta-stream”，即一个 observable 的 observables 或一个流的流（这可能会令人困惑......）。

我的计划如下：

1 - 为所有传入的缓冲区消息创建一个 observable 2 - 订阅第 1 步，并创建一个检测消息开头 (02) 和消息结尾 (03) 的 observable 3 - 订阅步骤 2 来处理数据。

所以，我的图表如下所示：

- 05 34 43 31 - - - 12 23 43 - - 02 53 44 5a - - - 52 6a 72 7a - - - 62 32 62 0d - 0a 03 - - - - > 
                             - - 02 53 44 5a - - - 52 6a 72 7a - - - 62 32 62 0d - 0a 03 - - |

问题

1 - 因为我是通过 node.js 中的套接字接收这些数据的：

socket = net.createConnection( { host: "localhost", port: 8080}, () => {
    socket.on( "data", console.log );
} );

我认为我不能使用 rxjs 提供的 EventEmmiter 桥接器。我必须使用Observer.create，对吗？

2 - 即使我解决了问题 1，我也不知道如何为这种效果创建一个流。有人可以发布一个代码 sn-p 作为示例吗？

【问题讨论】：

标签： javascript node.js stream rxjs observable

【解决方案1】：

你在正确的轨道上。但是，您想创建一个 observable，因此您可能需要Observable.create 而不是Observer.create。这很令人困惑，因为Observable.create 需要一个接收Observer 的函数。然后该函数可以将项目发送到该Observer。

你也很想知道你想要一个热门的 observable。然而，Observable.create 无论如何都会给你一个热门的 observable。您仍然希望使用某种发布来避免多次订阅。我们将使用share 做到这一点。至于缓冲，您可能可以使用buffer。在一天结束的时候，我认为它会像......

A - 一个使用 Observable.create 创建的可观察对象，包裹着您的套接字回调。
B - 用 share 包裹 A
C - flatMap B 从缓冲区流到项目流
D - 观看 C 并使用 filter/where 仅在 03 上发射。
E - buffer C 使用 D 作为结束选择器

然后将 E 公开为您服务的公共 API。 E 将在消息到达时发出整个消息。

*注意，这个方法假设你收到消息是背靠背的，因此我们不会打扰02 信号，因为我们知道消息结束后的第一个字节必须是下一条消息的开始.如果不是这种情况，你会想要更好地处理它。

** 您可以在flatMap 之后分享。这将稍微提高性能。缓冲后不能分享。

*** 在重新阅读您的问题时，我注意到您想要一个流而不是数组流。要获得此结果，您可以获取buffer 的输出并使用Observable.of 运行flatMap 这将为您提供冷流的热流。

【讨论】：