如何使用读取回调处理程序中的节点流读取的数据添加其他数据? [英] How to add additional data with data read from a node stream in read callback handler?

问题描述

我正在创建一个Readable流数组(从包含JSON文档的文件中)，并试图将它们通过管道传输到另一个流.

I'm creating an array of Readable streams (from files containing JSON docs) and am trying to pipe them to another stream.

文件中的数据正在通过……但是对于管道中接收到的每个对象，我想知道这些数据来自哪个文件:

The data in the files is coming through… but for every object I receive in the piped to stream, I would like to know from which file this data originated from:

var fs = require('fs');
var path = require('path');
var JSONStream = require('JSONStream');

var tmp1 = path.join(__dirname, 'data', 'tmp1.json');
var tmp2 = path.join(__dirname, 'data', 'tmp2.json');

var jsonStream = JSONStream.parse();
jsonStream.on('data', function (data) {
  console.log('---\nFrom which file does this data come from?');
  console.log(data);
});

[tmp1, tmp2].map(p => {
  return fs.createReadStream(p);
}).forEach(stream => stream.pipe(jsonStream));

输出:

---
From which file does this data come from?
{ a: 1, b: 2 }
---
From which file does this data come from?
{ a: 3, b: 4 }
---
From which file does this data come from?
{ a: 5, b: 6 }
---
From which file does this data come from?
{ a: 100, b: 200 }
---
From which file does this data come from?
{ a: 300, b: 400 }
---
From which file does this data come from?
{ a: 500, b: 600 }

需要文件路径来进一步处理读取的对象(在jsonStream.on('data') callback中)，但是我不知道如何传递这些附加数据.

The file path is needed to further process the object read (in jsonStream.on('data') callback) but I don't know how to go about passing this additional data.

推荐答案

可能的解决方案如下(除非获得更好的答案，否则将其标记为答案):

A possible solution follows (am marking this as the answer unless I get a better answer):

var fs = require('fs');
var path = require('path');
var JSONStream = require('JSONStream');
var through = require('through2');

var tmp1 = path.join(__dirname, 'data', 'tmp1.json');
var tmp2 = path.join(__dirname, 'data', 'tmp2.json');

[tmp1, tmp2]
  .map(p => fs.createReadStream(p))
  .map(stream => [path.parse(stream.path), stream.pipe(JSONStream.parse())])
  .map(([parsedPath, jsonStream]) => {
    return jsonStream.pipe(through.obj(function (obj, _, cb) {
      this.push({
        fileName: parsedPath.name,
        data: obj
      });
      cb();
    }));
  })
  .map(stream => {
    stream.on('data', function (data) {
      console.log(JSON.stringify(data, null, 2));
    });
  })
  ;

输出:

{
  "fileName": "tmp1",
  "data": {
    "a": 1,
    "b": 2
  }
}
{
  "fileName": "tmp1",
  "data": {
    "a": 3,
    "b": 4
  }
}
{
  "fileName": "tmp1",
  "data": {
    "a": 5,
    "b": 6
  }
}
{
  "fileName": "tmp2",
  "data": {
    "a": 100,
    "b": 200
  }
}
{
  "fileName": "tmp2",
  "data": {
    "a": 300,
    "b": 400
  }
}
{
  "fileName": "tmp2",
  "data": {
    "a": 500,
    "b": 600
  }
}

这篇关于如何使用读取回调处理程序中的节点流读取的数据添加其他数据?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！