问题标签 [node-streams]

假设我有一个readable流，例如request(URL). 我想通过请求将它的响应写在磁盘上fs.createWriteStream()。但同时我想通过crypto.createHash()流计算下载数据的校验和。

而且我想即时执行此操作，而无需在内存中缓冲整个响应。

看来我可以使用 oldschool on('data')hook 来实现它。伪代码如下：

但是这样的做法看起来很尴尬。我尝试使用stream.Transform或stream.Writable实现类似的东西，read | calc + echo | write但我坚持执行。

我有 3 个流（A、B、C），它们通过管道传输到另一个（A->B->C）。当我启动我的程序 B 的 _read get 时，立即调用它，因为它通过管道传输到 C。但是 B 流中还没有数据，因为 A 异步获取数据。一旦 B 接收到传递给 B 的 _write 方法的数据，它就会转换数据并发出一个“可读”事件（我手动触发 - 这是应该的方式吗？）。

然而，没有任何反应，任何人都不会使用来自 B 的数据（因此不会调用 B 的 _read）。我可以通过在我的 _write() 方法末尾调用（在 B 上）this._read() 来解决这个问题。但这也可能会将数据推送给消费者，尽管队列已满，对吧？

基本上我想将较大的数据块发送到 B 流中，将其拆分为较小的数据块，然后将它们一个一个地传递给 C。所以我想在B中有某种缓冲区。

我的 Cassandra 数据库中有数百万行我想以 zip 文件的形式流式传输到客户端（不希望内存中可能存在巨大的 zip 文件）。我正在使用 Cassandra-Node 驱动程序中的 stream() 函数，通过管道连接到转换器，该转换器从我关心的每一行中提取一个字段并附加一个换行符，并通过管道将哪些管道归档到 Express Response 对象。这似乎工作正常，但我无法弄清楚如何在流式传输期间正确处理错误。我必须在为客户端流式传输之前设置适当的标头/状态，但是如果流式传输期间出现错误，例如在 dbStream 上，我想清理所有管道并将响应状态重置为类似于 404 .但是如果我在设置标题并开始流式传输后尝试重置状态，我会得到Can't set headers after they are sent. 当管道/流式传输到 Response 对象时，我已经查看了所有内容，但找不到如何正确处理 Node 中的错误。如果我无法在错误时发送正确的响应代码，客户端如何判断是否实际传输了有效数据？任何人都可以帮忙吗？

我使用request 带有crypto. 似乎request实现了旧stream协议，并且在writable.

所以，如果我使用下一个代码：

它产生sha256('')（即从空值）。但是当我使用hasher.setEncoding('hex')代码时会产生sha256(<response_body>)并hasher.writable给出true.

我不明白这样做的原因是什么？文档中在哪里说明了这一点？

我想通过 websocket 将 PostgreSQL 查询的结果流式传输到客户端。

使用 pg-promise 和pg-query-stream从数据库中获取数据。要通过 websocket 流式传输数据，我使用socket.io-stream。

单独来看，所有组件的性能都非常好。虽然当我将 pg-query-stream 传输到 socket.io-stream 时，性能会急剧下降。

我开始了：

我曾尝试使用非对象流：

或者：

查询和传输所有解决方案的数据大约需要一分钟。

查询结果可以在一秒内写入文件：

并且该文件可以在一秒钟内传输：

所以不知何故，这些流似乎并没有很好地结合起来。

作为一个愚蠢的实验，我尝试在两者之间放置一些东西：

突然间数据可以在一秒钟内查询和传输......

不幸的是，这不会作为我的最终解决方案。它会浪费太多的CPU。是什么导致这种流组合的性能下降？如何解决这个问题？

我需要运行一个子进程，然后它将在后台运行；同时，我想操纵它的输出（添加时间戳等）并将其重定向到特定文件。目前我正在这样做：

问题：

• 我是流的新手。我做对了吗？如果不是，那么这样做的“正确”方式是什么？
• 我注意到流的“数据”回调中的错误不会冒泡——甚至不是UnexpectedErrors，因此是try {} catch {}语句。那是正确的方法吗？

我试图找到一种简单、模块化和惯用的方式来解析基于文本的 TCP 流协议。

假设协议如下所示： "[begin][length][blah][blah]...[blah][end][begin]...[end][begin]...[end]"

我想正确使用流（转换？）来构建一个只提取单个消息的小组件（以 [begin] 开头并以 [end] 结尾）。解析到更高级别的数据结构留给其他组件。

我现在也不太关心性能，所以我只想使用一个简单的正则表达式（这个协议可以用正则表达式解析）。

我在几个概念上遇到了麻烦：

1. 由于缓冲区可能没有完整的消息，我如何正确处理状态并保留部分消息，以便在更多数据进入时进行解析？我必须保留自己的缓冲区还是有办法“放回”我没有使用的数据？
2. 由于新数据可能包含多条消息，Transform流可以处理多条消息（比如我是否this.push(data);多次调用）？

（请注意，我正在尝试在套接字连接逻辑之外构建此帧解码器......我想它将是一个扩展stream.Transform和实现该read方法的类）

我正在遵循Sails 文档中给出的简单示例

我遇到了这个错误

知道会发生什么吗？

我正在尝试调试节点应用程序，我正在使用IntelliJ IDEA 14.0.5和nodejs 6.10.2。应用程序配置似乎没问题，实际上应用程序正在运行，但是当我尝试调试应用程序时，我遇到了这个错误：

和其他内部结构：child_process.js、socket_list.js和cluster.js

1. 可能与错误的调试器版本有关吗？

2. 缺少一些配置？

多谢指教，

安德烈亚

如何编写一个拦截消息的简单流？

例如，假设我想记录（或最终转换）用户的 socket.write(...) 调用通过网络发送的消息。

以下是尝试执行此操作的最小程序：

当我执行 socket.pipe(transformer) 时，客户端会打印“OUT：”（就像我想要的那样），但实际上并没有向服务器发送任何内容。当我交换管道位置时，transformer.pipe(socket)，没有任何东西打印到客户端，但消息被发送到服务器。

虽然这里没有列出，但我也尝试使用 Writable 流，它确实在客户端上打印消息，但它永远不会发送到服务器（如果我在 Writable 流中执行 this.push(...)，它似乎仍然没有发送到服务器）

我在这里想念什么？

编辑：为清晰起见重新格式化代码并更新了文本