javascript - 如何使用 RxJS 编写简单的元流

Question

背景

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

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

例子:

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

目标

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

研究

看了几个tutorials on reactive programming和阅读 RxJS ，我知道我想创建一个 hot observable .

通过我的研究，我相信我想创建一个“元可观察”或“元流”，即可观察的可观察或流的流(这可能会令人困惑......)。

我的计划如下:

1 - 为所有传入的缓冲区消息创建一个可观察对象2 - 订阅第 1 步，并创建一个检测消息开头 (02) 和消息结尾 (03) 的可观察对象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 );
} );

我不认为我可以使用 EventEmmiter rxjs 提供的桥梁。我必须使用 Observer.create，对吗？

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

Answer 1

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

您也敏锐地认识到您需要一个热可观察对象。然而，Observable.create 无论如何都会给你一个热的 observable。您仍然需要使用某种发布来避免多次订阅。我们将使用 share 来做到这一点。至于缓冲，您可能可以使用 buffer。在一天结束时，我认为它会像...

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

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

*请注意，此方法假定您接连收到消息，因此我们不必理会 02 信号，因为我们知道消息结束后的第一个字节必须是下一条消息的开始。如果不是这种情况，您将希望更好地处理它。

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

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