如何延迟流读取调用

Question

我仍然试图通过streams来摸索。我已经能够从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() // this hangs at 10% 
const pass2 = through(function write(data) { // this hangs from the beginning
this.queue(data)
},
function end() {
this.queue(null)
})
const pass3 = through2() // this hangs at 10%
/* 
// This way works for large files, but I want to defer 
// invocation
const form_data = new FormData()
form_data.append(savepath, part, {
filename,
})
const r = request.post(url, {
headers: {
'transfer-encoding': 'chunked'
}
}, responseCallback(resolve))
r._form = form
*/
form_files.push({
part: part.pipe(pass1),
// part: part.pipe(pass2),
// part: part.pipe(pass3),
})
} else {
part.resume()
}
})
form.on('close', () => {
resolve({
fields: qs.parse(q_str),
forms: form_files,
})
})
form.parse(req)
})
}

附言当然，如果有人可以使用适当的术语，标题可能会更好。谢谢。

Answer 1

我相信这是因为您没有正确使用through2- 即一旦缓冲区已满，实际上并没有清空缓冲区(这就是为什么它在较大文件上以 10% 挂起，但在较小的文件上有效(。

我相信这样的实现应该这样做：

const pass2 = through2(function(chunk, encoding, next) {
// do something with the data

// Use this only if you want to send the data further to another stream reader 
// Note - From your implementation you don't seem to need it
// this.push(data)
// This is what tells through2 it's ready to empty the 
//  buffer and read more data
next();
})