收集流中的数据直到 id 改变？答案

【问题标题】：Collect data in stream until id is changed?收集流中的数据直到 id 改变？
【发布时间】：2020-05-09 03:55:52
【问题描述】：

我有流的管道：

pipeline(readStream, transformStream, writeStream);

readStream 传递给 transformStream（在每个“数据”事件上）对象，如下所示：

{
   id: 1,
   name: 'John',
   phone: 1000000,
}

我需要将手机存储在转换流中，直到该对象中的 id 发生更改，然后我应该将这样的对象推送到流缓冲区：

{
   id: 1,
   name: 'John',
   phones: [1000000, 1000001, 1000002],
}

所以数组：

[
  {
   id: 1,
   name: 'John',
   phone: 1000000,
  },
  {
   id: 1,
   name: 'John',
   phone: 1000001,
  },
  {
   id: 2,
   name: 'Ray',
   phone: 1000002,
  },
  {
   id: 3,
   name: 'Santa',
   phone: 1000003,
  },
]

转换后的流将是：

[
  {
   id: 1,
   name: 'John',
   phones: [1000000, 1000001],
  },
  {
   id: 2,
   name: 'Ray',
   phones: [1000002],
  },
  {
   id: 3,
   name: 'Santa',
   phones: [1000003],
  },
]

我该如何实现？

【问题讨论】：

标签： javascript node.js stream

【解决方案1】：

这是一个简单的例子。主要思想是跟踪前一个条目/块并将其与当前条目进行比较，以决定我们是否应该更新phones 属性或将数据推送到下一个流。

const streamify = require('stream-array'); // using this package for testing purposes only
const Stream = require('stream');

const input = [
  {
    id: 1,
    name: 'John',
    phone: 1000000,
  },
  {
    id: 1,
    name: 'John',
    phone: 1000001,
  },
  {
    id: 2,
    name: 'Ray',
    phone: 1000002,
  },
  {
    id: 3,
    name: 'Santa',
    phone: 1000003,
  },
];

function createObjectMergeStream() {
  let previousEntry = null;

  return new Stream.Transform({
    writableObjectMode: true,
    transform: transformFunc,
    flush(callback) {
      callback(null, JSON.stringify(previousEntry)); // stringifying for demonstration only
    }
  });

  function transformFunc(currentEntry, encoding, callback){
    if (previousEntry === null) {
      // if this is the first the stream is receiving
      previousEntry = {
        id: currentEntry.id,
        name: currentEntry.name,
        phones: [currentEntry.phone]
      }

      callback();
    }
    else if (previousEntry.id === currentEntry.id) {
      // if the id's match, only update the phones array
      previousEntry.phones.push(currentEntry.phone);
      callback();
    }
    else {
      // if this entry does not match the id of the previous entry

      // stringifying for demonstration only
      const output = JSON.stringify(previousEntry) + '\n';

      previousEntry = {
        id: currentEntry.id,
        name: currentEntry.name,
        phones: [currentEntry.phone]
      }

      callback(null, output);
    }
  }
}


// turn the `input` array into a readable  stream
const inputStream = streamify(input);

// create our transform stream
const transformStream = createObjectMergeStream();

// take objects from input array, process the objects in our transform stream then print them to the console
inputStream.pipe(transformStream).pipe(process.stdout);

输出

{"id":1,"name":"John","phones":[1000000,1000001]}
{"id":2,"name":"Ray","phones":[1000002]}
{"id":3,"name":"Santa","phones":[1000003]}

【讨论】：