Трансформировать поток, получая доступ к внутренним данным

Question

Я хочу прочитать файл (в идеале с fs.createReadStream, передать его через процесс преобразования, а затем записать его (в идеале, используя fs.createWriteStream) в другой файл.

Я использую поток преобразования (new stream.Transform()) для этого и вроде бы работает, разве что теперь я застрял

в методе потока потока преобразования у меня есть

 strm._flush = function (done) {
        if (this._lastLineData) {
            this.push(this._lastLineData + '\n');
        }
        this._lastLineData = null;
        done();
    };

Я хочу иметь возможность изменять некоторые данные во внутренней структуре данных, где преобразование хранит данные. Как я могу получить доступ к этой внутренней структуре данных.

Другими словами, this.push, безусловно, выдвигает на некоторый массив, я хочу иметь возможность читать из определенных элементов этого массива в методе _flush.

Заходя в ядро ​​узла, я вижу это:

Transform.prototype.push = function(chunk, encoding) {
  this._transformState.needTransform = false;
  return Duplex.prototype.push.call(this, chunk, encoding);
};

а также

Readable.prototype.push = function(chunk, encoding) {
  var state = this._readableState;

  if (!state.objectMode && typeof chunk === 'string') {
    encoding = encoding || state.defaultEncoding;
    if (encoding !== state.encoding) {
      chunk = new Buffer(chunk, encoding);
      encoding = '';
    }
  }

  return readableAddChunk(this, state, chunk, encoding, false);
};

а также

function readableAddChunk(stream, state, chunk, encoding, addToFront) {
  var er = chunkInvalid(state, chunk);
  if (er) {
    stream.emit('error', er);
  } else if (chunk === null) {
    state.reading = false;
    onEofChunk(stream, state);
  } else if (state.objectMode || chunk && chunk.length > 0) {
    if (state.ended && !addToFront) {
      var e = new Error('stream.push() after EOF');
      stream.emit('error', e);
    } else if (state.endEmitted && addToFront) {
      var e = new Error('stream.unshift() after end event');
      stream.emit('error', e);
    } else {
      if (state.decoder && !addToFront && !encoding)
        chunk = state.decoder.write(chunk);

      if (!addToFront)
        state.reading = false;

      // if we want the data now, just emit it.
      if (state.flowing && state.length === 0 && !state.sync) {
        stream.emit('data', chunk);
        stream.read(0);
      } else {
        // update the buffer info.
        state.length += state.objectMode ? 1 : chunk.length;
        if (addToFront)
          state.buffer.unshift(chunk);
        else
          state.buffer.push(chunk);

        if (state.needReadable)
          emitReadable(stream);
      }

      maybeReadMore(stream, state);
    }
  } else if (!addToFront) {
    state.reading = false;
  }

  return needMoreData(state);
}

так что мне кажется нет простого способа добраться до внутренних данных?

Answer 1

Вы не можете сделать это.

Для чего нужно использовать потоки - это передавать данные из одного места в другое. Если вы хотите изменить данные целиком (перед тем, как передать их в следующий поток), вам не следует использовать поток вообще. Думайте о них как о бесконечном потоке данных.

Что вы можете сделать, это:

var data = [];

stream._transform = function(chunk, enc, done) { 
   data.push(chunk);
   done(chunk);
}

stream._flush = function(done) { 
   /** ... Do something with data ... */
   data = []; // Null data variable, to allow GC collecting it.
   done();
};