读取流（nodejs）时“管道失败。响应已经发出”

Question

所以我的代码应该从 CSV 文件中读取一些行，将它们转换为 JSON 对象数组，然后返回该数组。 要将文件作为流读取，我使用got，然后在fast-csv 中使用它。 为了返回结果数组，我将整个东西放入 Promise 中，如下所示：

async GetPage() : Promise<{OutputArray:any[], StartingIndex:number}>{
        return new Promise(async (resolve, reject) => {
            const output:any[] = []; 
            const startingIndex = this.currentLocation;
            try{
                parseStream(this.source, {headers:true, maxRows:this.maxArrayLength, skipRows:this.currentLocation, ignoreEmpty:true, delimiter:this.delimiter})
                    .on('error', error => console.log(`parseStream: ${error}`))
                    .on('data', row => {
                        const obj = this.unflatten(row); // data is flattened JSON, need to unflatten it
                        output.push(obj); // append to output array
                        this.currentLocation++;
                    })
                    .on('end', (rowCount: number) => {
                        console.log(`Parsed ${this.currentLocation} rows`);
                        resolve({OutputArray:output, StartingIndex:startingIndex});
                    });
            }
            catch(ex){
                console.log(`parseStream: ${ex}`);
                throw new Error(ex);
            }
        })
    }

现在，当我调用它一次 (await GetPage()) 时，它工作得非常好。 问题是当我连续第二次调用它时。我得到以下信息：

UnhandledPromiseRejectionWarning: Error: Failed to pipe. The response has been emitted already.

我在这里看到了一个类似的案例：https://github.com/sindresorhus/file-type/issues/342，但据我所知，这是一个不同的案例，或者更确切地说，如果它是相同的，我不知道如何在这里应用解决方案。

GetPage 是类 CSVStreamParser 中的一个方法，它在构造函数中被赋予了一个 Readable，我像这样创建 Readable：readable:Readable = got.stream(url)

让我困惑的是，我的第一个版本的 GetPage 没有包含 Promise，而是接受了一个回调（我只是发送了console.log 来测试它），当我连续调用它几次时没有错误，但它无法返回值，所以我将其转换为 Promise。

谢谢！ :)

编辑：我已经设法通过在GetPage() 的开头重新打开流来使其工作，但我想知道是否有一种方法可以实现相同的结果而不必这样做？有没有办法让流保持打开状态？

Answer 1

首先，删除两个async，因为您已经返回了Promise。

然后删除try/catch 块和throw，因为您不应该提出承诺。而是使用reject 函数。

GetPage() : Promise<{OutputArray:any[], StartingIndex:number}>{
    return new Promise((resolve, reject) => {
        const output:any[] = []; 
        const startingIndex = this.currentLocation;
        parseStream(this.source, {headers:true, maxRows:this.maxArrayLength, skipRows:this.currentLocation, ignoreEmpty:true, delimiter:this.delimiter})
            .on('error', error => reject(error))
            .on('data', row => {
                const obj = this.unflatten(row); // data is flattened JSON, need to unflatten it
                output.push(obj); // append to output array
                this.currentLocation++;
            })
            .on('end', (rowCount: number) => {
                console.log(`Parsed ${this.currentLocation} rows`);
                resolve({OutputArray:output, StartingIndex:startingIndex});
            });
    });
}

这里有一些资源可以帮助您了解async functions 和promises。