Allows for writing a Node.js object stream to a non-object format and then later parsing it.

object-stream-marshaller.js

'use strict';

const { Transform } = require('stream');

exports.Serialiser = class Serialiser extends Transform {
  constructor(options) {
    super({ writableObjectMode: true, ...options });
  }

  _transform(chunk, encoding, callback) {
    const size = Buffer.allocUnsafe(2);
    size.writeUInt16LE(chunk.length);
    this.push(Buffer.concat([
      size,
      chunk,
    ]));
    callback();
  }
};

exports.Deserialiser = class Deserialiser extends Transform {
  constructor(options) {
    super({ readableObjectMode: true, ...options });
    this.remainder = Buffer.alloc(0);
  }

  _transform(chunk, encoding, callback) {
    this.remainder = Buffer.concat([this.remainder, chunk]);
    this.readChunk();
    callback();
  }

  _flush(cb) {
    this.readChunk(true);
    cb();
  }

  readChunk(force = false) {
    const buffer = this.remainder;
    if (buffer.length === 0) {
      return;
    }
    const size = buffer.readUInt16LE(0);
    const data = buffer.slice(2);
    if (data.length >= size) {
      const canContinue = this.push(data.slice(0, size));
      this.remainder = data.slice(size);
      if (canContinue || force) {
        this.readChunk(force);
      }
    }
  }
};

Example usage with prism-media:

Saving an Opus stream to filesystem:

createReadStream('file.ogg')
  .pipe(new prism.opus.OggDemuxer())
  .pipe(new Serialiser())
  .pipe(createWriteStream('file.nodestream'))

Later re-reading the Opus stream:

createReadStream('file.nodestream')
  .pipe(new Deserialiser())
  // now you have the original Opus packet object stream back

---

Implementation notes

• A very lightweight container format - not fault-tolerant
• Assumes that objects can have a maximum size of 2^16 bytes (~65kb). This is more than suitable for the needs of prism-media Opus streams, but if you're looking to use this elsewhere for larger objects, consider upping this to 2^32 bytes.