如何从函数内将数据推送到可读流？

我正在努力实现以下目标：

• 功能getPaths读取目录路径并将它们推入readable找到它们后进行流式传输
• The readable流将传入路径保持管道（流式传输）到write当它接收到路径时进行流式传输。

Code

const fs = require('fs')
const zlib = require('zlib')
const zip = zlib.createGzip()
const Stream = require('stream')


let wstream = fs.createWriteStream('C:/test/file.txt.gz') 
let readable = new Stream.Readable({
  objectMode: true,
  read(item) {
    this.push(item)
  }
})

readable.pipe(zip).pipe(wstream)
.on('finish', (err) => {
  console.log('done');
})

let walkdir = require('walkdir')
function getPaths(dir) {
  let walker = walkdir.sync(dir, {"max_depth": 0, "track_inodes": true}, (path, stat) => {
    readable.push(path)
    console.log('pushing a path to readable')
  }) 
}
getPaths("C:/")
console.log('getPaths() ran')
readable.push(null)  // indicates the end of the stream

Problem

路径没有被压缩并写入文件getPaths函数找到它们并将它们推入流中，直到找到所有它们才会发生。我知道这可能是由于该过程是同步的，但无法弄清楚如何使其工作。

我从日志中看到以下输出：

> // .gz file gets created with size of 0
> // Nothing happens for about 1 minute
> x(184206803) "pushing a path to readable"
> "getPaths() ran"
> // I see the data started being written into the file
> "Done"

UPDATE:

如果我像这样异步执行此操作（或使用下面答案中的代码）：

let walker = walkdir(dir, {"max_depth": 0, "track_inodes": true})
  walker.on('path', (path, stat) => {
    readable.push(path)
  }) 
  walker.on('end', (path, stat) => {
    readable.push(null)
  }) 

  ...

  // readable.push(null) 

我收到一个错误（我认为，当您将数据推入其中后，它没有收到预期的数据块时，它会抛出该特定错误。如果您从代码中删除最后一行：readable.push(null)，并尝试再次运行代码，它会抛出相同的错误）：

TypeError [ERR_INVALID_ARG_TYPE]: The "chunk" argument must be one of type
 string or Buffer. Received type number

你的代码非常好并且工作正常。你只需要删除this.push(item)并设置read具有空体功能。

这是一个工作片段

const fs = require('fs')
const zlib = require('zlib')
const zip = zlib.createGzip()
const Stream = require('stream')


let wstream = fs.createWriteStream('C:/test/file.txt.gz') 
let readable = new Stream.Readable({
  objectMode: true,
  read() { }
})

readable.pipe(zip).pipe(wstream)
.on('finish', (err) => {
  console.log('done');
})

let walkdir = require('walkdir')
function getPaths(dir) {
  let walker = walkdir(dir, {"max_depth": 0, "track_inodes": true})
  walker.on('path', (path, stat) => {
    readable.push(path)
  }) 
  walker.on('end', (path, stat) => {
    readable.push(null)
  }) 
}
getPaths("C:/")
console.log('getPaths() ran')

顺便说一句，正确的参数名称是read(size)。它代表的数量要读取的字节数 https://nodejs.org/dist/latest-v10.x/docs/api/stream.html#stream_readable_read_size_1

EDIT不需要可读流。可以直接写入zip。

const fs = require('fs');
const zlib = require('zlib');
const zip = zlib.createGzip();
const wstream = fs.createWriteStream('C:/test/file.txt.gz');

zip.pipe(wstream)
.on('finish', (err) => {
  console.log('done');
})

let walkdir = require('walkdir')
function getPaths(dir) {
  let walker = walkdir(dir, {"max_depth": 0, "track_inodes": true})
  walker.on('path', (path, stat) => {
    zip.write(path);
  })
  walker.on('end', (path, stat) => {
    zip.end();
  })
}
getPaths("C:/")
console.log('getPaths() ran')