标签: node-streams

我目前正在使用名为s3-upload-stream的node.js插件将非常大的文件流式传输到Amazon S3.它使用多部分API,并且在大多数情况下它运行良好.

但是,这个模块显示了它的年龄,我已经不得不对它进行修改(作者也弃用了它).今天我遇到了亚马逊的另一个问题,我真的想接受作者的推荐并开始使用官方的aws-sdk来完成我的上传.

但.

官方SDK似乎不支持管道s3.upload().s3.upload的本质是您必须将可读流作为参数传递给S3构造函数.

我有大约120多个用户代码模块进行各种文件处理,并且它们与输出的最终目的地无关.引擎为它们提供了一个可管理的可写输出流,然后它们就会输出它.我无法将AWS.S3它们交给对象,并要求它们upload()在不向所有模块添加代码的情况下调用它.我使用的原因s3-upload-stream是因为它支持管道.

有没有办法制作aws-sdk s3.upload()我能管道流的东西？

目前，我正在执行将请求req传送到目标 url 并将响应传送回的技巧res，如下所示：

const request = require('request');

const url = 'http://some.url.com' + req.originalUrl;
const destination = request(url);

// pipe req to destination...
const readableA = req.pipe(destination);
readableA.on('end', function () {
    // do optional stuff on end
});

// pipe response to res...
const readableB = readableA.pipe(res);
readableB.on('end', function () {
    // do optional stuff on end
});

由于request现在已正式弃用（boo hoo），使用gaxios 库是否可以使用此技巧？我认为responseType: 'stream'对请求的设置会做与上面类似的事情，但它似乎不起作用。

同样，gaxios 可以在以下上下文中使用：

request
.get('https://some.data.com')
.on('error', function(err) {
    console.log(err);
}) …

包.json

{
  "type": "module",
  "dependencies": {
    "@types/node": "^18.6.5",
    "typescript": "^4.7.4"
  }
}

tsconfig.json

{
  "compilerOptions": {
    "target": "ES2021",
    "module": "ESNext",
    "moduleResolution": "node",
    "forceConsistentCasingInFileNames": true,
    "strict": true,
    "skipLibCheck": true
  }
}

使用节点 v18.7.0，我尝试将从本机 fetch API 返回的网络流转换为节点流。我使用“node:stream”模块中的 Readable.fromWeb 方法。Typescript 返回类型检查错误，但代码使用 @ts-ignore 注释按预期工作。

类型错误

 Argument of type 'ReadableStream<Uint8Array>' is not assignable to parameter of type 'ReadableStream<any>'.
  Type 'ReadableStream<Uint8Array>' is missing the following properties from type 'ReadableStream<any>': values, [Symbol.asyncIterator]

const nodeStream = Readable.fromWeb(fetchRequest.body)

源代码

 Argument of type 'ReadableStream<Uint8Array>' is not assignable to parameter of …

我使用client.upload在pkgcloud上传文件的目录.所有流完成后执行回调的最佳方法是什么？是否有一种内置方式来注册每个流的"完成"事件并在它们全部被触发后执行回调？

var filesToUpload = fs.readdirSync("./local_path"); // will make this async

for(let file of filesToUpload) {
    var writeStream = client.upload({
        container: "mycontainer,
        remote: file
    });
    // seems like i should register finish events with something
    writeStream.on("finish", registerThisWithSomething);
    fs.createReadStream("./local_path/" + file).pipe(writeStream);
}

streams一般来说,我仍然试图改变自己的方式.我已经能够从内部使用multiparty流式传输大型文件form.on('part').但我需要推迟调用并在读取之前解析流.我曾尝试PassThrough,through.through2,但已经得到了不同的结果,它主要挂起,我无法弄清楚该怎么做,也没有步骤调试.我对所有选择都持开放态度.感谢您的所有见解.

import multiparty from 'multiparty'
import {
  PassThrough
} from 'stream';
import through from 'through'
import through2 from 'through2'

export function promisedMultiparty(req) {
  return new Promise((resolve, reject) => {

    const form = new multiparty.Form()
    const form_files = []
    let q_str = ''

    form.on('field', (fieldname, value) => {
      if (value) q_str = appendQStr(fieldname, value, q_str)
    })

    form.on('part', async (part) => {
      if (part.filename) {

        const pass1 = new PassThrough() // …

-期望的行为
-实际行为
-我尝试过的内容
-再现步骤
-研究

期望的行为

从多个api请求接收的多个可读流通过管道传输到单个可写流。

api响应来自ibm-watson的textToSpeech.synthesize（）方法。

需要多个请求的原因是因为该服务5KB对文本输入有限制。

因此18KB，例如一个字符串，需要四个请求才能完成。

实际行为

可写流文件不完整且出现乱码。

该应用程序似乎“挂起”。

当我尝试.mp3在音频播放器中打开不完整的文件时，它说文件已损坏。

打开和关闭文件的过程似乎会增加文件的大小-就像打开文件一样，它会促使更多数据流入其中。

对于较大的输入（例如，四个4000字节或更少的字符串），不良行为更加明显。

我尝试过的

我尝试了几种方法，使用npm包Combined-stream，Combined-stream2，multistream和archiver 将可读流传递给单个可写流或多个可写流，它们都会导致文件不完整。我最后一次尝试不使用任何软件包，并在Steps To Reproduce下面的部分中显示。

因此，我在质疑应用程序逻辑的每个部分：

01. Watson文本对语音API请求的响应类型是什么？

语音文档的文本，表示api响应类型为：

Response type: NodeJS.ReadableStream|FileObject|Buffer

我很困惑，响应类型是三种可能的事情之一。

在所有尝试中，我一直假设它是一个readable stream。

02. 我可以在一个地图函数中发出多个api请求吗？

03. 我可以在中包装每个请求promise()并解决response吗？

04.是否 可以将结果数组分配给promises变量？

05. 我可以声明var audio_files = await Promise.all(promises)吗？

06. 声明之后，所有答复都“完成”了吗？

07. 如何正确地将每个响应传递给可写流？

08. 如何检测所有管道何时完成，以便可以将文件发送回客户端？ …

NodeJS 可写流的“关闭”和“完成”事件有什么区别？

如果我们假设我们有一个写入磁盘的可写流，那么在数据保存到磁盘后，“关闭”和“完成”事件是否都会激活？

有没有办法将可读流连接到Node.js中的可写流,其中writeable尚未准备好接收数据？换句话说,我想将可读与可写连接,但我想在程序的稍后阶段初始化可写,包括定义写方法.也许我们必须实现write方法,但有没有办法以类似的方式暂停可写流,您可以暂停可读流？或者,在将数据传输到可写入之前,我们可以使用中间通过/转换流并缓存数据!

举个例子,通常我们这样做:

readable.pipe(transform).pipe(writable);

但我想做的事情如下:

const tstrm = readable.pipe(transform);

doSomethingAsync().then(function(){

      tstrm.pipe(writable);

});

只是想知道这是否可能以及如何正确地做到这一点,到目前为止都无法搞清楚两者.

我想我希望在连接/管道连接到可写流之前缓冲中间变换流中的数据,然后,一旦连接,就在任何新数据之前首先流式传输缓冲数据.似乎是一个合理的事情,找不到任何关于此的信息.

我正在学习 Node.js 并认为我有一个非常简单的脚本，但是任何调整都不会让我的脚本永远挂起。

假设我有一个愚蠢的服务器正在运行：

$ nc -l 32001 <<EOF
HTTP/1.1 200 OK
Content-Type: text/plain

Works
EOF

我运行以下脚本：

var http = require('http');
var options = {
  hostname: 'localhost',
  port: 32001,
  method: 'POST',
  headers: {'Content-Type': 'text/plain'}
};
var req = http.request(options, res => {
  var exitCode = res.statusCode >= 200 && res.statusCode < 300 ? 0 : 1;
  res.pipe(process.stdout).on('end', () => process.exit(exitCode));
});
req.on('error', error => console.error(error));
process.stdin.pipe(req)
  .on('end', () => console.log('this does trigger'));

但是当我执行以下操作时：

$ echo foobar | node my-script.js

它只是挂起，永远不会到达请求回调。我希望 …

我正在使用 ssh2-sftp-client 从远程服务器获取文件。我正在将文件放入可读流中。我想将此可读流转换为所需的文件（sample.png 作为 png 文件，sample.doc 文件作为 doc 文件等）

这是我的代码-

let Client = require(\'ssh2-sftp-client\');\nlet sftp = new Client();\n sftp.connect({\n    host: sftpCredentials.host,\n    port: sftpCredentials.port,\n    username: sftpCredentials.username,\n    password: sftpCredentials.password\n}).then(res => sftp.list(\'/\'))\n.then(res => {\n     const file = await sftp.get(\'/sample.png\');\n\n})\n

我想将此文件保存在本地。文件是一个可读流，如下 -

ReadStream {_readableState: ReadableState, readable: true, domain: null, _events: Object, _eventsCount: 3, \xe2\x80\xa6}\n    _events:Object {end: , error: , readable: }\n    _eventsCount:3\n    _maxListeners:undefined\n    _readableState:ReadableState {objectMode: false, highWaterMark: 65536, buffer: BufferList, \xe2\x80\xa6}\n    autoClose:true\n    destroyed:false\n    domain:null\n    end:undefined\n    flags:"r"\n    handle:Buffer(4) [0, 0, 0, …