streams.js

'use strict';
const from = require('from2');

class StdoutNode {
  constructor(id) {
    this.id = id;
  }
  getInputs() {
    return [{
      id: 'stdout',
      stream: process.stdout
    }];
  }
  getOutputs() {
    return [];
  }
}

class StringNode {
  constructor(id, props) {
    this.id = id;
    this.props = props;
    this.str = props.str;
    this.ptr = 0;
    this.stream = from((size, next) => {
      setTimeout(() => {
        next(null, this.str[this.ptr]);
        if (++this.ptr === this.str.length) {
          this.ptr = 0;
        }
      }, 300);
    });
  }
  getInputs() {
    return [];
  }
  getOutputs() {
    return [{
      id: 'output',
      stream: this.stream
    }];
  }
}

const node1 = new StringNode('str', {
  str: 'hello,world!'
});

const node2 = new StdoutNode('std');

node1.getOutputs()[0].stream.pipe(node2.getInputs()[0].stream);