'use strict';

var stream = require('stream');

class RangeSource extends stream.Readable {

    constructor(min, max, options) {

        super(options);

        this.min  = min|0;
        this.max  = max|0;
        this.next = this.min;

    }

    _read() {

        for (let data = this.next; data <= this.max; ++data) {
            if (!this.push(data.toString())) {
                this.next = data + 1;
                return;
            }
        }

        this.push(null);

    }

}

class Sink extends stream.Writable {
    _write(data, encoding, next) { next() }
}

class PushDuplex extends stream.Duplex {

    constructor(options) {

        super(options);

        // uncomment for the correct (?) behavior:
        /*

        this.on('finish', () => {
            if (!this.allowHalfOpen) {
                this.push(null);
            }
        });

        //*/

    }

    _read() {}

    _write(data, enc, next) {
        this.push(data);
        next();
    }

}

var source   = new RangeSource(1, 3);
var allow    = new PushDuplex({ allowHalfOpen: true });
var disallow = new PushDuplex({ allowHalfOpen: false });
var sink     = new Sink();

source
    .on('data', (data) => console.log('source      data:', data.toString()))
    .on('end',  ()     => console.log('source      ended'));

source.pipe(allow, { end: true })
    .on('data', (data) => console.log('allow       data:', data.toString()))
    .on('end',  ()     => console.log('allow       ended'))
    .pipe(sink);

source.pipe(disallow, { end: true })
    .on('data', (data) => console.log('disallow    data:', data.toString()))
    .on('end',  ()     => console.log('disallow    ended'))
    .pipe(sink);